Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use list_users method in tempest
Best Python code snippet using tempest_python
table.py
Source:table.py 
1#importing libraries2import sqlite33database_name='Details.db'4database_name_status="Status.db"5database_name_command="Command.db"6#creating necessary tables7def creating_table():8 connection=sqlite3.connect(database_name)9 cursor=connection.cursor()10 create_table_USERS="CREATE TABLE IF NOT EXISTS USERS (ID INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,USERNAME text,PASSWORD text,NAME text,EMAIL text)"11 create_table_ADMIN='CREATE TABLE IF NOT EXISTS [ADMIN] ([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,[USERNAME] TEXT,[ADMIN] TEXT NOT NULL,FOREIGN KEY(USERNAME) REFERENCES USERS(USERNAME))'12 create_table_ROUTER_MAPPING='CREATE TABLE IF NOT EXISTS [ROUTER_MAPPING] ([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,[USERNAME] TEXT NOT NULL,[ROUTER_ID] INTEGER NOT NULL,FOREIGN KEY(USERNAME) REFERENCES USERS(USERNAME))'13 create_table_ROUTER='CREATE TABLE IF NOT EXISTS [ROUTER]([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,ROUTER_ID TEXT,DEVICES TEXT,STATE TEXT)'14 create_table_USER_ADMIN_MAPPING='CREATE TABLE IF NOT EXISTS [USER_ADMIN_MAPPING] ([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,[USERNAME] TEXT NOT NULL,[ADMIN] TEXT,[CONTROL] TEXT)'15 create_table_USER_ADMIN_KEY='CREATE TABLE IF NOT EXISTS [ADMIN_KEY] ([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,ADMINKEY text)'16 create_table_ROUTER_PASSWORD='CREATE TABLE IF NOT EXISTS [ROUTER_PASSWORD]([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,ROUTER_ID TEXT,PASSWORD TEXT)'17 cursor.execute(create_table_USERS)18 cursor.execute(create_table_ADMIN)19 cursor.execute(create_table_ROUTER_MAPPING)20 cursor.execute(create_table_ROUTER)21 cursor.execute(create_table_USER_ADMIN_MAPPING)22 cursor.execute(create_table_USER_ADMIN_KEY)23 cursor.execute(create_table_ROUTER_PASSWORD)24 connection.commit()25 connection.close()26#deleting all tables27def deleting_table():28 connection=sqlite3.connect(database_name)29 cursor=connection.cursor()30 for i in {'USERS','ADMIN','ROUTER_MAPPING','ROUTER','USER_ADMIN_MAPPING'}:31 query='DROP TABLE '+i32 cursor.execute(query)33 connection.commit()34 connection.close()35#Start_USERS36#Table to store the username password email and name37#Case Non Sensitive38def insert_into_USERS(username,password,name,email):39 username=str(username).upper()40 password=str(password).upper()41 name=str(name).upper()42 email=str(email).upper()43 connection=sqlite3.connect(database_name)44 cursor=connection.cursor()45 insert_query = "INSERT INTO USERS (USERNAME,PASSWORD,NAME,EMAIL) VALUES (?,?,?,?)"46 cursor.execute(insert_query,(username,password,name,email))47 connection.commit()48 connection.close()49def content_USERS():50 list_users=list()51 connection=sqlite3.connect(database_name)52 cursor=connection.cursor()53 query='SELECT * FROM users'54 for row in cursor.execute(query):55 list_users.append(row)56 connection.commit()57 connection.close()58 return list_users59def get_USERS(username):60 username=str(username).upper()61 list_users=list()62 connection=sqlite3.connect(database_name)63 cursor=connection.cursor()64 query="SELECT * FROM USERS WHERE USERNAME = '" + str(username) +"'"65 for row in cursor.execute(query):66 list_users.append(row)67 connection.commit()68 connection.close()69 return list_users70def get_USERS_id(id):71 list_users=list()72 connection=sqlite3.connect(database_name)73 cursor=connection.cursor()74 query="SELECT USERNAME FROM USERS WHERE ID = '" + str(id) +"'"75 for row in cursor.execute(query):76 list_users.append(row)77 connection.commit()78 connection.close()79 if(len(list_users)!=0):80 return(list_users[0])81 else:82 return 083def del_user_USERS(username):84 username=str(username).upper()85 connection=sqlite3.connect(database_name)86 cursor=connection.cursor()87 query="DELETE FROM USERS WHERE USERNAME = '" + str(username) +"'"88 cursor.execute(query)89 connection.commit()90 connection.close()91def forgot_password(username,email,new_password):92 username=str(username).upper()93 email=str(email).upper()94 new_password=str(new_password).upper()95 connection=sqlite3.connect(database_name)96 cursor=connection.cursor()97 list_users=list()98 query="SELECT PASSWORD FROM USERS WHERE USERNAME = '"+str(username)+"' AND EMAIL = '"+str(email)+"'"99 for row in cursor.execute(query):100 list_users.append(row)101 if(len(list_users)==0):102 return 0103 query="UPDATE USERS SET PASSWORD = '"+new_password+"' WHERE USERNAME='"+str(username)+"' AND EMAIL = '"+str(email)+"'"104 cursor.execute(query)105 connection.commit()106 connection.close()107 return 1108#End_USERS109#Start_ADMIN110#Store if the user is just a normal one or a ADMIN111#Start_ADMIN112#Store if the user is just a normal one or a ADMIN113def insert_into_ADMIN(username,admin):114 username=str(username).upper()115 admin=str(admin).upper()116 connection=sqlite3.connect(database_name)117 cursor=connection.cursor()118 insert_query = "INSERT INTO ADMIN (USERNAME,ADMIN) VALUES (?,?)"119 cursor.execute(insert_query,(username,admin))120 connection.commit()121 connection.close()122def content_ADMIN():123 list_users=list()124 connection=sqlite3.connect(database_name)125 cursor=connection.cursor()126 query='SELECT * FROM ADMIN'127 for row in cursor.execute(query):128 list_users.append(row)129 connection.commit()130 connection.close()131 return list_users132def del_user_ADMIN(username):133 username=str(username).upper()134 connection=sqlite3.connect(database_name)135 cursor=connection.cursor()136 query="DELETE FROM ADMIN WHERE USERNAME = '" + str(username) +"'"137 cursor.execute(query)138 connection.commit()139 connection.close()140def get_ADMIN_STATUS(username):141 username=str(username).upper()142 connection=sqlite3.connect(database_name)143 cursor=connection.cursor()144 query="SELECT ADMIN FROM ADMIN WHERE USERNAME = '" + str(username) +"'"145 for row in cursor.execute(query):146 return row[0]147 connection.commit()148 connection.close()149##END ADMIN150#Admin_Table151#ROUTER_MAPPING START152def insert_into_ROUTER_MAPPING(username,router_id):153 username=str(username).upper()154 router_id=str(router_id).upper()155 connection=sqlite3.connect(database_name)156 cursor=connection.cursor()157 insert_query = "INSERT INTO ROUTER_MAPPING(USERNAME,ROUTER_ID) VALUES (?,?)"158 cursor.execute(insert_query,(username,router_id))159 connection.commit()160 connection.close()161def content_ROUTER_MAPPING():162 list_users=list()163 connection=sqlite3.connect(database_name)164 cursor=connection.cursor()165 query='SELECT * FROM ROUTER_MAPPING'166 for row in cursor.execute(query):167 list_users.append(row)168 connection.commit()169 connection.close()170 return list_users171def del_user_ROUTER_MAPPING_2(username,router_id):172 username=str(username).upper()173 router_id=str(router_id).upper()174 connection=sqlite3.connect(database_name)175 cursor=connection.cursor()176 query="DELETE FROM ROUTER_MAPPING WHERE USERNAME = '" + str(username) +"' AND ROUTER_ID='"+str(router_id)+"'"177 cursor.execute(query)178 connection.commit()179 connection.close()180def find_ROUTER_MAPPING_2(username,router_id):181 username=str(username).upper()182 router_id=str(router_id).upper()183 list_users=list()184 connection=sqlite3.connect(database_name)185 cursor=connection.cursor()186 query="SELECT * FROM ROUTER_MAPPING WHERE USERNAME = '" + str(username) +"' AND ROUTER_ID='"+str(router_id)+"'"187 for row in cursor.execute(query):188 list_users.append(row)189 if(len(list_users)==0):190 return 0191 else:192 return len(list_users)193 connection.commit()194 connection.close()195def del_user_ROUTER_MAPPING(username):196 username=str(username).upper()197 connection=sqlite3.connect(database_name)198 cursor=connection.cursor()199 query="DELETE FROM ROUTER_MAPPING WHERE USERNAME = '" + str(username) +"'"200 cursor.execute(query)201 connection.commit()202 connection.close()203def get_ROUTER_MAPPING(username):204 username=str(username).upper()205 list_users=list()206 connection=sqlite3.connect(database_name)207 cursor=connection.cursor()208 query="SELECT * FROM ROUTER_MAPPING WHERE USERNAME = '" + str(username) +"'"209 for row in cursor.execute(query):210 list_users.append(row)211 connection.commit()212 connection.close()213 return list_users214#END Router Mapping215#Router_Mapping Tablw216#USER_ADMIN_MAPPING START217def insert_into_USER_ADMIN_MAPPING(username,admin,control):218 username=str(username).upper()219 admin=str(admin).upper()220 control=str(control).upper()221 connection=sqlite3.connect(database_name)222 cursor=connection.cursor()223 search_query="SELECT * FROM USER_ADMIN_MAPPING WHERE USERNAME='"+username+"'"+" AND ADMIN='"+admin+"'"224 list_users=list()225 for row in cursor.execute(search_query):226 list_users.append(row)227 x=len(list_users)228 if(x==0):229 insert_query = "INSERT INTO USER_ADMIN_MAPPING(USERNAME,ADMIN,CONTROL) VALUES (?,?,?)"230 cursor.execute(insert_query,(username,admin,control))231 else:232 update_query="UPDATE USER_ADMIN_MAPPING SET CONTROL='"+control+"' WHERE USERNAME='"+username+"'"+" AND ADMIN='"+admin+"'"233 cursor.execute(update_query)234 connection.commit()235 connection.close()236def content_USER_ADMIN_MAPPING():237 list_users=list()238 connection=sqlite3.connect(database_name)239 cursor=connection.cursor()240 query='SELECT * FROM USER_ADMIN_MAPPING'241 for row in cursor.execute(query):242 list_users.append(row)243 connection.commit()244 connection.close()245 return list_users246def get_USER_ADMIN_MAPPING(username):247 username=str(username).upper()248 list_users=list()249 connection=sqlite3.connect(database_name)250 cursor=connection.cursor()251 query="SELECT ADMIN FROM USER_ADMIN_MAPPING WHERE USERNAME = '" + str(username) +"'"252 for row in cursor.execute(query):253 list_users.append(row[0])254 connection.commit()255 connection.close()256 return list_users257def del_user_USER_ADMIN_MAPPING(username):258 username=str(username).upper()259 connection=sqlite3.connect(database_name)260 cursor=connection.cursor()261 query="DELETE FROM USER_ADMIN_MAPPING WHERE USERNAME = '" + str(username) +"'"262 cursor.execute(query)263 connection.commit()264 connection.close()265def del_admin_USER_ADMIN_MAPPING(username):266 username=str(username).upper()267 connection=sqlite3.connect(database_name)268 cursor=connection.cursor()269 query="DELETE FROM USER_ADMIN_MAPPING WHERE ADMIN = '" + str(username) +"'"270 cursor.execute(query)271 connection.commit()272 connection.close()273#USER_ADMIN_MAPPING END274#ADMIN_KEY Start275def insert_into_ADMIN_KEY(adminkey):276 adminkey=str(adminkey).upper()277 connection=sqlite3.connect(database_name)278 cursor=connection.cursor()279 insert_query = "INSERT INTO ADMIN_KEY(ADMINKEY) VALUES (?)"280 if(len(get_ADMIN_KEY(adminkey))):281 {}282 else:283 cursor.execute(insert_query,(adminkey,))284 connection.commit()285 connection.close()286def content_ADMIN_KEY():287 list_users=list()288 connection=sqlite3.connect(database_name)289 cursor=connection.cursor()290 query='SELECT * FROM ADMIN_KEY'291 for row in cursor.execute(query):292 list_users.append(row)293 connection.commit()294 connection.close()295 return list_users296def get_ADMIN_KEY(adminkey):297 adminkey=str(adminkey).upper()298 list_users=list()299 connection=sqlite3.connect(database_name)300 cursor=connection.cursor()301 query="SELECT * FROM ADMIN_KEY WHERE ADMINKEY = '" + str(adminkey) +"'"302 for row in cursor.execute(query):303 list_users.append(row)304 connection.commit()305 connection.close()306 return list_users307def del_ADMIN_KEY(adminkey):308 adminkey=str(adminkey).upper()309 connection=sqlite3.connect(database_name)310 cursor=connection.cursor()311 query="DELETE FROM ADMIN_KEY WHERE ADMINKEY = '" + str(adminkey) +"'"312 cursor.execute(query)313 connection.commit()314 connection.close()315#ADMIN_KEY END316#ROUTER_PASSWORD Start317def get_ROUTER_PASSWORD(router_id):318 router_id=str(router_id).upper()319 list_users=list()320 connection=sqlite3.connect(database_name)321 cursor=connection.cursor()322 query="SELECT PASSWORD FROM ROUTER_PASSWORD WHERE ROUTER_ID = '" + str(router_id) +"'"323 for row in cursor.execute(query):324 list_users.append(row)325 connection.commit()326 connection.close()327 return list_users328def content_ROUTER_PASSWORD():329 list_users=list()330 connection=sqlite3.connect(database_name)331 cursor=connection.cursor()332 query="SELECT * FROM ROUTER_PASSWORD"333 for row in cursor.execute(query):334 list_users.append(row)335 connection.commit()336 connection.close()337 return list_users338def insert_into_ROUTER_PASSWORD(router_id,password):339 router_id=str(router_id).upper()340 password=str(password).upper()341 x=len(get_ROUTER_PASSWORD(router_id))342 if(x==0):343 connection=sqlite3.connect(database_name)344 cursor=connection.cursor()345 insert_query = "INSERT INTO ROUTER_PASSWORD(ROUTER_ID,PASSWORD) VALUES (?,?)"346 cursor.execute(insert_query,(router_id,password))347 connection.commit()348 connection.close()349 else:350 connection=sqlite3.connect(database_name)351 cursor=connection.cursor()352 update_query="UPDATE ROUTER_PASSWORD SET PASSWORD='"+str(password)+"' WHERE ROUTER_ID='"+str(router_id)+"'"353 cursor.execute(update_query)354 connection.commit()355 connection.close()356#ROUTER_PASSWORD End357#ROUTER Start358def insert_into_ROUTER(router_id,devices,state):359 router_id=str(router_id).upper()360 devices=str(devices).upper()361 state=str(state).upper()362 x=find_router_devices(router_id,devices)363 if(x==0):364 connection=sqlite3.connect(database_name)365 cursor=connection.cursor()366 insert_query = "INSERT INTO ROUTER(ROUTER_ID,DEVICES,STATE) VALUES (?,?,?)"367 cursor.execute(insert_query,(router_id,devices,state))368 connection.commit()369 connection.close()370 else:371 connection=sqlite3.connect(database_name)372 cursor=connection.cursor()373 insert_query = "UPDATE ROUTER SET STATE='"+str(state)+"' WHERE ROUTER_ID = '" + str(router_id) +"' AND DEVICES ='"+str(devices)+"'"374 cursor.execute(insert_query)375 connection.commit()376 connection.close()377def content_ROUTER():378 list_users=list()379 connection=sqlite3.connect(database_name)380 cursor=connection.cursor()381 query='SELECT * FROM ROUTER'382 for row in cursor.execute(query):383 list_users.append(row)384 connection.commit()385 connection.close()386 return list_users387def get_ROUTER(ROUTER_ID):388 ROUTER_ID=str(ROUTER_ID).upper()389 list_users=list()390 connection=sqlite3.connect(database_name_status)391 cursor=connection.cursor()392 query="SELECT ROUTER_ID,GROUP_ID,DEVICES,STATE FROM ROUTER WHERE ROUTER_ID = '" + str(ROUTER_ID) +"'"393 for row in cursor.execute(query):394 list_users.append(row)395 connection.commit()396 connection.close()397 return list_users398def del_user_ROUTER(username):399 username=str(username).upper()400 connection=sqlite3.connect(database_name)401 cursor=connection.cursor()402 query="DELETE FROM ROUTER WHERE USERNAME = '" + str(username) +"'"403 cursor.execute(query)404 connection.commit()405 connection.close()406def get_ROUTER_2(ROUTER_ID,devices):407 ROUTER_ID=str(ROUTER_ID).upper()408 devices=str(device).upper()409 list_users=list()410 connection=sqlite3.connect(database_name)411 cursor=connection.cursor()412 query="SELECT STATE FROM ROUTER WHERE ROUTER_ID = '" + str(ROUTER_ID) +"'AND DEVICES='"+str(devices)+"'"413 for row in cursor.execute(query):414 list_users.append(row)415 connection.commit()416 connection.close()417 return list_users418def update_ROUTER(router_id,devices,state):419 router_id=str(router_id).upper()420 devices=str(devices).upper()421 state=str(state).upper()422 connection=sqlite3.connect(database_name)423 cursor=connection.cursor()424 query="UPDATE ROUTER SET STATE = '"+str(state)+"' WHERE ROUTER_ID='"+str(router_id)+"' AND DEVICES = '"+str(devices)+"'"425 cursor.execute(query)426 connection.commit()427 connection.close()428def find_router_devices(router_id,device):429 router_id=str(router_id).upper()430 device=str(device).upper()431 list_users=list()432 connection=sqlite3.connect(database_name)433 cursor=connection.cursor()434 query="SELECT * FROM ROUTER WHERE ROUTER_ID = '" + str(router_id) +"' AND DEVICES ='"+str(device)+"'"435 for row in cursor.execute(query):436 list_users.append(row)437 if (len(list_users)==0):438 return 0439 else:440 return(len(list_users))441 connection.commit()442 connection.close()443 return list_users444#ROUTER End445#####Table Status446import sqlite3447database_name_status="Status.db"448def create_table_status():449 connection=sqlite3.connect(database_name_status)450 cursor=connection.cursor()451 create_table_ROUTER='CREATE TABLE IF NOT EXISTS [ROUTER]([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,ROUTER_ID TEXT,GROUP_ID TEXT,DEVICES TEXT,STATE TEXT,LAST_UPDATED_TIME DATETIME,CURRENT_TIME_N DATETIME,DIFFERENCE_TIME DATETIME)'452 cursor.execute(create_table_ROUTER)453 connection.commit()454 connection.close()455def insert_into_status(router_id,group_id,devices,state):456 router_id=str(router_id).upper()457 group_id=str(group_id).upper()458 devices=str(devices).upper()459 state=str(state).upper()460 x=status_find_duplicates(router_id,group_id,devices)461 if(x==0):462 connection=sqlite3.connect(database_name_status)463 cursor=connection.cursor()464 insert_query = "INSERT INTO ROUTER(ROUTER_ID,GROUP_ID,DEVICES,STATE) VALUES (?,?,?,?)"465 cursor.execute(insert_query,(router_id,group_id,devices,state))466 insert_query = "UPDATE ROUTER SET LAST_UPDATED_TIME= datetime() WHERE ROUTER_ID = '" + str(router_id) +"'AND GROUP_ID ='"+str(group_id)+"'AND DEVICES ='"+str(devices)+"'"467 cursor.execute(insert_query)468 connection.commit()469 connection.close()470 else:471 connection=sqlite3.connect(database_name_status)472 cursor=connection.cursor()473 insert_query = "UPDATE ROUTER SET STATE='"+str(state)+"'WHERE ROUTER_ID ='" + str(router_id) +"' AND GROUP_ID='"+str(group_id)+ "'AND DEVICES ='"+str(devices)+"'"474 cursor.execute(insert_query)475 insert_query = "UPDATE ROUTER SET LAST_UPDATED_TIME= datetime() WHERE ROUTER_ID ='" + str(router_id) +"' AND GROUP_ID='"+str(group_id)+ "'AND DEVICES ='"+str(devices)+"'"476 cursor.execute(insert_query)477 connection.commit()478 connection.close()479 set_current_time()480def status_find_duplicates(router_id,group_id,devices):481 router_id=str(router_id).upper()482 group_id=str(group_id).upper()483 devices=str(devices).upper()484 list_users=list()485 connection=sqlite3.connect(database_name_status)486 cursor=connection.cursor()487 query="SELECT * FROM ROUTER WHERE ROUTER_ID = '" + str(router_id) +"'AND GROUP_ID ='"+str(group_id)+"'AND DEVICES ='"+str(devices)+"'"488 for row in cursor.execute(query):489 list_users.append(row)490 if (len(list_users)==0):491 return 0492 else:493 return(len(list_users))494 connection.commit()495 connection.close()496 return list_users497def remove_from_status(router_id,group_id,devices):498 router_id=str(router_id).upper()499 devices=str(devices).upper()500 connection=sqlite3.connect(database_name_status)501 cursor=connection.cursor()502 query="DELETE FROM ROUTER WHERE ROUTER_ID = '" + str(router_id) +"'AND GROUP_ID ='"+str(group_id)+"'AND DEVICES ='"+str(devices)+"'"503 cursor.execute(query)504 connection.commit()505 connection.close()506def set_current_time():507 connection=sqlite3.connect(database_name_status)508 cursor=connection.cursor()509 query="UPDATE ROUTER SET CURRENT_TIME_N=datetime()"510 cursor.execute(query)511 connection.commit()512 connection.close()513 set_difference_time()514def set_difference_time():515 connection=sqlite3.connect(database_name_status)516 cursor=connection.cursor()517 query="UPDATE ROUTER SET DIFFERENCE_TIME=(julianday(CURRENT_TIME_N) - julianday(LAST_UPDATED_TIME)) * 86400.0"518 cursor.execute(query)519 connection.commit()520 connection.close()521def remove_device_status(threshold):522 connection=sqlite3.connect(database_name_status)523 cursor=connection.cursor()524 query="DELETE FROM ROUTER WHERE DIFFERENCE_TIME >'"+str(threshold)+"'"525 cursor.execute(query)526 connection.commit()527 connection.close()528def status_sort():529 connection=sqlite3.connect(database_name_status)530 cursor=connection.cursor()531 query="SELECT * FROM ROUTER ORDER BY GROUP_ID ASC"532 cursor.execute(query)533 connection.commit()534 connection.close()535##End536##Table Command537def create_table_command():538 connection=sqlite3.connect(database_name_command)539 cursor=connection.cursor()540 create_table_ROUTER='CREATE TABLE IF NOT EXISTS [COMMAND]([ID] INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,ROUTER_ID TEXT,GROUP_ID TEXT,DEVICES TEXT,STATE TEXT)'541 cursor.execute(create_table_ROUTER)542 cursor.close()543 connection.commit()544 connection.close()545def insert_command(router_id,group_id,devices_id,state):546 router_id=str(router_id).upper()547 group_id=str(group_id).upper()548 devices_id=str(devices_id).upper()549 state=str(state).upper()550 connection=sqlite3.connect(database_name_command)551 cursor=connection.cursor()552 insert_query='INSERT INTO COMMAND(ROUTER_ID,GROUP_ID,DEVICES,STATE) VALUES(?,?,?,?)'553 cursor.execute(insert_query,(router_id,group_id,devices_id,state))554 cursor.close()555 connection.commit()556 connection.close()557def table_queue():558 connection=sqlite3.connect(database_name_command)559 cursor=connection.cursor()560 queue_querry='SELECT ROUTER_ID,GROUP_ID,DEVICES,STATE FROM COMMAND'561 list_commands=list()562 for row in cursor.execute(queue_querry):563 list_commands.append(row)564 cursor.close()565 connection.commit()566 connection.close()567 message=list_commands568 list_commands_formatted=list()569 for i in message:570 list_commands_formatted.append(i[0]+"/"+i[1]+"/"+i[2]+"/"+i[3])571 connection=sqlite3.connect(database_name_command)572 cursor=connection.cursor()573 clear_query='DELETE FROM COMMAND'574 cursor.execute(clear_query)575 connection.commit()576 connection.close()577 return list_commands_formatted...
get_info.py
Source:get_info.py
1import requests2import time3import csv4import numpy as np5from argparse import ArgumentParser6from tqdm import tqdm7def get_likes(list_users, type_content, owner_id, count, token, quiet):8 print(f'Get likes on {type_content}s:')9 base_url = 'https://api.vk.com/method/'10 if type_content == 'post':11 resp = requests.post(base_url + 'wall.get', data={12 'owner_id': owner_id, 'count': count, 'v': '5.80', 13 'access_token': token}).json()14 counter_name = 'posts_likes'15 elif type_content == 'photo':16 resp = requests.post(base_url + 'photos.get', data={17 'owner_id': owner_id, 'album_id': 'profile', 'count': count, 18 'rev': '1', 'v': '5.80', 'access_token': token}).json()19 counter_name = 'photos_likes'20 21 try:22 resp = resp['response']23 except KeyError as e:24 print(owner_id, ':', resp['error'])25 items = [item['id'] for item in resp['items']]26 if not quiet:27 print(f'{type_content}: {items}')28 for item_id in tqdm(items):29 resp_item = requests.post(base_url + 'likes.getList', data={30 'type': type_content, 'owner_id': owner_id, 'item_id': item_id, 31 'filter': 'likes', 'v': '5.80', 'access_token': token}).json()32 if not quiet:33 print(f'{type_content}_{item_id}: {resp_item}')34 for user in resp_item['response']['items']:35 if user not in list_users:36 list_users[user] = {'posts_likes': 0,37 'photos_likes': 0,38 'gifts': 0}39 list_users[user][counter_name] = 140 else:41 list_users[user][counter_name] += 142 time.sleep(1)43 return list_users44def get_gifts(list_users, owner_id, token, quiet):45 print(f'Get gifts:')46 base_url = 'https://api.vk.com/method/'47 resp = requests.post(base_url + 'gifts.get', data={'user_id': owner_id, 'v': '5.80', 'access_token': token}).json()48 if not quiet:49 print(f'Gifts response: {resp}')50 for gift in tqdm(resp['response']['items']):51 user = gift['from_id']52 message = gift['message']53 if len(message) != 0 and user > 0:54 with open('gift_message.txt', 'a') as f:55 new_mes = f'from {user} to {owner_id}: {message}\n'56 f.write(new_mes)57 if user > 0:58 if user not in list_users:59 list_users[user] = {'posts_likes': 0,60 'photos_likes': 0,61 'gifts': 1}62 else:63 list_users[user]['gifts'] += 164 return list_users65def get_likes_by_user(list_users, type_content, user_id, count, token, quiet):66 print(f'Get likes on {type_content}s by id{user_id}:')67 base_url = 'https://api.vk.com/method/'68 for owner_id in tqdm(list_users):69 if not quiet:70 print(f'{type_content}s by {owner_id}')71 if type_content == 'post':72 resp = requests.post(base_url + 'wall.get', data={73 'owner_id': owner_id, 'count': count, 'v': '5.80', 74 'access_token': token}).json()75 counter_name = 'posts_likes_by_user'76 elif type_content == 'photo':77 resp = requests.post(base_url + 'photos.get', data={78 'owner_id': owner_id, 'album_id': 'profile', 'count': count, 79 'rev': '1', 'v': '5.80', 'access_token': token}).json()80 counter_name = 'photos_likes_by_user'81 list_users[owner_id][counter_name] = 082 try:83 resp = resp['response']84 except KeyError as e:85 if not quiet:86 print(owner_id, ':', resp['error'])87 continue88 89 items = [item['id'] for item in resp['items']]90 if not quiet:91 print(f'{type_content} of {owner_id}: {items}')92 for item_id in items:93 resp_item = requests.post(base_url + 'likes.isLiked', data={94 'user_id': user_id, 'type': type_content, 'owner_id': owner_id,95 'item_id': item_id, 'v': '5.80', 'access_token': token}).json()96 if not quiet:97 print(f'likes_by_user {owner_id}: {resp_item}')98 if resp_item['response']['liked']:99 list_users[owner_id][counter_name] += 1100 time.sleep(1)101 #time.sleep(1)102 return list_users103def get_gifts_by_user(list_users, user_id, token, quiet):104 print(f'Get gifts by id{user_id}:')105 base_url = 'https://api.vk.com/method/'106 for owner_id in tqdm(list_users):107 list_users[owner_id]['gifts_by_user'] = 0108 resp = requests.post(base_url + 'gifts.get', data={109 'user_id': owner_id, 'v': '5.80', 'access_token': token}).json()110 try:111 resp = resp['response']112 except KeyError as e:113 if not quiet:114 print(owner_id, ':', resp['error'])115 continue116 117 if not quiet:118 print(f'Gifts response: {resp}')119 for gift in resp['items']:120 id_from = gift['from_id']121 message = gift['message']122 if len(message) != 0 and id_from > 0:123 with open('gift_message.txt', 'a') as f:124 new_mes = f'from {id_from} to {owner_id}: {message}\n'125 f.write(new_mes)126 if id_from == user_id:127 list_users[owner_id]['gifts_by_user'] += 1128 time.sleep(1)129 return list_users130def get_users_info(list_users, token, quiet):131 base_url = 'https://api.vk.com/method/'132 resp = requests.post(base_url + 'users.get', data={'user_ids': ','.join([str(i) for i in list_users.keys()]), 'fields': 'sex,common_count,friend_status,relation', 'v': '5.89', 'access_token': token}).json()133 if not quiet:134 print(f'Users response: {resp}')135 for user in resp['response']:136 user_id = user['id']137 list_users[user_id]['name'] = ' '.join([user['first_name'], user['last_name']])138 if 'is_closed' in user:139 list_users[user_id]['is_closed'] = user['is_closed']140 else:141 list_users[user_id]['is_closed'] = 1142 if 'sex' in user:143 list_users[user_id]['sex'] = user['sex']144 else:145 list_users[user_id]['sex'] = 0146 if 'friend_status' in user:147 list_users[user_id]['friend_status'] = user['friend_status']148 else:149 list_users[user_id]['friend_status'] = 0150 if 'common_count' in user:151 list_users[user_id]['common_count'] = user['common_count']152 else:153 list_users[user_id]['common_count'] = 0154 if 'relation' in user:155 list_users[user_id]['relation'] = user['relation']156 else:157 list_users[user_id]['relation'] = 0158 return list_users159 160def to_csv(list_users, user_id):161 file_name = 'info_id' + str(user_id) + '.csv'162 with open(file_name, 'w') as f:163 fieldnames = ['id', 'name', 'sex', 'is_closed', 'posts_likes', 'photos_likes', 'gifts', 'posts_likes_by_user', 'photos_likes_by_user', 'gifts_by_user', 'friend_status', 'common_count', 'relation']164 writer = csv.DictWriter(f, fieldnames=fieldnames)165 writer.writeheader()166 for user, item in list_users.items():167 item['id'] = user168 writer.writerow(item) 169def parse_args():170 parser = ArgumentParser()171 parser.add_argument('--id',172 default=48245655,173 help='user id you want to get information')174 parser.add_argument('--count', '-c',175 default='5',176 help='number of photos/posts to be examined')177 parser.add_argument('--quiet', '-q',178 help='do not display the text on command line',179 action='store_true')180 return parser.parse_args() 181if __name__ == '__main__':182 token = # insert your token here183 args = parse_args()184 my_id = args.id185 count = args.count186 quiet = args.quiet187 188 related_users = {}189 # get users and number of likes on posts of the user:190 related_users = get_likes(list_users=related_users, 191 type_content='post', 192 owner_id=my_id, 193 count=count, 194 token=token, 195 quiet=quiet)196 # get users and number of likes on photos of the user:197 related_users = get_likes(list_users=related_users, 198 type_content='photo', 199 owner_id=my_id, 200 count=count, 201 token=token, 202 quiet=quiet)203 # get users and number of gifts to the user:204 related_users = get_gifts(list_users=related_users, 205 owner_id=my_id, 206 token=token, 207 quiet=quiet)208 # get number of likes on posts of each user:209 related_users = get_likes_by_user(list_users=related_users, 210 type_content='post',211 user_id=my_id, 212 count=count, 213 token=token, 214 quiet=quiet)215 # get number of likes on photos of each user:216 related_users = get_likes_by_user(list_users=related_users, 217 type_content='photo',218 user_id=my_id, 219 count=count, 220 token=token, 221 quiet=quiet)222 # get number of gifts to each user:223 related_users = get_gifts_by_user(list_users=related_users, 224 user_id=my_id,225 token=token, 226 quiet=quiet) 227 228 related_users = get_users_info(list_users=related_users, 229 token=token, 230 quiet=quiet)231 if not quiet:232 print(f'\nRelated: {related_users}\n')233 234 to_csv(related_users, my_id)235 ...
test_models_user.py
Source:test_models_user.py
1import pytest2def test_cant_read_user_password(list_users):3 for u in list_users:4 with pytest.raises(AttributeError):5 u.password6def test_password_setter(list_users):7 u1, *_ = list_users8 u1.password = 'pa$$word!'9 assert u1.password_hash is not None10def test_password_verification__correct(list_users):11 u1, u2, _ = list_users12 u1.password = 'pa$$word!'13 u2.password = 'secr3t'14 assert u1.verify_password('pa$$word!')15 assert u2.verify_password('secr3t')16def test_password_verification__incorrect(list_users):17 u1, *_ = list_users18 u1.password = 'pa$$word!'19 assert not u1.verify_password('password!')20def test_password_salts_are_random(list_users):21 u1, u2, _ = list_users22 u1.password = 'cat'23 u2.password = 'cat'24 assert u1.password_hash != u2.password_hash25def test_confirm_verify_token__valid(app, list_users):26 u1, *_ = list_users27 token = u1.generate_verify_token()28 assert u1.confirm(token)29def test_confirm_verify_token__invalid(app, list_users):30 u1, u2, _ = list_users31 u1_token = u1.generate_verify_token()32 assert not u2.confirm(u1_token)33def test_confirm_exception(app, list_users):34 u1, *_ = list_users35 assert not u1.confirm(None)36def test_user_repr(app, list_users):37 u1, *_ = list_users...
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 tempest 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?
