KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _helper method in localstack
Best Python code snippet using localstack_python
test_ganesha.py
Source:test_ganesha.py 
...137 mock.Mock(side_effect=RuntimeError('fake error')))138 self.assertRaises(RuntimeError, self._helper._load_conf_dir,139 self.fake_conf_dir_path)140 ganesha.os.listdir.assert_called_once_with(self.fake_conf_dir_path)141 def test_init_helper(self):142 mock_template = mock.Mock()143 mock_ganesha_manager = mock.Mock()144 self.mock_object(ganesha.ganesha_manager, 'GaneshaManager',145 mock.Mock(return_value=mock_ganesha_manager))146 self.mock_object(self._helper, '_load_conf_dir',147 mock.Mock(return_value=mock_template))148 self.mock_object(self._helper, '_default_config_hook')149 ret = self._helper.init_helper()150 ganesha.ganesha_manager.GaneshaManager.assert_called_once_with(151 self._execute, 'faketag',152 ganesha_config_path='/fakedir0/fakeconfig',153 ganesha_export_dir='/fakedir0/export.d',154 ganesha_db_path='/fakedir1/fake.db',155 ganesha_service_name='ganesha.fakeservice')156 self._helper._load_conf_dir.assert_called_once_with(157 '/fakedir2/faketempl.d', must_exist=False)158 self.assertFalse(self._helper._default_config_hook.called)159 self.assertEqual(mock_ganesha_manager, self._helper.ganesha)160 self.assertEqual(mock_template, self._helper.export_template)161 self.assertIsNone(ret)162 def test_init_helper_conf_dir_empty(self):163 mock_template = mock.Mock()164 mock_ganesha_manager = mock.Mock()165 self.mock_object(ganesha.ganesha_manager, 'GaneshaManager',166 mock.Mock(return_value=mock_ganesha_manager))167 self.mock_object(self._helper, '_load_conf_dir',168 mock.Mock(return_value={}))169 self.mock_object(self._helper, '_default_config_hook',170 mock.Mock(return_value=mock_template))171 ret = self._helper.init_helper()172 ganesha.ganesha_manager.GaneshaManager.assert_called_once_with(173 self._execute, 'faketag',174 ganesha_config_path='/fakedir0/fakeconfig',175 ganesha_export_dir='/fakedir0/export.d',176 ganesha_db_path='/fakedir1/fake.db',177 ganesha_service_name='ganesha.fakeservice')178 self._helper._load_conf_dir.assert_called_once_with(179 '/fakedir2/faketempl.d', must_exist=False)180 self._helper._default_config_hook.assert_called_once_with()181 self.assertEqual(mock_ganesha_manager, self._helper.ganesha)182 self.assertEqual(mock_template, self._helper.export_template)183 self.assertIsNone(ret)184 def test_default_config_hook(self):185 fake_template = {'key': 'value'}186 self.mock_object(ganesha.ganesha_utils, 'path_from',187 mock.Mock(return_value='/fakedir3/fakeconfdir'))188 self.mock_object(self._helper, '_load_conf_dir',189 mock.Mock(return_value=fake_template))190 ret = self._helper._default_config_hook()191 ganesha.ganesha_utils.path_from.assert_called_once_with(192 ganesha.__file__, 'conf')193 self._helper._load_conf_dir.assert_called_once_with(194 '/fakedir3/fakeconfdir')195 self.assertEqual(fake_template, ret)196 def test_fsal_hook(self):197 ret = self._helper._fsal_hook('/fakepath', self.share, self.access)198 self.assertEqual({}, ret)199 def test_cleanup_fsal_hook(self):200 ret = self._helper._cleanup_fsal_hook('/fakepath', self.share,201 self.access)202 self.assertIsNone(ret)203 def test_allow_access(self):204 mock_ganesha_utils_patch = mock.Mock()205 def fake_patch_run(tmpl1, tmpl2, tmpl3):206 mock_ganesha_utils_patch(copy.deepcopy(tmpl1), tmpl2, tmpl3)207 tmpl1.update(tmpl3)208 self.mock_object(self._helper.ganesha, 'get_export_id',209 mock.Mock(return_value=101))210 self.mock_object(self._helper, '_fsal_hook',211 mock.Mock(return_value='fakefsal'))212 self.mock_object(ganesha.ganesha_utils, 'patch',213 mock.Mock(side_effect=fake_patch_run))214 ret = self._helper._allow_access(fake_basepath, self.share,215 self.access)216 self._helper.ganesha.get_export_id.assert_called_once_with()217 self._helper._fsal_hook.assert_called_once_with(218 fake_basepath, self.share, self.access)219 mock_ganesha_utils_patch.assert_called_once_with(220 {}, self._helper.export_template, fake_output_template)221 self._helper._fsal_hook.assert_called_once_with(222 fake_basepath, self.share, self.access)223 self._helper.ganesha.add_export.assert_called_once_with(224 fake_export_name, fake_output_template)225 self.assertIsNone(ret)226 def test_allow_access_error_invalid_share(self):227 access = fake_share.fake_access(access_type='notip')228 self.assertRaises(exception.InvalidShareAccess,229 self._helper._allow_access, '/fakepath',230 self.share, access)231 def test_deny_access(self):232 ret = self._helper._deny_access('/fakepath', self.share, self.access)233 self._helper.ganesha.remove_export.assert_called_once_with(234 'fakename--fakeaccid')235 self.assertIsNone(ret)236 def test_update_access_for_allow(self):237 self.mock_object(self._helper, '_allow_access')238 self.mock_object(self._helper, '_deny_access')239 self._helper.update_access(240 self._context, self.share, access_rules=[self.access],241 add_rules=[self.access], delete_rules=[])242 self._helper._allow_access.assert_called_once_with(243 '/', self.share, self.access)244 self.assertFalse(self._helper._deny_access.called)245 self.assertFalse(self._helper.ganesha.reset_exports.called)246 self.assertFalse(self._helper.ganesha.restart_service.called)247 def test_update_access_for_deny(self):248 self.mock_object(self._helper, '_allow_access')249 self.mock_object(self._helper, '_deny_access')250 self._helper.update_access(251 self._context, self.share, access_rules=[],252 add_rules=[], delete_rules=[self.access])253 self._helper._deny_access.assert_called_once_with(254 '/', self.share, self.access)255 self.assertFalse(self._helper._allow_access.called)256 self.assertFalse(self._helper.ganesha.reset_exports.called)257 self.assertFalse(self._helper.ganesha.restart_service.called)258 def test_update_access_recovery(self):259 self.mock_object(self._helper, '_allow_access')260 self.mock_object(self._helper, '_deny_access')261 self._helper.update_access(262 self._context, self.share, access_rules=[self.access],263 add_rules=[], delete_rules=[])264 self._helper._allow_access.assert_called_once_with(265 '/', self.share, self.access)266 self.assertFalse(self._helper._deny_access.called)267 self.assertTrue(self._helper.ganesha.reset_exports.called)268 self.assertTrue(self._helper.ganesha.restart_service.called)269@ddt.ddt270class GaneshaNASHelper2TestCase(test.TestCase):271 """Tests GaneshaNASHelper2."""272 def setUp(self):273 super(GaneshaNASHelper2TestCase, self).setUp()274 CONF.set_default('ganesha_config_path', '/fakedir0/fakeconfig')275 CONF.set_default('ganesha_db_path', '/fakedir1/fake.db')276 CONF.set_default('ganesha_export_dir', '/fakedir0/export.d')277 CONF.set_default('ganesha_export_template_dir',278 '/fakedir2/faketempl.d')279 CONF.set_default('ganesha_service_name', 'ganesha.fakeservice')280 CONF.set_default('ganesha_rados_store_enable', True)281 CONF.set_default('ganesha_rados_store_pool_name', 'ceph_pool')282 CONF.set_default('ganesha_rados_export_index', 'fake_index')283 CONF.set_default('ganesha_rados_export_counter', 'fake_counter')284 self._context = context.get_admin_context()285 self._execute = mock.Mock(return_value=('', ''))286 self.ceph_vol_client = mock.Mock()287 self.fake_conf = config.Configuration(None)288 self.fake_conf_dir_path = '/fakedir0/exports.d'289 self._helper = ganesha.GaneshaNASHelper2(290 self._execute, self.fake_conf, tag='faketag',291 ceph_vol_client=self.ceph_vol_client)292 self._helper.ganesha = mock.Mock()293 self._helper.export_template = {}294 self.share = fake_share.fake_share()295 self.rule1 = fake_share.fake_access(access_level='ro')296 self.rule2 = fake_share.fake_access(access_level='rw',297 access_to='10.0.0.2')298 @ddt.data(False, True)299 def test_init_helper_with_rados_store(self, rados_store_enable):300 CONF.set_default('ganesha_rados_store_enable', rados_store_enable)301 mock_template = mock.Mock()302 mock_ganesha_manager = mock.Mock()303 self.mock_object(ganesha.ganesha_manager, 'GaneshaManager',304 mock.Mock(return_value=mock_ganesha_manager))305 self.mock_object(self._helper, '_load_conf_dir',306 mock.Mock(return_value={}))307 self.mock_object(self._helper, '_default_config_hook',308 mock.Mock(return_value=mock_template))309 ret = self._helper.init_helper()310 if rados_store_enable:311 kwargs = {312 'ganesha_config_path': '/fakedir0/fakeconfig',313 'ganesha_export_dir': '/fakedir0/export.d',314 'ganesha_service_name': 'ganesha.fakeservice',315 'ganesha_rados_store_enable': True,316 'ganesha_rados_store_pool_name': 'ceph_pool',317 'ganesha_rados_export_index': 'fake_index',318 'ganesha_rados_export_counter': 'fake_counter',319 'ceph_vol_client': self.ceph_vol_client320 }321 else:322 kwargs = {323 'ganesha_config_path': '/fakedir0/fakeconfig',324 'ganesha_export_dir': '/fakedir0/export.d',325 'ganesha_service_name': 'ganesha.fakeservice',326 'ganesha_db_path': '/fakedir1/fake.db'327 }328 ganesha.ganesha_manager.GaneshaManager.assert_called_once_with(329 self._execute, '<no name>', **kwargs)330 self._helper._load_conf_dir.assert_called_once_with(331 '/fakedir2/faketempl.d', must_exist=False)332 self.assertEqual(mock_ganesha_manager, self._helper.ganesha)333 self._helper._default_config_hook.assert_called_once_with()334 self.assertEqual(mock_template, self._helper.export_template)335 self.assertIsNone(ret)336 @ddt.data(False, True)337 def test_init_helper_conf_dir_empty(self, conf_dir_empty):338 mock_template = mock.Mock()339 mock_ganesha_manager = mock.Mock()340 self.mock_object(ganesha.ganesha_manager, 'GaneshaManager',341 mock.Mock(return_value=mock_ganesha_manager))342 if conf_dir_empty:343 self.mock_object(self._helper, '_load_conf_dir',344 mock.Mock(return_value={}))345 else:346 self.mock_object(self._helper, '_load_conf_dir',347 mock.Mock(return_value=mock_template))348 self.mock_object(self._helper, '_default_config_hook',349 mock.Mock(return_value=mock_template))350 ret = self._helper.init_helper()351 ganesha.ganesha_manager.GaneshaManager.assert_called_once_with(352 self._execute, '<no name>',353 ganesha_config_path='/fakedir0/fakeconfig',354 ganesha_export_dir='/fakedir0/export.d',355 ganesha_service_name='ganesha.fakeservice',356 ganesha_rados_store_enable=True,357 ganesha_rados_store_pool_name='ceph_pool',358 ganesha_rados_export_index='fake_index',359 ganesha_rados_export_counter='fake_counter',360 ceph_vol_client=self.ceph_vol_client)361 self._helper._load_conf_dir.assert_called_once_with(362 '/fakedir2/faketempl.d', must_exist=False)363 self.assertEqual(mock_ganesha_manager, self._helper.ganesha)364 if conf_dir_empty:...
dbcontrol.py
Source:dbcontrol.py
1import hashlib2import binascii3from .mariadbhelper import *4from .tui.cli import crt5class UsernameNotFound(Exception):6 """Exception UsernameNotFound."""7 def __init__(self, message="Username not found"):8 self.message = message9 super().__init__(self.message)10class WrongPassword(Exception):11 """Exception WrongPassword."""12 def __init__(self, message="Wrong password"):13 self.message = message14 super().__init__(self.message)15class DBControl(object):16 def __init__(self):17 """Initializes DBControl.""" 18 self._helper = MariaDBHelper()19 self._helper.bindErrorCallback(crt.writeError)20 def start(self):21 """22 Starts DBControl.23 Raises:24 ConnectionNotEstablished: raised when MariaDBHelper isn't connected25 """26 self._helper.connect()27 if not self._helper.isConnected():28 raise ConnectionNotEstablished()29 def stop(self):30 """Stops DBControl.""" 31 if self._helper.isConnected():32 self._helper.disconnect()33 def fetchAppId(self, appId):34 """35 Fetchs App ID.36 WHERE THE HELL IS THIS CALLED???37 Args:38 appId (int): Application ID39 Returns:40 [type]: [description] 41 """ 42 self.start()43 self._helper \44 .Select([("`key`", None)]) \45 .From("apps") \46 .Where("appid=?") \47 .execute((appId,))48 self._helper.resetQuery()49 try:50 record = self._helper.getCursor().next()[0]51 #for (r,) in self._helper.getCursor():52 # record = r53 self.stop()54 return record55 except (StopIteration, Exception, mariadb.Error):56 self.stop()57 return None58 def getHMACKey(self):59 """60 Gets HMAC Key from the config.61 Returns:62 str: HMAC key63 """ 64 return self._helper.config['VALIDATION']['hmac']65 def valueExists(self, table, field, value):66 """67 Checks if value exists.68 Args:69 table (str): Table70 field (str): Field71 value (str): Value72 Returns:73 bool: True IF exists ELSE False74 """ 75 self.start()76 self._helper \77 .Select([(field, None)]) \78 .From(table) \79 .Where(f"{field}=?") \80 .execute((value,))81 self._helper.resetQuery()82 ok = False83 for (c,) in self._helper.getCursor():84 ok = True85 self.stop()86 return ok87 def userExists(self, username):88 """89 Checks if user exists.90 Args:91 username (str): Username92 Returns:93 bool: True IF exists ELSE False94 """95 return self.valueExists(96 table = "utilizadores",97 field = "username",98 value = username99 )100 def emailExists(self, email):101 """102 Checks if an email already exists.103 Args:104 email (str): email105 Returns:106 bool: True IF exists ELSE False107 """108 return self.valueExists(109 table = "utilizadores",110 field = "email",111 value = email112 )113 def loginUser(self, username, password):114 """115 Logs the user in.116 Args:117 username (str): Username118 password (str): Password119 Raises:120 Exception: Generic Exception121 UsernameNotFound: Could not find username122 WrongPassword: Wrong Password123 Returns:124 (bool, int): (Success, User ID)125 """126 self.start()127 key, salt = "", ""128 self._helper \129 .Select([("id_user", None), ("password", None), ("salt", None)]) \130 .From("utilizadores") \131 .Where("username=?") \132 .execute((username,))133 134 self._helper.resetQuery()135 try:136 for (x, y, z) in self._helper.getCursor():137 (id_user, key, salt) = (x, y, z)138 if (password == "" or salt == ""):139 raise Exception()140 except (Exception, mariadb.Error) as ex:141 raise UsernameNotFound(f"User '{username}' does not exist.")142 self.stop()143 new_key = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), binascii.unhexlify(salt), 100000)144 if new_key == binascii.unhexlify(key):145 return (True, id_user)146 else:147 raise WrongPassword(f"Wrong password for user '{username}'.")148 def registerUser(self, username, password, email):149 """150 Registers User.151 Generates salt, calculates sha256 (10000x)152 Args:153 username (str): Username154 password (str): Password155 email (str): Email156 Returns:157 bool: True IF successful ELSE False158 """159 self.start()160 salt = os.urandom(32) # A new salt for this user161 key = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, 100000)162 try:163 self._helper \164 .InsertInto("utilizadores", ["username", "email", "password", "salt"]) \165 .execute((username, email, binascii.hexlify(key), binascii.hexlify(salt),))166 self._helper.commit()167 except (Exception, mariadb.Error) as ex:168 crt.writeError(f"Error at database: {ex}")169 self._helper.resetQuery()170 self.stop()171 return False172 self._helper.resetQuery()173 self.stop()174 return True175 def addCypherChallenge(self, id_user, tip, msg, val, iv, hmacdb, algorithm):176 """177 Adds cypher challenge.178 Args:179 id_user (int): User ID180 tip (str): Challenge Tip181 msg (str): Message182 val (str): Value183 iv (str): Inicializer Vector184 hmacdb (str): HMAC185 algorithm (str): Algorithm186 Returns:187 bool: True IF successful ELSE False188 """ 189 self.start()190 try:191 self._helper \192 .InsertInto(193 "desafios_cifras",194 ["id_user", "dica", "resposta", "texto_limpo", "iv", "hmac", "algoritmo"] ) \195 .execute((id_user, tip, msg, val, iv, hmacdb, algorithm))196 self._helper.commit()197 except (Exception, mariadb.Error) as ex:198 crt.writeError(f"Error at database: {ex}")199 self._helper.resetQuery()200 self.stop()201 return False202 self._helper.resetQuery()203 self.stop()204 return True205 def getAllCypherChallenges(self):206 """207 Gets every cypher challenge.208 Returns:209 (list, Any): Row Headers, Row Values210 None: When failed211 """212 self.start()213 try:214 self._helper \215 .Select([216 ("desafios_cifras.id_desafio_cifras", "ID"),217 ("desafios_cifras.algoritmo", None),218 ("utilizadores.username", "Proposto por")]) \219 .From("desafios_cifras") \220 .InnerJoin("utilizadores", on="desafios_cifras.id_user=utilizadores.id_user") \221 .execute()222 row_headers=[x[0] for x in self._helper.getCursor().description]223 rv = self._helper.getCursor().fetchall()224 self._helper.resetQuery()225 self.stop()226 return (row_headers, rv)227 except (Exception, mariadb.Error) as ex:228 crt.writeError(f"Error at database: {ex}")229 self._helper.resetQuery()230 self.stop()231 return None232 def getCypherChallenge(self, id_challenge):233 """234 Gets cypher challenge.235 Args:236 id_challenge (int): Challenge ID237 Returns:238 dict: {239 'answer': Answer, 240 'tip': , 241 'algorithm': Challenge algorithm, 242 'plaintext': Plain Text,243 'iv': Initialized Vector,244 'hmac': HMAC,245 'username': Username246 }247 None: When failed248 """249 self.start()250 try:251 self._helper \252 .Select([253 ("desafios_cifras.resposta", None),254 ("desafios_cifras.dica", None),255 ("desafios_cifras.algoritmo", None),256 ("desafios_cifras.texto_limpo", None),257 ("desafios_cifras.iv", None),258 ("desafios_cifras.hmac", None),259 ("utilizadores.username", None) ]) \260 .From("desafios_cifras") \261 .InnerJoin("utilizadores", on="desafios_cifras.id_user=utilizadores.id_user") \262 .Where("id_desafio_cifras=?") \263 .execute((id_challenge,))264 self._helper.resetQuery()265 for (a, t, x, p, i, hm, u) in self._helper.getCursor():266 answer = a267 tip = t268 algorithm = x269 plaintext = p270 iv = i271 hmacdb = hm272 username = u273 self.stop()274 return {275 'answer' : answer,276 'tip' : tip,277 'algorithm' : algorithm,278 'plaintext' : plaintext,279 'iv' : iv,280 'hmac' : hmacdb,281 'username' : username282 }283 except (Exception, mariadb.Error) as ex:284 crt.writeError(f"Error at database: {ex}")285 self._helper.resetQuery()286 self.stop()287 return None288 def getCypherLastTry(self, id_user, id_challenge):289 """290 Gets last try of cypher.291 Args:292 id_user (int): UserID293 id_challenge (int): ChallengeID294 Returns:295 float: Last Date of an attempt296 """297 last_date = None298 self.start()299 try:300 self._helper \301 .Select([("data_ultima_tentativa", None)]) \302 .From("utilizadores_cifras") \303 .Where("id_user=? AND id_desafio_cifras=?") \304 .OrderBy(305 predicate="data_ultima_tentativa",306 desc=True,307 limit=1308 ).execute((id_user, id_challenge))309 for (ld,) in self._helper.getCursor():310 last_date = ld311 except (Exception, mariadb.Error) as ex:312 crt.writeError(f"Error at database: {ex}")313 self._helper.resetQuery()314 self.stop()315 return last_date316 def updateCypherChallengeTry(self, id_user, id_challenge, date, success):317 """318 Adds attempt of concluding a challenge.319 Args:320 id_user (int): UserID321 id_challenge (int): ChallengeID322 date (float): Date of the attempt323 success (bool): Success324 Returns:325 bool: True IF successful ELSE False326 """327 self.start()328 try:329 self._helper \330 .InsertInto(331 table="utilizadores_cifras",332 keys=[333 "id_user",334 "id_desafio_cifras",335 "data_ultima_tentativa",336 "sucesso"337 ]338 ).execute((id_user, id_challenge, date, success))339 self._helper.commit()340 except (Exception, mariadb.Error) as ex:341 crt.writeError(f"Error at database: {ex}")342 self._helper.resetQuery()343 self.stop()344 return False345 self._helper.resetQuery()346 self.stop()347 return True348 def addHashChallenge(self, id_user, tip, msg, algorithm):349 """350 Adds hash challenge.351 Args:352 id_user (int): UserID353 tip (str): Tip354 msg (str): Message355 algorithm (str): Algorithm356 Returns:357 bool: True IF successful ELSE False358 """359 self.start()360 try:361 self._helper \362 .InsertInto(363 "desafios_hash",364 ["id_user", "dica", "resposta", "algoritmo"] ) \365 .execute((id_user, tip, msg, algorithm))366 self._helper.commit()367 except (Exception, mariadb.Error) as ex:368 crt.writeError(f"Error at database: {ex}")369 self._helper.resetQuery()370 self.stop()371 return False372 self._helper.resetQuery()373 self.stop()374 return True375 def getAllHashChallenges(self):376 """377 Gets every cypher challenge.378 Returns:379 (list, Any): Row Headers, Row Values380 None: When failed381 """382 self.start()383 try:384 self._helper \385 .Select([386 ("desafios_hash.id_desafio_hash", "ID"),387 ("desafios_hash.algoritmo", None),388 ("utilizadores.username", "Proposto por")]) \389 .From("desafios_hash") \390 .InnerJoin("utilizadores", on="desafios_hash.id_user=utilizadores.id_user") \391 .execute()392 row_headers=[x[0] for x in self._helper.getCursor().description]393 rv = self._helper.getCursor().fetchall()394 self._helper.resetQuery()395 self.stop()396 return (row_headers, rv)397 except (Exception, mariadb.Error) as ex:398 crt.writeError(f"Error at database: {ex}")399 self._helper.resetQuery()400 self.stop()401 return None402 def getHashChallenge(self, id_challenge):403 """404 Gets hash challenge.405 Args:406 id_challenge (int): Challenge ID407 Returns:408 dict: {409 'answer': str,410 'tip': str,411 'algorithm': str,412 'username': str413 }414 None: When fails 415 """ 416 self.start()417 try:418 self._helper \419 .Select([420 ("desafios_hash.resposta", None),421 ("desafios_hash.dica", None),422 ("desafios_hash.algoritmo", None),423 ("utilizadores.username", None) ]) \424 .From("desafios_hash") \425 .InnerJoin("utilizadores", on="desafios_hash.id_user=utilizadores.id_user") \426 .Where("id_desafio_hash=?") \427 .execute((id_challenge,))428 self._helper.resetQuery()429 for (a, t, x, u) in self._helper.getCursor():430 answer = a431 tip = t432 algorithm = x433 username = u434 self.stop()435 return {436 'answer' : answer,437 'tip' : tip,438 'algorithm' : algorithm,439 'username' : username440 }441 except (Exception, mariadb.Error) as ex:442 crt.writeError(f"Error at database: {ex}")443 self._helper.resetQuery()444 self.stop()445 return None446 def getHashLastTry(self, id_user, id_challenge):447 """448 Gets last try of hash challenge.449 Args:450 id_user (int): UserID451 id_challenge (int): ChallengeID452 Returns:453 float: Last Date of an attempt454 """455 last_date = None456 self.start()457 try:458 self._helper \459 .Select([("data_ultima_tentativa", None)]) \460 .From("utilizadores_hash") \461 .Where("id_user=? AND id_desafio_hash=?") \462 .OrderBy(463 predicate="data_ultima_tentativa",464 desc=True,465 limit=1466 ).execute((id_user, id_challenge))467 for (ld,) in self._helper.getCursor():468 last_date = ld469 except (Exception, mariadb.Error) as ex:470 crt.writeError(f"Error at database: {ex}")471 self._helper.resetQuery()472 self.stop()473 return last_date474 def updateHashChallengeTry(self, id_user, id_challenge, date, success):475 """476 Adds attempt of concluding a challenge.477 Args:478 id_user (int): UserID479 id_challenge (int): ChallengeID480 date (float): Date of the attempt481 success (bool): Success482 Returns:483 bool: True IF successful ELSE False484 """485 self.start()486 try:487 self._helper \488 .InsertInto(489 table="utilizadores_hash",490 keys=[491 "id_user",492 "id_desafio_hash",493 "data_ultima_tentativa",494 "sucesso"495 ]496 ).execute((id_user, id_challenge, date, success))497 self._helper.commit()498 except (Exception, mariadb.Error) as ex:499 crt.writeError(f"Error at database: {ex}")500 self._helper.resetQuery()501 self.stop()502 return False503 self._helper.resetQuery()504 self.stop()505 return True506 def getAllScoreboard(self):507 """Gets all scores."""508 self.start()509 try:510 self._helper \511 .AddCustomQuery(512"""513select514u.username as 'User',515CAST(if(a.CypherOK is null, 0, a.CypherOK) AS int) as 'Cypher',516CAST(if(a.HashOK is null, 0, a.HashOK) AS int) as 'Hash',517CAST(if(a.CypherOK is null, 0, a.CypherOK) + if(a.HashOK is null, 0, a.HashOK) as int) as 'Total'518from519(520select distinct521uc.id_user as 'CypherID',522sum(uc.sucesso) as 'CypherOK',523uc.id_desafio_cifras as 'CypherChal',524r.HashID,525r.HashOK,526r.HashChal527from utilizadores_cifras uc528left join529(530select distinct531uh.id_user as 'HashID',532sum(uh.sucesso) as 'HashOK',533uh.id_desafio_hash as 'HashChal'534from utilizadores_hash uh535where uh.sucesso=1536group by uh.id_user537) r on r.HashID = uc.id_user538where uc.sucesso=1539group by uc.id_user540) a541left join utilizadores u on u.id_user = a.CypherID542order by Total desc543"""544 ).execute()545 row_headers=[x[0] for x in self._helper.getCursor().description]546 rv = self._helper.getCursor().fetchall()547 self._helper.resetQuery()548 self.stop()549 return (row_headers, rv)550 except (Exception, mariadb.Error) as ex:551 crt.writeError(f"Error at database: {ex}")552 self._helper.resetQuery()553 self.stop()554 return None555 def getEmail(self, id_user):556 """557 Gets Email.558 Args:559 id_user (str): User ID560 Returns:561 str: User's Email562 None: When fails563 """ 564 self.start()565 try:566 self._helper \567 .Select([("email", None)]) \568 .From("utilizadores") \569 .Where("id_user = ?") \570 .execute((id_user,))571 self._helper.resetQuery()572 for (email,) in self._helper.getCursor():573 useremail = email574 self.stop()575 return useremail576 except (Exception, mariadb.Error) as ex:577 crt.writeError(f"Error at database: {ex}")578 self._helper.resetQuery()579 self.stop()580 return None581 582 583 def getUserCreatedAmount(self, id_user):584 """585 Gets user created challenges amount.586 Args:587 id_user (int): User ID588 589 Returns:590 dict: {591 'cypher': int,592 'hash': int,593 'total': int594 }595 None: when fails596 """ 597 self.start()598 try:599 self._helper\600 .AddCustomQuery(601"""602select603 count(dc.id_desafio_cifras) as 'Cypher',604 r.Hash,605 count(dc.id_desafio_cifras) + r.Hash as 'Total'606from desafios_cifras dc607left join (608select 609 count(dh.id_desafio_hash) as 'Hash'610from desafios_hash dh611where dh.id_user = ?612) r on true613where dc.id_user = ?614"""615 ).execute((id_user, id_user))616 self._helper.resetQuery()617 for (c, h, total) in self._helper.getCursor():618 Cypher = c619 Hash = h620 Total = total621 self.stop()622 return {623 'cypher': Cypher,624 'hash': Hash,625 'total': Total626 }627 except (Exception, mariadb.Error) as ex:628 crt.writeError(f"Error at database: {ex}")629 self._helper.resetQuery()630 self.stop()631 return None...
buffet_IntermittentConnectivity.py
Source:buffet_IntermittentConnectivity.py
1# Copyright 2015 The Chromium Authors. All rights reserved.2# Use of this source code is governed by a BSD-style license that can be3# found in the LICENSE file.4from autotest_lib.client.bin import test5from autotest_lib.client.common_lib.cros.tendo import buffet_tester6class buffet_IntermittentConnectivity(test.test):7 """Test that buffet reconnects if it loses connectivity."""8 version = 19 def initialize(self):10 self._helper = buffet_tester.BuffetTester()11 def run_once(self):12 """Test entry point."""13 # Erase all buffet state and restart it pointing to our fake14 # server, register with the cloud and check we can poll for15 # commands.16 self._helper.restart_buffet(reset_state=True)17 self._helper.check_buffet_status_is(buffet_tester.STATUS_UNCONFIGURED)18 device_id = self._helper.register_with_server()19 self._helper.check_buffet_is_polling(device_id)20 # Now make fake_device_server fail all request from Buffet21 # with HTTP Error Code 500 (Internal Server Error) and check22 # that we transition to the CONNECTING state.23 self._helper._fail_control_client.start_failing_requests()24 self._helper.check_buffet_status_is(25 buffet_tester.STATUS_CONNECTING,26 expected_device_id=device_id,27 timeout_seconds=20)28 # Stop failing request from and check that we transition to29 # the CONNECTED state.30 self._helper._fail_control_client.stop_failing_requests()31 self._helper.check_buffet_status_is(32 buffet_tester.STATUS_CONNECTED,33 expected_device_id=device_id,34 timeout_seconds=20)35 self._helper.check_buffet_is_polling(device_id)36 def cleanup(self):...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack 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!!
Was this article helpful?
