How to use invalid_arg method in autotest

Best Python code snippet using autotest_python Github


Full Screen

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# 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 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 '')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...

Full Screen

Full Screen Github


Full Screen

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__ = ''12__status__ = 'Operational'13import pytest14import 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})...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?