Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.