KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use invalid_arg method in autotest

Best Python code snippet using autotest_python

backend_services_utils.py

Source:backend_services_utils.py

1# -*- coding: utf-8 -*- #2# Copyright 2014 Google Inc. All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#    http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Code that's shared between multiple backend-services subcommands."""16from __future__ import absolute_import17from __future__ import division18from __future__ import unicode_literals19from googlecloudsdk.calliope import exceptions20from googlecloudsdk.core import exceptions as core_exceptions21from googlecloudsdk.core import log22from googlecloudsdk.core import properties23class CacheKeyQueryStringException(core_exceptions.Error):24  def __init__(self):25    super(CacheKeyQueryStringException, self).__init__(26        'cache-key-query-string-whitelist and '27        'cache-key-query-string-blacklist may only be set when '28        'cache-key-include-query-string is enabled.')29# TODO(b/35086027) - Remove this30def IsDefaultRegionalBackendServicePropertyNoneWarnOtherwise():31  """Warns if core/default_regional_backend_service property is set."""32  default_regional = (33      properties.VALUES.core.default_regional_backend_service.GetBool())34  if default_regional is not None:35    # Print a warning if it is set.36    log.warning(37        'core/default_regional_backend_service property is deprecated and '38        'has no meaning.')39# TODO(b/35086027) - Remove this40def GetDefaultScope():41  """Gets the default compute flags scope enum value."""42  IsDefaultRegionalBackendServicePropertyNoneWarnOtherwise()43  return None44def GetIAP(iap_arg, messages, existing_iap_settings=None):45  """Returns IAP settings from arguments."""46  # --iap is specified as str in flags.py.  We do that and then re-parse47  # here instead of just setting the flag to ArgDict in the first place48  # to fix the autogenerated help text.  TODO(b/34479878): Clean this up.49  subargs = iap_arg.split(',')50  iap_arg_parsed = {}51  for subarg in subargs:52    if not subarg:53      continue54    if '=' in subarg:55      subarg, value = subarg.split('=', 1)56    else:57      value = True58    def _Repr(s):59      r = repr(s)60      if r.startswith('u'):61        r = r[1:]62      return r63    if subarg in ('enabled', 'disabled', 'oauth2-client-id',64                  'oauth2-client-secret'):65      if subarg in iap_arg_parsed:66        raise exceptions.InvalidArgumentException(67            '--iap', 'Sub-argument %s specified multiple times' %68            _Repr(subarg))69      iap_arg_parsed[subarg] = value70    else:71      raise exceptions.InvalidArgumentException(72          '--iap', 'Invalid sub-argument %s' % _Repr(subarg))73  if not iap_arg_parsed or not iap_arg:74    raise exceptions.InvalidArgumentException(75        '--iap', 'Must provide value when specifying --iap')76  if 'enabled' in iap_arg_parsed and 'disabled' in iap_arg_parsed:77    raise exceptions.InvalidArgumentException(78        '--iap', 'Must specify only one of [enabled] or [disabled]')79  iap_settings = messages.BackendServiceIAP()80  if 'enabled' in iap_arg_parsed:81    iap_settings.enabled = True82  elif 'disabled' in iap_arg_parsed:83    iap_settings.enabled = False84  elif existing_iap_settings is None:85    iap_settings.enabled = False86  else:87    iap_settings.enabled = existing_iap_settings.enabled88  if ('oauth2-client-id' in iap_arg_parsed or89      'oauth2-client-secret' in iap_arg_parsed):90    iap_settings.oauth2ClientId = iap_arg_parsed.get('oauth2-client-id')91    iap_settings.oauth2ClientSecret = iap_arg_parsed.get('oauth2-client-secret')92    # If either oauth2-client-id or oauth2-client-secret is specified,93    # then the other should also be specified.94    if not iap_settings.oauth2ClientId or not iap_settings.oauth2ClientSecret:95      raise exceptions.InvalidArgumentException(96          '--iap',97          'Both [oauth2-client-id] and [oauth2-client-secret] must be '98          'specified together')99  return iap_settings100def IapBestPracticesNotice():101  return ('IAP only protects requests that go through the Cloud Load Balancer. '102          'See the IAP documentation for important security best practices: '103          'https://cloud.google.com/iap/')104def IapHttpWarning():105  return ('IAP has been enabled for a backend service that does not use HTTPS. '106          'Data sent from the Load Balancer to your VM will not be encrypted.')107def _ValidateGroupMatchesArgs(args):108  """Validate if the group arg is used with the correct group specific flags."""109  invalid_arg = None110  if args.instance_group:111    if args.max_rate_per_endpoint is not None:112      invalid_arg = '--max-rate-per-endpoint'113    elif args.max_connections_per_endpoint is not None:114      invalid_arg = '--max-connections-per-endpoint'115    if invalid_arg is not None:116      raise exceptions.InvalidArgumentException(117          invalid_arg,118          'cannot be set with --instance-group')119  elif args.network_endpoint_group:120    if args.max_rate_per_instance is not None:121      invalid_arg = '--max-rate-per-instance'122    elif args.max_connections_per_instance is not None:123      invalid_arg = '--max-connections-per-instance'124    if invalid_arg is not None:125      raise exceptions.InvalidArgumentException(126          invalid_arg,127          'cannot be set with --network-endpoint-group')128def ValidateBalancingModeArgs(messages, add_or_update_backend_args,129                              current_balancing_mode=None,130                              supports_neg=False):131  """Check whether the setup of the backend LB related fields is valid.132  Args:133    messages: API messages class, determined by release track.134    add_or_update_backend_args: argparse Namespace. The arguments135      provided to add-backend or update-backend commands.136    current_balancing_mode: BalancingModeValueValuesEnum. The balancing mode137      of the existing backend, in case of update-backend command. Must be138      None otherwise.139    supports_neg: bool, if the args contains network endpoint group related140      args.141  """142  balancing_mode_enum = messages.Backend.BalancingModeValueValuesEnum143  balancing_mode = current_balancing_mode144  if add_or_update_backend_args.balancing_mode:145    balancing_mode = balancing_mode_enum(146        add_or_update_backend_args.balancing_mode)147  if supports_neg:  # Validate flags are specified with the correct group.148    _ValidateGroupMatchesArgs(add_or_update_backend_args)149  invalid_arg = None150  if balancing_mode == balancing_mode_enum.RATE:151    if add_or_update_backend_args.max_utilization is not None:152      invalid_arg = '--max-utilization'153    elif add_or_update_backend_args.max_connections is not None:154      invalid_arg = '--max-connections'155    elif add_or_update_backend_args.max_connections_per_instance is not None:156      invalid_arg = '--max-connections-per-instance'157    elif (supports_neg and158          add_or_update_backend_args.max_connections_per_endpoint is not None):159      invalid_arg = '--max-connections-per-endpoint'160    if invalid_arg is not None:161      raise exceptions.InvalidArgumentException(162          invalid_arg,163          'cannot be set with RATE balancing mode')164  elif balancing_mode == balancing_mode_enum.CONNECTION:165    if add_or_update_backend_args.max_utilization is not None:166      invalid_arg = '--max-utilization'167    elif add_or_update_backend_args.max_rate is not None:168      invalid_arg = '--max-rate'169    elif add_or_update_backend_args.max_rate_per_instance is not None:170      invalid_arg = '--max-rate-per-instance'171    elif (supports_neg and172          add_or_update_backend_args.max_rate_per_endpoint is not None):173      invalid_arg = '--max-rate-per-endpoint'174    if invalid_arg is not None:175      raise exceptions.InvalidArgumentException(176          invalid_arg,177          'cannot be set with CONNECTION balancing mode')178  elif balancing_mode == balancing_mode_enum.UTILIZATION:179    if (supports_neg and180        add_or_update_backend_args.network_endpoint_group is not None):181      raise exceptions.InvalidArgumentException(182          '--network-endpoint-group',183          'cannot be set with UTILIZATION balancing mode')184def UpdateCacheKeyPolicy(args, cache_key_policy):185  """Sets the cache_key_policy according to the command line arguments.186  Args:187    args: Arguments specified through command line.188    cache_key_policy: new CacheKeyPolicy to be set (or preexisting one if189      using update).190  """191  if args.cache_key_include_protocol is not None:192    cache_key_policy.includeProtocol = args.cache_key_include_protocol193  if args.cache_key_include_host is not None:194    cache_key_policy.includeHost = args.cache_key_include_host195  if args.cache_key_include_query_string is not None:196    cache_key_policy.includeQueryString = args.cache_key_include_query_string197    if not args.cache_key_include_query_string:198      cache_key_policy.queryStringWhitelist = []199      cache_key_policy.queryStringBlacklist = []200  if args.cache_key_query_string_whitelist is not None:201    (cache_key_policy.queryStringWhitelist202    ) = args.cache_key_query_string_whitelist203    cache_key_policy.includeQueryString = True204    cache_key_policy.queryStringBlacklist = []205  if args.cache_key_query_string_blacklist is not None:206    (cache_key_policy.queryStringBlacklist207    ) = args.cache_key_query_string_blacklist208    cache_key_policy.includeQueryString = True209    cache_key_policy.queryStringWhitelist = []210def ValidateCacheKeyPolicyArgs(cache_key_policy_args):211  # If includeQueryString is not set, it should default to True212  include_query_string = (213      cache_key_policy_args.cache_key_include_query_string is None or214      cache_key_policy_args.cache_key_include_query_string)215  if not include_query_string:216    if (cache_key_policy_args.cache_key_query_string_whitelist is not None or217        cache_key_policy_args.cache_key_query_string_blacklist is not None):218      raise CacheKeyQueryStringException()219def HasCacheKeyPolicyArgsForCreate(args):220  """Returns true if create request requires a CacheKeyPolicy message.221  Args:222    args: The arguments passed to the gcloud command.223  Returns:224    True if there are cache key policy related arguments which require adding225    a CacheKeyPolicy message in the create request.226  """227  # When doing create cache_key_include_host, cache_key_include_protocol,228  # and cache_key_include_query_string have defaults in the API set to True.229  # So only if the user specifies False for any of these or if the user has230  # specified cache_key_query_string_whitelist,231  # cache_key_query_string_blacklist we need to add a CacheKeyPolicy message232  # in the request.233  return (not args.cache_key_include_host or234          not args.cache_key_include_protocol or235          not args.cache_key_include_query_string or236          args.IsSpecified('cache_key_query_string_whitelist') or237          args.IsSpecified('cache_key_query_string_blacklist'))238def HasCacheKeyPolicyArgsForUpdate(args):239  """Returns true if update request requires a CacheKeyPolicy message.240  Args:241    args: The arguments passed to the gcloud command.242  Returns:243    True if there are cache key policy related arguments which require adding244    a CacheKeyPolicy message in the update request.245  """246  # When doing update, if any of the cache key related fields have been247  # specified by the user in the command line, we need to add a248  # CacheKeyPolicy message in the request.249  return (args.IsSpecified('cache_key_include_protocol') or250          args.IsSpecified('cache_key_include_host') or251          args.IsSpecified('cache_key_include_query_string') or252          args.IsSpecified('cache_key_query_string_whitelist') or253          args.IsSpecified('cache_key_query_string_blacklist'))254def GetCacheKeyPolicy(client, args, backend_service):255  """Validates and returns the cache key policy.256  Args:257    client: The client used by gcloud.258    args: The arguments passed to the gcloud command.259    backend_service: The backend service object. If the backend service object260    contains a cache key policy already, it is used as the base to apply261    changes based on args.262  Returns:263    The cache key policy.264  """265  cache_key_policy = client.messages.CacheKeyPolicy()266  if (backend_service.cdnPolicy is not None and267      backend_service.cdnPolicy.cacheKeyPolicy is not None):268    cache_key_policy = backend_service.cdnPolicy.cacheKeyPolicy269  ValidateCacheKeyPolicyArgs(args)270  UpdateCacheKeyPolicy(args, cache_key_policy)271  return cache_key_policy272def ApplyCdnPolicyArgs(client,273                       args,274                       backend_service,275                       is_update=False,276                       apply_signed_url_cache_max_age=False):277  """Applies the CdnPolicy arguments to the specified backend service.278  If there are no arguments related to CdnPolicy, the backend service remains279  unmodified.280  Args:281    client: The client used by gcloud.282    args: The arguments passed to the gcloud command.283    backend_service: The backend service object.284    is_update: True if this is called on behalf of an update command instead285    of a create command, False otherwise.286    apply_signed_url_cache_max_age: If True, also adds the287    signedUrlCacheMaxAgeSec parameter to the CdnPolicy if present in the input288    arguments.289  """290  cdn_policy_args = {}291  add_cache_key_policy = (HasCacheKeyPolicyArgsForUpdate(args) if is_update else292                          HasCacheKeyPolicyArgsForCreate(args))293  if add_cache_key_policy:294    cdn_policy_args['cacheKeyPolicy'] = GetCacheKeyPolicy(295        client, args, backend_service)296  if apply_signed_url_cache_max_age and args.IsSpecified(297      'signed_url_cache_max_age'):298    cdn_policy_args['signedUrlCacheMaxAgeSec'] = args.signed_url_cache_max_age299  if cdn_policy_args:300    backend_service.cdnPolicy = client.messages.BackendServiceCdnPolicy(301        **cdn_policy_args)302def ApplyFailoverPolicyArgs(messages, args, backend_service):303  """Applies the FailoverPolicy arguments to the specified backend service.304  If there are no arguments related to FailoverPolicy, the backend service305  remains unmodified.306  Args:307    messages: The avalible API proto messages.308    args: The arguments passed to the gcloud command.309    backend_service: The backend service proto message object.310  """311  if ((args.connection_drain_on_failover is not None or312       args.drop_traffic_if_unhealthy is not None) and313      backend_service.loadBalancingScheme ==314      messages.BackendService.LoadBalancingSchemeValueValuesEnum.EXTERNAL):315    raise exceptions.InvalidArgumentException(316        '--load-balancing-scheme',317        'can only specify --connection-drain-on-failover or '318        '--drop-traffic-if-unhealthy if the load balancing scheme is INTERNAL.')319  if (args.connection_drain_on_failover is not None and320      backend_service.protocol !=321      messages.BackendService.ProtocolValueValuesEnum.TCP):322    raise exceptions.InvalidArgumentException(323        '--protocol', 'can only specify --connection-drain-on-failover '324        'if the protocol is TCP.')325  if (args.connection_drain_on_failover is not None or326      args.drop_traffic_if_unhealthy is not None or args.failover_ratio):327    failover_policy = (backend_service.failoverPolicy328                       if backend_service.failoverPolicy else329                       messages.BackendServiceFailoverPolicy())330    if args.connection_drain_on_failover is not None:331      failover_policy.disableConnectionDrainOnFailover = (332          not args.connection_drain_on_failover)333    if args.drop_traffic_if_unhealthy is not None:334      failover_policy.dropTrafficIfUnhealthy = args.drop_traffic_if_unhealthy335    if args.failover_ratio:336      failover_policy.failoverRatio = args.failover_ratio...

test_db_manager.py

Source:test_db_manager.py

1# -*- coding: utf-8 -*-2#3# test_db_manager.py4#5# The module is part of pydavis.6#7"""8Testing of DatabaseManager.9"""10__author__ = 'Severin E. R. Langberg'11__email__ = 'Langberg91@gmail.no'12__status__ = 'Operational'13import pytest14import nose.tools as ntools15import mocks16from datetime import datetime17from pytest_mock import mocker18from pydavis import utils19from pydavis import DatabaseManager20class TestDatabaseManager:21    """Testing the DatabaseManager."""22    testing_class = DatabaseManager23    @pytest.fixture24    def manager(self):25        """Returns a database manager instance for test purposes."""26        return self.testing_class('test', 'test')27    def test_update_limit_varchars(self, manager):28        """Test changing the character limit of carchar variables."""29        test_limit = 3030        with pytest.raises(AssertionError):31            ntools.eq_(manager.limit_varchars, test_limit)32        manager.update_limit_varchars(test_limit)33        ntools.eq_(manager.limit_varchars, test_limit)34    def test_init_status(self, manager):35        """Tests initial status of attributes."""36        attributes = [manager.connection, manager.current_db, manager.query]37        assert all(attribute is None for attribute in attributes)38    def test_connection(self, manager, mocker):39        """Test no error is raised if existing connetion."""40        # Mocking a connection.41        mocker.patch('pydavis.db_manager.DatabaseManager.connection',42                     'test_connection')43        manager.connect()44    def test_no_connection(self, manager):45        """Tests if error is raised should if no connection exists."""46        with pytest.raises(utils.DatabaseConnectionError):47            manager.connect()48        with pytest.raises(utils.DatabaseConnectionError):49            manager.execute()50    def test_no_database(self, manager):51        """Tests if error is raised should no working database be set."""52        with pytest.raises(utils.MissingDatabaseError):53            manager.create_table('test', 'test')54        with pytest.raises(utils.MissingDatabaseError):55            manager.drop_table('test')56        with pytest.raises(utils.MissingDatabaseError):57            manager.describe_table('test')58        with pytest.raises(utils.MissingDatabaseError):59            manager.insert_values('test', 'test')60    def test_invalid_args(self, manager, mocker):61        """Tests if error is raised as invalid arguments are passed."""62        # Mocking a working database63        mocker.patch('pydavis.db_manager.DatabaseManager.current_db', 'test')64        invalid_args = [1, 1.0, {}, [], (), None, False, True]65        for invalid_arg in invalid_args:66            with pytest.raises(TypeError):67                manager.create_database(invalid_arg)68                manager.use_database(invalid_arg)69                manager.drop_database(invalid_arg)70                manager.create_table(invalid_arg, invalid_arg)71                manager.insert_values(invalid_arg, invalid_arg)72                manager.create_table('', invalid_arg)73                manager.insert_values('', invalid_arg)74                manager.drop_table(invalid_arg)75                manager.describe_table(invalid_arg)76    def test_create_database(self, manager):77        """Test database creation query."""78        # Mocking query.79        target_query = mocks.MockCreateDB('test')80        manager.create_database('test')81        ntools.eq_(manager.query, target_query())82    def test_use_database(self, manager):83        """Test use database query."""84        # Mocking query.85        target_query = mocks.MockUseDB('test')86        manager.use_database('test')87        ntools.eq_(manager.query, target_query())88    def test_drop_database(self, manager):89        """Test drop database query."""90        # Mocking query.91        target_query = mocks.MockDropDB('test')92        manager.drop_database('test')93        ntools.eq_(manager.query, target_query())94    def test_create_table(self, manager, mocker):95        """Test table creation query."""96        # Mocking a working database97        mocker.patch('pydavis.db_manager.DatabaseManager.current_db', 'test')98        # Mocking query.99        target_query = mocks.MockCreateTable(100            'test', ['ColA', 'ColB', 'ColC'],101            ['INT', 'FLOAT', 'VARCHAR({})'.format(manager.limit_varchars)]102        )103        manager.create_table('test', {'ColA': int, 'ColB': float, 'ColC': str})104        ntools.eq_(manager.query.strip(), target_query().strip())105    def test_convert_dtypes(self, manager):106        """Test conversion of Python data types to MySQL data types."""107        target_dtypes = [108            'INT', 'FLOAT', 'DATETIME',109            'VARCHAR({})'.format(manager.limit_varchars)110        ]111        valid_dtypes = [int, float, datetime, str]112        transformed_dtypes = manager.convert_dtypes(valid_dtypes)113        for target, transformed in zip(target_dtypes, transformed_dtypes):114            ntools.eq_(target, transformed)115        with pytest.raises(TypeError):116            manager.convert_dtypes[()]117            manager.convert_dtypes[[]]118    def test_drop_table(self, manager, mocker):119        """Test drop table query."""120        # Mocking a working database121        mocker.patch('pydavis.db_manager.DatabaseManager.current_db', 'test')122        # Mocking query.123        target_query = mocks.MockDropTable('test')124        manager.drop_table('test')125        ntools.eq_(manager.query, target_query())126    def test_describe_table(self, manager, mocker):127        """Test describe table query."""128        # Mocking a working database129        mocker.patch('pydavis.db_manager.DatabaseManager.current_db', 'test')130        # Mocking query.131        target_query = mocks.MockDescribeTable('test')132        manager.describe_table('test')133        ntools.eq_(manager.query, target_query())134    def test_insert_values(self, manager, mocker):135        """Test insertin into table query."""136        # Mocking a working database137        mocker.patch('pydavis.db_manager.DatabaseManager.current_db', 'test')138        # Mocking query.139        target_query = mocks.MockInsertValues(140            'test', ['ColA', 'ColB', 'ColC'], ['1', '2', '3']141        )142        manager.insert_values('test', {'ColA': 1, 'ColB': 2, 'ColC': 3})...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.