[PSM interop] Introduce isort/pylint to the GKE framework (grpc#29476)

* [PSM interop] Introduce isort/pylint to the GKE framework

* Update the rpc_types

* Update variable type annotation style in tests/*.py

bin/run_channelz.py

@@ -178,7 +178,6 @@ def main(argv):
 
     # Resource names.
     resource_prefix: str = xds_flags.RESOURCE_PREFIX.value
-    resource_suffix: str = xds_flags.RESOURCE_SUFFIX.value
 
     # Server
     server_name = xds_flags.SERVER_NAME.value

bin/run_td_setup.py

@@ -67,7 +67,7 @@
 KubernetesServerRunner = server_app.KubernetesServerRunner
 
 
-def main(argv):
+def main(argv):  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
     if len(argv) > 1:
         raise app.UsageError('Too many command-line arguments.')
 

framework/helpers/datetime.py

@@ -14,7 +14,7 @@
 """This contains common helpers for working with dates and time."""
 import datetime
 import re
-from typing import Pattern
+from typing import Optional, Pattern
 
 RE_ZERO_OFFSET: Pattern[str] = re.compile(r'[+\-]00:?00$')
 
@@ -29,7 +29,7 @@ def shorten_utc_zone(utc_datetime_str: str) -> str:
     return RE_ZERO_OFFSET.sub('Z', utc_datetime_str)
 
 
-def iso8601_utc_time(timedelta: datetime.timedelta = None) -> str:
+def iso8601_utc_time(timedelta: Optional[datetime.timedelta] = None) -> str:
     """Return datetime relative to current in ISO-8601 format, UTC tz."""
     time: datetime.datetime = utc_now()
     if timedelta:

framework/helpers/retryers.py

@@ -70,7 +70,7 @@ def exponential_retryer_with_timeout(
 def constant_retryer(*,
                      wait_fixed: timedelta,
                      attempts: int = 0,
-                     timeout: timedelta = None,
+                     timeout: Optional[timedelta] = None,
                      retry_on_exceptions: Optional[Sequence[Any]] = None,
                      logger: Optional[logging.Logger] = None,
                      log_level: Optional[int] = logging.DEBUG) -> Retrying:

framework/infrastructure/gcp/api.py

@@ -149,8 +149,9 @@ def networkservices(self, version):
 
         raise NotImplementedError(f'Network Services {version} not supported')
 
+    @staticmethod
     @functools.lru_cache(None)
-    def secrets(self, version):
+    def secrets(version: str):
         if version == 'v1':
             return secretmanager_v1.SecretManagerServiceClient()
 
@@ -263,7 +264,7 @@ class OperationError(Error):
     """
     api_name: str
     name: str
-    metadata: str
+    metadata: Any
     code_name: code_pb2.Code
     error: status_pb2.Status
 
@@ -431,9 +432,10 @@ def _delete_resource(self, collection: discovery.Resource,
         return False
 
     # TODO(sergiitk): Use ResponseError and TransportError
-    def _execute(self,
-                 request: HttpRequest,
-                 timeout_sec=GcpProjectApiResource._WAIT_FOR_OPERATION_SEC):
+    def _execute(  # pylint: disable=arguments-differ
+            self,
+            request: HttpRequest,
+            timeout_sec=GcpProjectApiResource._WAIT_FOR_OPERATION_SEC):
         operation = request.execute(num_retries=self._GCP_API_RETRIES)
         self._wait(operation, timeout_sec)
 

framework/infrastructure/gcp/compute.py

@@ -26,7 +26,7 @@
 logger = logging.getLogger(__name__)
 
 
-class ComputeV1(gcp.api.GcpProjectApiResource):
+class ComputeV1(gcp.api.GcpProjectApiResource):  # pylint: disable=too-many-public-methods
     # TODO(sergiitk): move someplace better
     _WAIT_FOR_BACKEND_SEC = 60 * 10
     _WAIT_FOR_OPERATION_SEC = 60 * 10
@@ -58,7 +58,7 @@ def create_health_check(self,
                             name: str,
                             protocol: HealthCheckProtocol,
                             *,
-                            port: Optional[int] = None) -> GcpResource:
+                            port: Optional[int] = None) -> 'GcpResource':
         if protocol is self.HealthCheckProtocol.TCP:
             health_check_field = 'tcpHealthCheck'
         elif protocol is self.HealthCheckProtocol.GRPC:
@@ -88,7 +88,7 @@ def delete_health_check(self, name):
 
     def create_firewall_rule(self, name: str, network_url: str,
                              source_ranges: List[str],
-                             ports: List[str]) -> Optional[GcpResource]:
+                             ports: List[str]) -> Optional['GcpResource']:
         try:
             return self._insert_resource(
                 self.api.firewalls(), {
@@ -107,7 +107,7 @@ def create_firewall_rule(self, name: str, network_url: str,
             # TODO(lidiz) use status_code() when we upgrade googleapiclient
             if http_error.resp.status == 409:
                 logger.debug('Firewall rule %s already existed', name)
-                return
+                return None
             else:
                 raise
 
@@ -117,10 +117,10 @@ def delete_firewall_rule(self, name):
     def create_backend_service_traffic_director(
             self,
             name: str,
-            health_check: GcpResource,
-            affinity_header: str = None,
+            health_check: 'GcpResource',
+            affinity_header: Optional[str] = None,
             protocol: Optional[BackendServiceProtocol] = None,
-            subset_size: Optional[int] = None) -> GcpResource:
+            subset_size: Optional[int] = None) -> 'GcpResource':
         if not isinstance(protocol, self.BackendServiceProtocol):
             raise TypeError(f'Unexpected Backend Service protocol: {protocol}')
         body = {
@@ -144,7 +144,7 @@ def create_backend_service_traffic_director(
             }
         return self._insert_resource(self.api.backendServices(), body)
 
-    def get_backend_service_traffic_director(self, name: str) -> GcpResource:
+    def get_backend_service_traffic_director(self, name: str) -> 'GcpResource':
         return self._get_resource(self.api.backendServices(),
                                   backendService=name)
 
@@ -185,9 +185,9 @@ def create_url_map(
         name: str,
         matcher_name: str,
         src_hosts,
-        dst_default_backend_service: GcpResource,
-        dst_host_rule_match_backend_service: Optional[GcpResource] = None,
-    ) -> GcpResource:
+        dst_default_backend_service: 'GcpResource',
+        dst_host_rule_match_backend_service: Optional['GcpResource'] = None,
+    ) -> 'GcpResource':
         if dst_host_rule_match_backend_service is None:
             dst_host_rule_match_backend_service = dst_default_backend_service
         return self._insert_resource(
@@ -206,10 +206,10 @@ def create_url_map(
                 }],
             })
 
-    def create_url_map_with_content(self, url_map_body: Any) -> GcpResource:
+    def create_url_map_with_content(self, url_map_body: Any) -> 'GcpResource':
         return self._insert_resource(self.api.urlMaps(), url_map_body)
 
-    def patch_url_map(self, url_map: GcpResource, body, **kwargs):
+    def patch_url_map(self, url_map: 'GcpResource', body, **kwargs):
         self._patch_resource(collection=self.api.urlMaps(),
                              urlMap=url_map.name,
                              body=body,
@@ -221,9 +221,9 @@ def delete_url_map(self, name):
     def create_target_grpc_proxy(
         self,
         name: str,
-        url_map: GcpResource,
+        url_map: 'GcpResource',
         validate_for_proxyless: bool = True,
-    ) -> GcpResource:
+    ) -> 'GcpResource':
         return self._insert_resource(
             self.api.targetGrpcProxies(), {
                 'name': name,
@@ -238,8 +238,8 @@ def delete_target_grpc_proxy(self, name):
     def create_target_http_proxy(
         self,
         name: str,
-        url_map: GcpResource,
-    ) -> GcpResource:
+        url_map: 'GcpResource',
+    ) -> 'GcpResource':
         return self._insert_resource(self.api.targetHttpProxies(), {
             'name': name,
             'url_map': url_map.url,
@@ -252,10 +252,10 @@ def delete_target_http_proxy(self, name):
     def create_forwarding_rule(self,
                                name: str,
                                src_port: int,
-                               target_proxy: GcpResource,
+                               target_proxy: 'GcpResource',
                                network_url: str,
                                *,
-                               ip_address: str = '0.0.0.0') -> GcpResource:
+                               ip_address: str = '0.0.0.0') -> 'GcpResource':
         return self._insert_resource(
             self.api.globalForwardingRules(),
             {
@@ -367,14 +367,14 @@ def get_backend_service_backend_health(self, backend_service, backend):
             }).execute()
 
     def _get_resource(self, collection: discovery.Resource,
-                      **kwargs) -> GcpResource:
+                      **kwargs) -> 'GcpResource':
         resp = collection.get(project=self.project, **kwargs).execute()
         logger.info('Loaded compute resource:\n%s',
                     self.resource_pretty_format(resp))
         return self.GcpResource(resp['name'], resp['selfLink'])
 
-    def _exists_resource(self, collection: discovery.Resource,
-                         filter: str) -> bool:
+    def _exists_resource(
+            self, collection: discovery.Resource, filter: str) -> bool:  # pylint: disable=redefined-builtin
         resp = collection.list(
             project=self.project, filter=filter,
             maxResults=1).execute(num_retries=self._GCP_API_RETRIES)
@@ -384,7 +384,7 @@ def _exists_resource(self, collection: discovery.Resource,
         return 'items' in resp and resp['items']
 
     def _insert_resource(self, collection: discovery.Resource,
-                         body: Dict[str, Any]) -> GcpResource:
+                         body: Dict[str, Any]) -> 'GcpResource':
         logger.info('Creating compute resource:\n%s',
                     self.resource_pretty_format(body))
         resp = self._execute(collection.insert(project=self.project, body=body))
@@ -420,11 +420,12 @@ def _delete_resource(self, collection: discovery.Resource,
     def _operation_status_done(operation):
         return 'status' in operation and operation['status'] == 'DONE'
 
-    def _execute(self,
-                 request,
-                 *,
-                 test_success_fn=None,
-                 timeout_sec=_WAIT_FOR_OPERATION_SEC):
+    def _execute(  # pylint: disable=arguments-differ
+            self,
+            request,
+            *,
+            test_success_fn=None,
+            timeout_sec=_WAIT_FOR_OPERATION_SEC):
         operation = request.execute(num_retries=self._GCP_API_RETRIES)
         logger.debug('Response %s', operation)
 

framework/infrastructure/gcp/iam.py

@@ -33,7 +33,6 @@ class EtagConflict(gcp.api.Error):
 
     https://cloud.google.com/iam/docs/policies#etag
     """
-    pass
 
 
 def handle_etag_conflict(func):
@@ -54,7 +53,7 @@ def _replace_binding(policy: 'Policy', binding: 'Policy.Binding',
     new_bindings = set(policy.bindings)
     new_bindings.discard(binding)
     new_bindings.add(new_binding)
-    return dataclasses.replace(policy, bindings=frozenset(new_bindings))
+    return dataclasses.replace(policy, bindings=frozenset(new_bindings))  # pylint: disable=too-many-function-args
 
 
 @dataclasses.dataclass(frozen=True)
@@ -190,7 +189,7 @@ class IamV1(gcp.api.GcpProjectApiResource):
     # Otherwise conditions are omitted, and role names returned with a suffix,
     # f.e. roles/iam.workloadIdentityUser_withcond_f1ec33c9beb41857dbf0
     # https://cloud.google.com/iam/docs/reference/rest/v1/Policy#FIELDS.version
-    POLICY_VERSION: str = 3
+    POLICY_VERSION: int = 3
 
     def __init__(self, api_manager: gcp.api.GcpApiManager, project: str):
         super().__init__(api_manager.iam('v1'), project)
@@ -271,8 +270,9 @@ def add_service_account_iam_policy_binding(self, account: str, role: str,
             updated_binding = Policy.Binding(role, frozenset([member]))
         else:
             updated_members: FrozenSet[str] = binding.members.union({member})
-            updated_binding: Policy.Binding = dataclasses.replace(
-                binding, members=updated_members)
+            updated_binding: Policy.Binding = dataclasses.replace(  # pylint: disable=too-many-function-args
+                binding,
+                members=updated_members)
 
         updated_policy: Policy = _replace_binding(policy, binding,
                                                   updated_binding)
@@ -303,8 +303,9 @@ def remove_service_account_iam_policy_binding(self, account: str, role: str,
             return
 
         updated_members: FrozenSet[str] = binding.members.difference({member})
-        updated_binding: Policy.Binding = dataclasses.replace(
-            binding, members=updated_members)
+        updated_binding: Policy.Binding = dataclasses.replace(  # pylint: disable=too-many-function-args
+            binding,
+            members=updated_members)
         updated_policy: Policy = _replace_binding(policy, binding,
                                                   updated_binding)
         self.set_service_account_iam_policy(account, updated_policy)

framework/infrastructure/gcp/network_security.py

@@ -92,6 +92,9 @@ class _NetworkSecurityBase(gcp.api.GcpStandardCloudApiResource,
                            metaclass=abc.ABCMeta):
     """Base class for NetworkSecurity APIs."""
 
+    # TODO(https://github.com/grpc/grpc/issues/29532) remove pylint disable
+    # pylint: disable=abstract-method
+
     def __init__(self, api_manager: gcp.api.GcpApiManager, project: str):
         super().__init__(api_manager.networksecurity(self.api_version), project)
         # Shortcut to projects/*/locations/ endpoints
@@ -101,7 +104,7 @@ def __init__(self, api_manager: gcp.api.GcpApiManager, project: str):
     def api_name(self) -> str:
         return 'networksecurity'
 
-    def _execute(self, *args, **kwargs):  # pylint: disable=signature-differs
+    def _execute(self, *args, **kwargs):  # pylint: disable=signature-differs,arguments-differ
         # Workaround TD bug: throttled operations are reported as internal.
         # Ref b/175345578
         retryer = tenacity.Retrying(