csds_test.py

Go to the documentation of this file.

# Copyright 2021 The gRPC Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
from typing import Tuple
 
from absl import flags
from absl.testing import absltest
 
from framework import xds_url_map_testcase
from framework.helpers import skips
from framework.test_app import client_app
 
# Type aliases
HostRule = xds_url_map_testcase.HostRule
PathMatcher = xds_url_map_testcase.PathMatcher
GcpResourceManager = xds_url_map_testcase.GcpResourceManager
DumpedXdsConfig = xds_url_map_testcase.DumpedXdsConfig
RpcTypeUnaryCall = xds_url_map_testcase.RpcTypeUnaryCall
RpcTypeEmptyCall = xds_url_map_testcase.RpcTypeEmptyCall
XdsTestClient = client_app.XdsTestClient
_Lang = skips.Lang
 
logger = logging.getLogger(__name__)
flags.adopt_module_key_flags(xds_url_map_testcase)
 
_NUM_RPCS = 50
 
 
class TestBasicCsds(xds_url_map_testcase.XdsUrlMapTestCase):
 
    @staticmethod
    def is_supported(config: skips.TestConfig) -> bool:
        if config.client_lang == _Lang.NODE:
            return config.version_gte('v1.5.x')
        return True
 
    @staticmethod
    def url_map_change(
            host_rule: HostRule,
            path_matcher: PathMatcher) -> Tuple[HostRule, PathMatcher]:
        return host_rule, path_matcher
 
    def xds_config_validate(self, xds_config: DumpedXdsConfig):
        # Validate Endpoint Configs
        self.assertNumEndpoints(xds_config, 1)
        # Validate Node
        self.assertEqual(self.test_client.ip,
                         xds_config['node']['metadata']['INSTANCE_IP'])
        # Validate Listeners
        self.assertIsNotNone(xds_config.lds)
        self.assertEqual(self.hostname(), xds_config.lds['name'])
        # Validate Route Configs
        self.assertTrue(xds_config.rds['virtualHosts'])
        # Validate Clusters
        self.assertEqual(1, len(xds_config.cds))
        self.assertEqual('EDS', xds_config.cds[0]['type'])
 
    def rpc_distribution_validate(self, test_client: XdsTestClient):
        rpc_distribution = self.configure_and_send(
            test_client,
            rpc_types=[RpcTypeUnaryCall, RpcTypeEmptyCall],
            num_rpcs=_NUM_RPCS)
        self.assertEqual(_NUM_RPCS, rpc_distribution.num_oks)
 
 
if __name__ == '__main__':
    absltest.main()