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

How to use __convert_result method in hypothesis

Best Python code snippet using hypothesis

queue_manager.py

Source:queue_manager.py

...74            AND75                deleted = 076            """,77            (song.uri,))78        return self.__convert_result(c)79    def get_queue(self, index=None):80        if index == None:81            index = self.current_id82        c = self.db.cursor()83        c.execute("""84            SELECT *85            FROM queue86            WHERE87                id = ?88            LIMIT 189            """,90            (index,))91        return self.__convert_result(c, 1)92    def max_index(self):93        c = self.db.cursor()94        c.execute("""95            SELECT MAX(id) as max96            FROM queue97            """)98        data = c.fetchone()99        if data == None:100            return 0101        else:102            return data['max']103    def current_index(self):104        c = self.db.cursor()105        c.execute("""106            SELECT *107            FROM queue_data108            WHERE109                key = 'current'110            LIMIT 1111            """)112        data = c.fetchone()113        if data == None:114            return 0115        else:116            return data['num']117    def random(self, limit=1):118        c = self.db.cursor()119        c.execute("""120            SELECT *121            FROM queue122            WHERE123                deleted = 0124            ORDER BY RANDOM()125            LIMIT ?126            """,127            (limit,))128        return self.__convert_result(c, limit)129    def by_user(self, user, limit=1):130        c = self.db.cursor()131        c.execute("""132            SELECT *133            FROM queue134            WHERE135                user = ?136            AND137                deleted = 0138            ORDER BY id DESC139            LIMIT ?140            """,141            (user,limit))142        return self.__convert_result(c, limit)143    def delete(self, index):144        c = self.db.cursor()145        try:146            c.execute("""147                UPDATE queue148                SET deleted=1149                WHERE150                    id = ?151            """, (index,))152            c.close()153            self.db.commit()154            return True155        except:156            return False157    def next(self, limit=1):158        c = self.db.cursor()159        c.execute("""160            SELECT *161            FROM queue162            WHERE163                id > ?164            AND165                deleted = 0166            LIMIT ?167            """,168            (self.current_id,limit))169        return self.__convert_result(c, limit)170    def on_track_end(self, current):171        # Sometimes end track triggers multiple times ensure 2 second cooldown  period172        now = int(time.time())173        if self._end_track_timeout + 2 > now:174            print u'End track timeout prevention'175            return176        else:177            self._end_track_timeout = now178        nxt = self.next()179        q = self.get_queue()180        if nxt:181            self.current(nxt.id)182            self.player.play(nxt.song, self.MODE_QUEUE)183            return184        if self.mode == self.MODE_RELATED:185            result = self.player.related(song=current, single=True)186            if result:187                self.player.play(result, self.MODE_RELATED)188                return189        if self.player.mode == self.MODE_QUEUE and \190            q != None and \191            q.song == current and \192            self.player.playing:193                self.current(self.current_id + 1)194        r = self.random()195        if r != None:196            self.player.play(r.song, self.MODE_RANDOM)197            return198    def __convert_result(self, cursor, limit=None):199        if limit == 1:200            data = cursor.fetchone()201            if data == None:202                return None203            return Queue().from_db(data)204        else:205            return map(lambda x: Queue().from_db(x), cursor.fetchall())206    def __db_init(self):207        c = self.db.cursor()208        # Create table209        c.execute("""210            CREATE TABLE IF NOT EXISTS queue(211                id INTEGER PRIMARY KEY AUTOINCREMENT,212                uri TEXT,...

database.py

Source:database.py

...38            port=db_user_info["port"],39            user=db_user_info["user"],40            password=db_user_info["password"]41        )42    def __convert_result(self, cursor: object) -> list:43        logging.debug("[aico_db][__convert_result] start")44        result = []45        try:46            fetchall = cursor.fetchall()47        except psycopg2.ProgrammingError as e:48            logging.debug("[aico_db][__convert_result] no return")49            return result50        if not len(fetchall):51            logging.debug("[aico_db][__convert_result] no data")52            return result53        headers = [x[0] for x in cursor.description]54        for row in fetchall:55            result.append(dict(zip(headers, row)))56        logging.debug("[aico_db][__convert_result] end")57        return result58    def __excute_sql(59        self, reqSQL: str, reqParams: tuple = None, is_many: bool = False60    ) -> list:61        self.result["status"] = True62        conn = None63        cursor = None64        try:65            conn = self.__get_dbconnect()66            with conn.cursor() as cursor:67                if self.sql_type is SQL_TYPE.SELECT:68                    if reqParams is not None:69                        logging.debug("[aico_db][__excute_sql] select and params")70                        cursor.execute(reqSQL, reqParams)71                    else:72                        logging.debug("[aico_db][__excute_sql] select and no params")73                        cursor.execute(reqSQL)74                    self.result["data"] = self.__convert_result(cursor)75                elif self.sql_type is SQL_TYPE.NO_SELECT:76                    if is_many and reqParams is not None:77                        logging.debug("[aico_db][__excute_sql] no select and many")78                        cursor.executemany(reqSQL, reqParams)79                    elif reqParams is not None:80                        logging.debug("[aico_db][__excute_sql] no select and params")81                        cursor.execute(reqSQL, reqParams)82                    else:83                        logging.debug("[aico_db][__excute_sql] no select and no params")84                        cursor.execute(reqSQL)85                    self.result["data"] = self.__convert_result(cursor)86                    conn.commit()87                else:88                    logging.debug("[aico_db][__excute_sql][Exception] Nothing")89                    raise Exception('[DatabaseMgmt][__excute_sql] Nothing')90        except Exception as error:91            logging.debug("[aico_db][__excute_sql][error]")92            logging.error(93                "[DatabaseMgmt][__excute_sql] Error while connecting to Aurora", error94            )95            self.result["status"] = False96            self.result[97                "data"98            ] = "[DatabaseMgmt][__excute_sql] Error while connecting to Aurora > error : {}".format(99                str(error)...

data_source.py

Source:data_source.py

...14    """ Constructor """15    self.factors = factors16    self.database = db17    self.week = week18  def __convert_result(self, data: list) -> dict:19    """ Convert the given students list into a dict """20    students = {}21    for student in data:22      _id, name = student[0], student[1]23      lesson = Lesson(values={24        key: value for key, value in zip(self.factors, student[2:])25      })26      if students.get(name) is None:27        students.update({name: Student(_id, name, [lesson])})28      else:29        students[name].lessons.append(lesson)30    return students31  def get_students_by(self, ids: tuple = None, names: tuple = None) -> dict:32    """ Get students by given list of ids or names """33    students_data = self.database.get_students_by(34      *self.week.values,35      ids=ids, names=names,36      factors=self.factors)37    return self.__convert_result(students_data)38  def get_students_by_week(self) -> dict:39    """ Get all students with classes on given week """40    students_data = self.database.get_students_by_week(41      *self.week.values,42      factors=self.factors)...

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.