1# Copyright 2021 The gRPC Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14import logging
15from typing import Tuple
16
17from absl import flags
18from absl.testing import absltest
19
20from framework import xds_url_map_testcase
21from framework.helpers import skips
22from framework.test_app import client_app
23
24# Type aliases
25HostRule = xds_url_map_testcase.HostRule
26PathMatcher = xds_url_map_testcase.PathMatcher
27GcpResourceManager = xds_url_map_testcase.GcpResourceManager
28DumpedXdsConfig = xds_url_map_testcase.DumpedXdsConfig
29RpcTypeUnaryCall = xds_url_map_testcase.RpcTypeUnaryCall
30RpcTypeEmptyCall = xds_url_map_testcase.RpcTypeEmptyCall
31XdsTestClient = client_app.XdsTestClient
32_Lang = skips.Lang
33
34logger = logging.getLogger(__name__)
35flags.adopt_module_key_flags(xds_url_map_testcase)
36
37_NUM_RPCS = 50
38
39
40class TestBasicCsds(xds_url_map_testcase.XdsUrlMapTestCase):
41
42    @staticmethod
43    def is_supported(config: skips.TestConfig) -> bool:
44        if config.client_lang == _Lang.NODE:
45            return config.version_gte('v1.5.x')
46        return True
47
48    @staticmethod
49    def url_map_change(
50            host_rule: HostRule,
51            path_matcher: PathMatcher) -> Tuple[HostRule, PathMatcher]:
52        return host_rule, path_matcher
53
54    def xds_config_validate(self, xds_config: DumpedXdsConfig):
55        # Validate Endpoint Configs
56        self.assertNumEndpoints(xds_config, 1)
57        # Validate Node
58        self.assertEqual(self.test_client.ip,
59                         xds_config['node']['metadata']['INSTANCE_IP'])
60        # Validate Listeners
61        self.assertIsNotNone(xds_config.lds)
62        self.assertEqual(self.hostname(), xds_config.lds['name'])
63        # Validate Route Configs
64        self.assertTrue(xds_config.rds['virtualHosts'])
65        # Validate Clusters
66        self.assertEqual(1, len(xds_config.cds))
67        self.assertEqual('EDS', xds_config.cds[0]['type'])
68
69    def rpc_distribution_validate(self, test_client: XdsTestClient):
70        rpc_distribution = self.configure_and_send(
71            test_client,
72            rpc_types=[RpcTypeUnaryCall, RpcTypeEmptyCall],
73            num_rpcs=_NUM_RPCS)
74        self.assertEqual(_NUM_RPCS, rpc_distribution.num_oks)
75
76
77if __name__ == '__main__':
78    absltest.main()
79