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

How to use _open_connection method in autotest

Best Python code snippet using autotest_python

sqlite.py

Source:sqlite.py

...106            self._update_spec(job_output.spec)107        metrics = [(self._spec_oid, job_output.iteration, m.name, str(m.value), m.units, int(m.lower_is_better))108                   for m in job_output.metrics]109        if metrics:110            with self._open_connection() as conn:111                conn.executemany('INSERT INTO metrics VALUES (?,?,?,?,?,?)', metrics)112    def export_run_output(self, run_output, target_info):  # pylint: disable=unused-argument113        if not self._run_initialized:114            self._init_run(run_output)115        metrics = [(self._spec_oid, run_output.iteration, m.name, str(m.value), m.units, int(m.lower_is_better))116                   for m in run_output.metrics]117        if metrics:118            with self._open_connection() as conn:119                conn.executemany('INSERT INTO metrics VALUES (?,?,?,?,?,?)', metrics)120        info = run_output.info121        with self._open_connection() as conn:122            conn.execute('''UPDATE runs SET start_time=?, end_time=?, duration=?123                            WHERE OID=?''', (info.start_time, info.end_time, info.duration, self._run_oid))124    def _init_run(self, run_output):125        if not self.database:  # pylint: disable=access-member-before-definition126            self.database = os.path.join(run_output.basepath, 'results.sqlite')127        self.database = os.path.expandvars(os.path.expanduser(self.database))128        if not os.path.exists(self.database):129            self._init_db()130        elif self.overwrite:  # pylint: disable=no-member131            os.remove(self.database)132            self._init_db()133        else:134            self._validate_schema_version()135        self._update_run(run_output.info.uuid)136        # if the database file happens to be in the output directory, add it as an137        # artifiact; if it isn't, then RunOutput doesn't need to keep track of it.138        if not os.path.relpath(self.database, run_output.basepath).startswith('..'):139            run_output.add_artifact('sqlitedb', self.database, kind='export')140        self._run_initialized = True141    def _init_db(self):142        with self._open_connection() as conn:143            for command in SCHEMA:144                conn.execute(command)145    def _validate_schema_version(self):146        with self._open_connection() as conn:147            try:148                c = conn.execute('SELECT schema_version FROM __meta')149                found_version = c.fetchone()[0]150            except sqlite3.OperationalError:151                message = '{} does not appear to be a valid WA results database.'.format(self.database)152                raise OutputProcessorError(message)153            if found_version != SCHEMA_VERSION:154                message = 'Schema version in {} ({}) does not match current version ({}).'155                raise OutputProcessorError(message.format(self.database, found_version, SCHEMA_VERSION))156    def _update_run(self, run_uuid):157        with self._open_connection() as conn:158            conn.execute('INSERT INTO runs (uuid) VALUES (?)', (run_uuid,))159            conn.commit()160            c = conn.execute('SELECT OID FROM runs WHERE uuid=?', (run_uuid,))161            self._run_oid = c.fetchone()[0]162    def _update_spec(self, spec):163        self._last_spec = spec164        spec_tuple = (spec.id, self._run_oid, spec.iterations, spec.label, spec.workload_name,165                      json.dumps(spec.boot_parameters.to_pod()),166                      json.dumps(spec.runtime_parameters.to_pod()),167                      json.dumps(spec.workload_parameters.to_pod()))168        with self._open_connection() as conn:169            conn.execute('INSERT INTO workload_specs VALUES (?,?,?,?,?,?,?,?)', spec_tuple)170            conn.commit()171            c = conn.execute('SELECT OID FROM workload_specs WHERE run_oid=? AND id=?', (self._run_oid, spec.id))172            self._spec_oid = c.fetchone()[0]173    @contextmanager174    def _open_connection(self):175        conn = sqlite3.connect(self.database)176        try:177            yield conn178        finally:...

database_requests.py

Source:database_requests.py

...4# Location of database file5DB_PATH = f"{pathlib.Path(__file__).parent.resolve()}/database.db"6# Sign in tokens are valid for 30 minutes7TOKEN_VALIDITY_PERIOD = 60*308def _open_connection():9    db_connection = connect(DB_PATH)10    db_cursor = db_connection.cursor()11    db_connection.execute("PRAGMA foreign_keys = ON;")12    return db_cursor13def _close_connection(db_cursor):14    db_connection = db_cursor.connection15    db_cursor.close()16    db_connection.close()17def init_db():18    db_connection = _open_connection()19    db_connection.execute("BEGIN;")20    db_connection.execute("CREATE TABLE IF NOT EXISTS users ("+ \21                            "username TEXT NOT NULL,"+ \22                            "password TEXT NOT NULL,"+ \23                            "PRIMARY KEY (username));")24    db_connection.execute("COMMIT;")25    db_connection.execute("BEGIN;")26    db_connection.execute("CREATE TABLE IF NOT EXISTS logins ("+ \27                            "token TEXT NOT NULL,"+ \28                            "username TEXT NOT NULL,"+ \29                            "issued_timestamp INT NOT NULL,"+ \30                            "PRIMARY KEY (token),"+ \31                            "FOREIGN KEY (username) REFERENCES users (username));")32    db_connection.execute("COMMIT;")33    _close_connection(db_connection)34def create_user(username, password):35    user = (username, password)36    try:37        db_connection = _open_connection()38        db_connection.execute("BEGIN;")39        db_connection.execute("INSERT INTO users VALUES (?,?);",user)40        db_connection.execute("COMMIT;")41        _close_connection(db_connection)42        return (True,None)43    except IntegrityError:44        return (True,None)45    except Exception as e:46        print(e)47        return (False,None)48def create_login(username, token):49    issued_timestamp = int(time())50    login = (token,username,issued_timestamp)51    try:52        db_connection = _open_connection()53        db_connection.execute("BEGIN;")54        db_connection.execute("INSERT OR REPLACE INTO logins VALUES (?,?,?);",login)55        db_connection.execute("COMMIT;")56        _close_connection(db_connection)57        return (True,None)58    except:59        return (False,None)60def delete_token(token):61    try:62        db_connection = _open_connection()63        db_connection.execute("BEGIN;")64        db_connection.execute("DELETE FROM logins WHERE token = ?;",[token])65        db_connection.execute("COMMIT;")66        _close_connection(db_connection)67        return True68    except:69        return False70def update_token(token):71    try:72        current_time = int(time())73        db_connection = _open_connection()74        db_connection.execute("BEGIN;")75        db_connection.execute("UPDATE logins SET issued_timestamp = ? WHERE token = ?;",[current_time,token])76        db_connection.execute("COMMIT;")77        _close_connection(db_connection)78        return True79    except:80        return False81def is_valid_token(token):82    try:83        current_time = int(time())84        db_connection = _open_connection()85        db_connection.execute("SELECT issued_timestamp FROM logins WHERE token = ?;",[token])86        valid_token = db_connection.fetchone()87        _close_connection(db_connection)88        if valid_token is None:89            return (True,False)90        issued_timestamp = valid_token[0]91        if current_time < issued_timestamp:92            return (False,False)93        if current_time - issued_timestamp > TOKEN_VALIDITY_PERIOD:94            success = delete_token(token)95            return (success,False)96        success = update_token(token)97        return (success,True)98    except:99        return (False,False)100def select_password(username):101    try:102        db_connection = _open_connection()103        db_connection.execute("SELECT password FROM users WHERE username = ?;",[username])104        password = db_connection.fetchone()105        _close_connection(db_connection)106        if password is None:107            return (True,None)108        return (True,password[0])109    except:110        return (False,None)111def select_users():112    try:113        db_connection = _open_connection()114        db_connection.execute("SELECT * FROM users;")115        users = db_connection.fetchall()116        _close_connection(db_connection)117        if users is None:118            return (True,None)119        return (True,users)120    except:...

database_manager.py

Source:database_manager.py

...7    def init(self, *args, **kwds):8        self.config = json_reader.read(CONFIG_NAME)9        self.verify_table_exists('users')10        self.connection = None11    def _open_connection(self):12        self.connection = sqlite3.connect(self.config['db']['filename'])13    def _close_connection(self):14        self.connection.close()15    def create_users_table(self):16        self._open_connection()17        cur = self.connection.cursor()18        cur.execute(''' CREATE TABLE `users` (`user_id` INTEGER PRIMARY KEY, `positive_relationship` INTEGER, `negative_relationship` INTEGER) ''')19        self.connection.commit()20        self._close_connection()21    def insert_user(self, user_id, positive_rep: int = 0, negative_rep: int = 0):22        self._open_connection()23        cur = self.connection.cursor()24        cur.execute(''' INSERT INTO `users` (`user_id`, `positive_relationship`, `negative_relationship`) VALUES (?, ?, ?) ''', (user_id, positive_rep, negative_rep))25        print('Added entry {0}'.format(user_id))26        self.connection.commit()27        self._close_connection()28    def update_user(self, user_id: int, positive_rep: int, negative_rep: int):29        self._open_connection()30        cur = self.connection.cursor()31        cur.execute(''' UPDATE `users` SET `positive_relationship` = ?, `negative_relationship` = ? WHERE `user_id` = ? ''', (positive_rep, negative_rep, user_id))32        print('Updated entry {0}'.format(user_id))33        self.connection.commit()34        self._close_connection()35    def verify_user_exists(self, user_id: int):36        self._open_connection()37        cur = self.connection.cursor()38        cur.execute(''' SELECT * FROM `users` WHERE `user_id` = ? ''', (user_id,))39        result = cur.fetchone()40        self._close_connection()41        return result42    def verify_table_exists(self, table_name: str):43        self._open_connection()44        cur = self.connection.cursor()45        cur.execute(''' SELECT count(`name`) as 'exists' FROM `sqlite_master` WHERE `type` = 'table' AND `name` = 'users' ''')46        if cur.fetchone()[0] == 1:47            return48        self.connection.commit()49        self._close_connection()...

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.