KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use convert_query method in autotest

Best Python code snippet using autotest_python
DB.py
Source:DB.py
...48            c.execute(create_power_status)49            conn.commit()50        except Exception as e:51            print("Error creating database: " + str(e))52def convert_query(query, postgres_suffix=None):53    if DatabaseType == DBType.Postgres:54        query = query.replace("?", "%s")55        if postgres_suffix:56            query += postgres_suffix57    return query58def get_last_insert_row_id(cursor):59    if DatabaseType == DBType.Postgres:60        return cursor.fetchone()[0]61    return cursor.lastrowid62def setupserver(PayloadCommsHost, EncKey, DomainFrontHeader, DefaultSleep, KillDate, GET_404_Response, PoshProjectDirectory, QuickCommand, DownloadURI, ProxyURL, ProxyUser, ProxyPass, URLS, SocksURLS, Insecure, UserAgent, Referrer, Pushover_APIToken, Pushover_APIUser, Slack_UserID, Slack_Channel, Slack_BotToken, EnableNotifications):63    c = conn.cursor()64    command = convert_query("INSERT INTO C2Server (PayloadCommsHost,EncKey,DomainFrontHeader,DefaultSleep,KillDate,GET_404_Response,PoshProjectDirectory,QuickCommand,DownloadURI,ProxyURL,ProxyUser,ProxyPass,URLS,SocksURLS,Insecure,UserAgent,Referrer,Pushover_APIToken,Pushover_APIUser,Slack_UserID,Slack_Channel,Slack_BotToken,EnableNotifications) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")65    c.execute(command, (PayloadCommsHost, EncKey, DomainFrontHeader, DefaultSleep, KillDate, GET_404_Response, PoshProjectDirectory, QuickCommand, DownloadURI, ProxyURL, ProxyUser, ProxyPass, URLS, SocksURLS, Insecure, UserAgent, Referrer, Pushover_APIToken, Pushover_APIUser, Slack_UserID, Slack_Channel, Slack_BotToken, EnableNotifications))66    conn.commit()67def get_c2server_all():68    c = get_conn().cursor()69    c.execute("SELECT * FROM C2Server")70    result = c.fetchone()71    return C2(result[1], result[2], result[3], result[4], result[5], result[6], result[7], result[8], result[9],72              result[10], result[11], result[12], result[13], result[14], result[15], result[16], result[17],73              result[18], result[19], result[20], result[21], result[22], result[23])74def get_implants_all():75    c = get_conn().cursor()76    c.execute("SELECT * FROM Implants")77    results = c.fetchall()78    implants = []79    for result in results:80        implants.append(Implant(result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7], result[8],81                                result[9], result[10], result[11], result[12], result[13], result[14], result[15], result[16], result[17]))82    return implants83def get_newtasks_all():84    c = get_conn().cursor()85    c.execute("SELECT * FROM NewTasks")86    results = c.fetchall()87    tasks = []88    for result in results:89        tasks.append(NewTask(result[0], result[1], result[2], result[3]))90    return tasks91def new_urldetails(Name, URL, HostHeader, ProxyURL, ProxyUsername, ProxyPassword, CredentialExpiry):92    c = get_conn().cursor()93    command = convert_query("INSERT INTO URLs (Name, URL, HostHeader, ProxyURL, ProxyUsername, ProxyPassword, CredentialExpiry) VALUES (?, ?, ?, ?, ?, ?, ?)", " RETURNING URLID")94    c.execute(command, (Name, URL, HostHeader, ProxyURL, ProxyUsername, ProxyPassword, CredentialExpiry))95    get_conn().commit()96    return get_last_insert_row_id(c)97def drop_newtasks():98    c = get_conn().cursor()99    c.execute("DELETE FROM NewTasks ")100    get_conn().commit()101def new_task(task, user, randomuri):102    c = get_conn().cursor()103    command = convert_query("INSERT INTO NewTasks (RandomURI, Command, \"User\") VALUES (?, ?, ?)")104    c.execute(command, (randomuri, task, user))105    get_conn().commit()106def get_implants():107    c = get_conn().cursor()108    c.execute("SELECT * FROM Implants WHERE Alive='Yes' ORDER BY implantid")109    results = c.fetchall()110    implants = []111    for result in results:112        implants.append(Implant(result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7], result[8],113                                result[9], result[10], result[11], result[12], result[13], result[14], result[15], result[16], result[17]))114    return implants115def get_implanttype(randomuri):116    c = get_conn().cursor()117    query = convert_query("SELECT Pivot FROM Implants WHERE RandomURI=?")118    c.execute(query, (randomuri,))119    result = str(c.fetchone()[0])120    if result:121        return result122    else:123        return None124def get_implantdetails(randomuri):125    c = get_conn().cursor()126    query = convert_query("SELECT * FROM Implants WHERE RandomURI=?")127    c.execute(query, (randomuri,))128    result = c.fetchone()129    if result:130        return Implant(result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7], result[8],131                       result[9], result[10], result[11], result[12], result[13], result[14], result[15], result[16],result[17])132    else:133        return None134def get_randomuri(implant_id):135    c = get_conn().cursor()136    try:137        implant_id = int(implant_id)138        query = convert_query("SELECT RandomURI FROM Implants WHERE ImplantID=?")139        c.execute(query, (implant_id,))140        result = str(c.fetchone()[0])141    except ValueError:142        return None143    if result:144        return result145    else:146        return None147def add_autorun(Task):148    c = get_conn().cursor()149    command = convert_query("INSERT INTO AutoRuns (Task) VALUES (?)")150    c.execute(command, (Task,))151    get_conn().commit()152def update_sleep(sleep, randomuri):153    c = get_conn().cursor()154    command = convert_query("UPDATE Implants SET Sleep=? WHERE RandomURI=?")155    c.execute(command, (sleep, randomuri))156    get_conn().commit()157def update_label(label, randomuri):158    c = get_conn().cursor()159    command = convert_query("UPDATE Implants SET Label=? WHERE RandomURI=?")160    c.execute(command, (label, randomuri))161    get_conn().commit()162def update_mods(modules, randomuri):163    c = get_conn().cursor()164    command = convert_query("UPDATE Implants SET ModsLoaded=? WHERE RandomURI=?")165    c.execute(command, (modules, randomuri))166    get_conn().commit()167def kill_implant(randomuri):168    c = get_conn().cursor()169    command = convert_query("UPDATE Implants SET Alive='No' WHERE RandomURI=?")170    c.execute(command, (randomuri,))171    get_conn().commit()172def unhide_implant(randomuri):173    c = get_conn().cursor()174    command = convert_query("UPDATE Implants SET Alive='Yes' WHERE RandomURI=?")175    c.execute(command, (randomuri,))176    get_conn().commit()177def hide_implant(randomuri):178    c = get_conn().cursor()179    command = convert_query("UPDATE Implants SET Alive='No' WHERE RandomURI=?")180    c.execute(command, (randomuri,))181    get_conn().commit()182def select_mods(randomuri):183    c = get_conn().cursor()184    query = convert_query("SELECT ModsLoaded FROM Implants WHERE RandomURI=?")185    c.execute(query, (randomuri,))186    result = str(c.fetchone()[0])187    if result:188        return result189    else:190        return None191def select_item(column, table):192    c = get_conn().cursor()193    c.execute(f"SELECT {column} FROM {table}")194    result = str(c.fetchone()[0])195    if result:196        return result197    else:198        return None199def del_newtasks(TaskID):200    c = get_conn().cursor()201    command = convert_query("DELETE FROM NewTasks WHERE TaskID=?")202    c.execute(command, (TaskID,))203    get_conn().commit()204def del_autorun(TaskID):205    c = get_conn().cursor()206    command = convert_query("DELETE FROM AutoRuns WHERE TaskID=?")207    c.execute(command, (TaskID,))208    get_conn().commit()209def del_autoruns():210    c = get_conn().cursor()211    c.execute("DELETE FROM AutoRuns ")212    get_conn().commit()213def update_implant_lastseen(time, randomuri):214    c = get_conn().cursor()215    command = convert_query("UPDATE Implants SET LastSeen=? WHERE RandomURI=?")216    c.execute(command, (time, randomuri))217    get_conn().commit()218def new_implant(RandomURI, URLID, User, Hostname, IpAddress, Key, FirstSeen, LastSeen, PID, ProcName, Arch, Domain, Alive, Sleep, ModsLoaded, Pivot, Label):219    c = get_conn().cursor()220    command = convert_query("INSERT INTO Implants (RandomURI, URLID, \"User\", Hostname, IpAddress, Key, FirstSeen, LastSeen, PID, ProcName, Arch, Domain, Alive, Sleep, ModsLoaded, Pivot, Label) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", " RETURNING ImplantID")221    c.execute(command, (RandomURI, URLID, User, Hostname, IpAddress, Key, FirstSeen, LastSeen, PID, ProcName, Arch, Domain, Alive, Sleep, ModsLoaded, Pivot, Label))222    get_conn().commit()223    return get_last_insert_row_id(c)224def insert_task(randomuri, command, user):225    now = datetime.now(timezone.utc)226    sent_time = now.strftime("%Y-%m-%d %H:%M:%S")227    implantId = get_implantbyrandomuri(randomuri).ImplantID228    c = get_conn().cursor()229    if user is None:230        user = ""231    query = convert_query("INSERT INTO Tasks (RandomURI, Command, Output, \"User\", SentTime, CompletedTime, ImplantID) VALUES (?, ?, ?, ?, ?, ?, ?)", " RETURNING TaskID")232    c.execute(query, (randomuri, command, "", user, sent_time, "", implantId))233    get_conn().commit()234    return get_last_insert_row_id(c)235def update_task(taskId, output):236    now = datetime.now(timezone.utc)237    completedTime = now.strftime("%Y-%m-%d %H:%M:%S")238    c = get_conn().cursor()239    command = convert_query("UPDATE Tasks SET Output=?, CompletedTime=? WHERE TaskID=?")240    c.execute(command, (output, completedTime, taskId))241    get_conn().commit()242    return taskId243def get_task_owner(taskId):244    c = get_conn().cursor()245    query = convert_query("SELECT \"User\" FROM Tasks WHERE TaskID=?")246    c.execute(query, (taskId,))247    result = c.fetchone()248    if result and result[0] != "":249        return result[0]250    else:251        return None252def update_item(column, table, value, wherecolumn=None, where=None):253    c = get_conn().cursor()254    if wherecolumn is None:255        query = convert_query(f"UPDATE {table} SET {column}=?")256        c.execute(query, (value,))257    else:258        query = convert_query(f"UPDATE {table} SET {column}=? WHERE {wherecolumn}=?")259        c.execute(query, (value, where))260    get_conn().commit()261def get_implantbyid(implantId):262    try:263        implantId = int(implantId)264    except ValueError:265        return None266    c = get_conn().cursor()267    query = convert_query("SELECT * FROM Implants WHERE ImplantID=?")268    c.execute(query, (implantId,))269    result = c.fetchone()270    if result:271        return Implant(result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7], result[8],272                       result[9], result[10], result[11], result[12], result[13], result[14], result[15], result[16],result[17])273    else:274        return None275def get_implantbyrandomuri(RandomURI):276    c = get_conn().cursor()277    query = convert_query("SELECT * FROM Implants WHERE RandomURI=?")278    c.execute(query, (RandomURI,))279    result = c.fetchone()280    if result:281        return Implant(result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7], result[8],282                       result[9], result[10], result[11], result[12], result[13], result[14], result[15], result[16],result[17])283    else:284        return None285def get_alldata(table):286    pd.set_option('display.max_colwidth', None)287    pd.options.mode.chained_assignment = None288    return pd.read_sql_query(f"SELECT * FROM {table}", get_conn())289def get_html_report_data(table_name):290    query_string = ""291    if (table_name == "Tasks"):292        query_string = "SELECT t.TaskID, i.Domain || '\\' || i.\"User\" || ' @ ' || i.Hostname AS Context, t.Command, t.Output, t.\"User\", t.SentTime, t.CompletedTime, t.ImplantID FROM Tasks t INNER JOIN Implants i USING(ImplantID)"293    elif (table_name == "C2Server"):294        query_string = "SELECT * FROM C2Server"295    elif (table_name == "Creds"):296        query_string = "SELECT * FROM Creds"297    elif (table_name == "Implants"):298        query_string = "SELECT ImplantID, Domain || '\\' || \"User\" || ' @ ' || Hostname AS Context, URLID, \"User\", Hostname, IpAddress, Key, FirstSeen, LastSeen, PID, ProcName, Arch, Domain, Alive, Sleep, ModsLoaded, Pivot, Label FROM Implants"299    elif (table_name == "URLs"):300        query_string = "SELECT * FROM URLs"301    elif (table_name == "OpSec_Entry"):302        query_string = "SELECT * FROM OpSec_Entry"303    if (query_string == ""):304        return None305    c = get_conn().cursor()306    c.execute(query_string)307    result = c.fetchall()308    if result:309        return result310    else:311        return None312def get_tasks():313    c = get_conn().cursor()314    c.execute("SELECT * FROM Tasks")315    result = c.fetchall()316    if result:317        return result318    else:319        return None320def get_tasksbyid(implantId):321    c = get_conn().cursor()322    query = convert_query("SELECT * FROM Tasks WHERE CompletedTaskID=?")323    c.execute(query, (implantId,))324    result = c.fetchone()325    if result:326        return result327    else:328        return None329def get_newtasksbyid(taskid):330    c = get_conn().cursor()331    query = convert_query("SELECT * FROM NewTasks WHERE TaskID=?")332    c.execute(query, (taskid,))333    result = c.fetchone()334    if result:335        return result336    else:337        return None338def get_baseenckey():339    c = get_conn().cursor()340    c.execute("SELECT EncKey FROM C2Server")341    result = str(c.fetchone()[0])342    if result:343        return result344    else:345        return None346def get_dfheader():347    c = get_conn().cursor()348    c.execute("SELECT DomainFrontHeader FROM C2Server")349    result = str(c.fetchone()[0])350    if result:351        return result352    else:353        return None354def get_cmd_from_task_id(taskId):355    c = get_conn().cursor()356    query = convert_query("SELECT Command FROM Tasks WHERE TaskId=?")357    c.execute(query, (taskId,))358    result = str(c.fetchone()[0])359    if result:360        return result361    else:362        return None363def get_notificationstatus():364    c = get_conn().cursor()365    c.execute("SELECT EnableNotifications FROM C2Server")366    result = str(c.fetchone()[0])367    if result:368        return result369    else:370        return None371def get_defaultuseragent():372    c = get_conn().cursor()373    c.execute("SELECT UserAgent FROM C2Server")374    result = str(c.fetchone()[0])375    if result:376        return result377    else:378        return None379def get_defaultbeacon():380    c = get_conn().cursor()381    c.execute("SELECT DefaultSleep FROM C2Server")382    result = str(c.fetchone()[0])383    if result:384        return result385    else:386        return None387def get_killdate():388    c = get_conn().cursor()389    c.execute("SELECT KillDate FROM C2Server")390    result = str(c.fetchone()[0])391    if result:392        return result393    else:394        return None395def get_sharpurls():396    c = get_conn().cursor()397    c.execute("SELECT SocksURLS FROM C2Server")398    result = str(c.fetchone()[0])399    if result:400        return result401    else:402        return None403def get_allurls():404    c = get_conn().cursor()405    c.execute("SELECT URLS FROM C2Server")406    c2Urls = str(c.fetchone()[0])407    c.execute("SELECT SocksURLS FROM C2Server")408    socksURLs = str(c.fetchone()[0])409    result = c2Urls + "," + socksURLs410    if result:411        return result412    else:413        return None414def get_url_by_id(id):415    c = get_conn().cursor()416    query = convert_query("SELECT * FROM URLs where URLID=?")417    c.execute(query, (id,))418    result = c.fetchone()419    return result420def get_default_url_id():421    c = get_conn().cursor()422    c.execute("SELECT * FROM URLs where Name='updated_host' ORDER BY URLID DESC LIMIT 1")423    result = c.fetchone()424    if result:425        return result426    else:427        c.execute("SELECT * FROM URLs where Name='default' ORDER BY URLID DESC LIMIT 1")428        return c.fetchone()429def get_beaconurl():430    c = get_conn().cursor()431    c.execute("SELECT URLS FROM C2Server")432    result = str(c.fetchone()[0])433    if result:434        url = result.split(",")435        return url[0]436    else:437        return None438def get_otherbeaconurls():439    c = get_conn().cursor()440    c.execute("SELECT URLS FROM C2Server")441    result = str(c.fetchone()[0])442    if result:443        return result444    else:445        return None446def get_newimplanturl():447    c = get_conn().cursor()448    c.execute("SELECT URLS FROM C2Server")449    result = str(c.fetchone()[0])450    if result:451        url = result.split(",")452        return "/" + url[0].replace('"', '')453    else:454        return None455def get_c2urls():456    c = get_conn().cursor()457    c.execute("SELECT * FROM URLs")458    result = c.fetchall()459    if result:460        return result461    else:462        return None463def get_autoruns():464    c = get_conn().cursor()465    c.execute("SELECT * FROM AutoRuns")466    result = c.fetchall()467    if result:468        return result469    else:470        return None471def get_autorun():472    c = get_conn().cursor()473    c.execute("SELECT * FROM AutoRuns")474    result = c.fetchall()475    autoruns = ""476    for autorun in result:477        autoruns += f"{autorun[0]}:{autorun[1]}\r\n"478    if autoruns:479        return autoruns480    else:481        return None482def get_pid(randomuri):483    c = get_conn().cursor()484    query = convert_query("SELECT PID FROM Implants WHERE RandomURI=?")485    c.execute(query, (randomuri,))486    result = c.fetchone()[0]487    if result:488        return result489    else:490        return None491def get_newtasks(randomuri):492    c = get_conn().cursor()493    query = convert_query("SELECT * FROM NewTasks WHERE RandomURI=?")494    c.execute(query, (randomuri,))495    result = c.fetchall()496    if result:497        return result498    else:499        return None500def get_keys():501    c = get_conn().cursor()502    result = c.execute("SELECT EncKey FROM C2Server")503    result = c.fetchall()504    if result:505        return result506    else:507        return None508def insert_cred(domain, username, password, hash):509    if check_if_cred_exists(domain, username, password, hash):510        return None511    c = get_conn().cursor()512    command = convert_query("INSERT INTO Creds (Domain, Username, Password, Hash) VALUES (?, ?, ?, ?)", " RETURNING CredID")513    c.execute(command, (domain, username, password, hash))514    get_conn().commit()515    return get_last_insert_row_id(c)516def check_if_cred_exists(domain, username, password, hash):517    c = get_conn().cursor()518    if not password:519        query = convert_query("SELECT * FROM Creds WHERE Domain=? AND Username=? AND Password IS NULL AND Hash=?")520        c.execute(query, (domain, username, hash))521    elif not hash:522        query = convert_query("SELECT * FROM Creds WHERE Domain=? AND Username=? AND Password=? AND Hash IS NULL")523        c.execute(query, (domain, username, password))524    else:525        query = convert_query("SELECT * FROM Creds WHERE Domain=? AND Username=? AND Password=? AND Hash=?")526        c.execute(query, (domain, username, password, hash))527    result = c.fetchall()528    if result:529        return True530    else:531        return False532def get_creds():533    c = get_conn().cursor()534    c.execute("SELECT * FROM Creds")535    result = c.fetchall()536    if result:537        return result538    else:539        return None540def get_creds_for_user(username):541    c = get_conn().cursor()542    query = convert_query("SELECT * FROM Creds WHERE Username=?")543    c.execute(query, (username,))544    result = c.fetchall()545    if result:546        return result547    else:548        return None549def get_cred_by_id(credId):550    c = get_conn().cursor()551    query = convert_query("SELECT * FROM Creds WHERE CredID=?")552    c.execute(query, (credId,))553    result = c.fetchone()554    if result:555        return result556    else:557        return None558def new_c2_message(message):559    now = datetime.now(timezone.utc)560    message = "\n%s%s: %s%s\n" % (Colours.BLUE, now.strftime("%Y-%m-%d %H:%M:%S"), message, Colours.END)561    c = get_conn().cursor()562    command = convert_query("INSERT INTO C2_Messages (Message,Read) VALUES (?,'No')", " RETURNING ID")563    c.execute(command, (message,))564    get_conn().commit()565    return get_last_insert_row_id(c)566def get_c2_messages():567    c = get_conn().cursor()568    c.execute("SELECT * FROM C2_Messages WHERE Read='No'")569    result = c.fetchall()570    if result:571        messages = []572        for item in result:573            command = convert_query("UPDATE C2_Messages Set Read='Yes' WHERE ID=?")574            c.execute(command, (item[0],))575            get_conn().commit()576            messages.append(item[1])577        return messages578    else:579        return None580def get_powerstatusbyrandomuri(randomuri):581    c = get_conn().cursor()582    query = convert_query("SELECT * FROM PowerStatus WHERE RandomURI=?")583    c.execute(query, (randomuri,))584    result = c.fetchone()585    if result:586        return result587    else:588        return None589def insert_powerstatus(randomuri, apmstatus, onacpower, charging, batterystatus, batterypercentleft, screenlocked, monitoron):590    now = datetime.now(timezone.utc)591    c = get_conn().cursor()592    now = datetime.now(timezone.utc)593    command = convert_query("INSERT INTO PowerStatus (RandomURI,APMStatus,OnACPower,Charging,BatteryStatus,BatteryPercentLeft,ScreenLocked,MonitorOn,LastUpdate) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")594    c.execute(command, (randomuri, apmstatus, onacpower, charging, batterystatus, batterypercentleft, screenlocked, monitoron, now.strftime("%Y-%m-%d %H:%M:%S")))595    get_conn().commit()596def insert_blankpowerstatus(randomuri):597    now = datetime.now(timezone.utc)598    c = get_conn().cursor()599    now = datetime.now(timezone.utc)600    command = convert_query("INSERT INTO PowerStatus (RandomURI,APMStatus,OnACPower,Charging,BatteryStatus,BatteryPercentLeft,ScreenLocked,MonitorOn,LastUpdate) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")601    c.execute(command, (randomuri, "", 255, 255, "", "", 0, 1, now.strftime("%Y-%m-%d %H:%M:%S")))602    get_conn().commit()603def update_powerstatus(randomuri, onacpower, charging, batterystatus, batterypercentleft):604    now = datetime.now(timezone.utc)605    c = get_conn().cursor()606    now = datetime.now(timezone.utc)607    command = convert_query("UPDATE PowerStatus SET OnACPower=?,Charging=?,BatteryStatus=?,BatteryPercentLeft=?,LastUpdate=? WHERE RandomURI=?")608    c.execute(command, (onacpower, charging, batterystatus, batterypercentleft, now.strftime("%Y-%m-%d %H:%M:%S"), randomuri))609    get_conn().commit()610def update_apmstatus(randomuri, apmstatus):611    now = datetime.now(timezone.utc)612    c = get_conn().cursor()613    now = datetime.now(timezone.utc)614    command = convert_query("UPDATE PowerStatus SET APMStatus=?, LastUpdate=? WHERE RandomURI=?")615    c.execute(command, (apmstatus, now.strftime("%Y-%m-%d %H:%M:%S"), randomuri))616    get_conn().commit()617def update_acstatus(randomuri, onacpower):618    now = datetime.now(timezone.utc)619    c = get_conn().cursor()620    now = datetime.now(timezone.utc)621    command = convert_query("UPDATE PowerStatus SET OnACPower=?, LastUpdate=? WHERE RandomURI=?")622    c.execute(command, (onacpower, now.strftime("%Y-%m-%d %H:%M:%S"), randomuri))623    get_conn().commit()624def update_screenlocked(randomuri, locked):625    now = datetime.now(timezone.utc)626    c = get_conn().cursor()627    now = datetime.now(timezone.utc)628    command = convert_query("UPDATE PowerStatus SET ScreenLocked=?, LastUpdate=? WHERE RandomURI=?")629    c.execute(command, (locked, now.strftime("%Y-%m-%d %H:%M:%S"), randomuri))630    get_conn().commit()631def update_monitoron(randomuri, monitoron):632    now = datetime.now(timezone.utc)633    c = get_conn().cursor()634    now = datetime.now(timezone.utc)635    command = convert_query("UPDATE PowerStatus SET MonitorOn=?, LastUpdate=? WHERE RandomURI=?")636    c.execute(command, (monitoron, now.strftime("%Y-%m-%d %H:%M:%S"), randomuri))637    get_conn().commit()638def enable_hosted_file(ID):639    c = get_conn().cursor()640    command = convert_query("UPDATE Hosted_Files SET Active='Yes' WHERE ID=?")641    c.execute(command, (ID,))642    get_conn().commit()643def del_hosted_file(ID):644    c = get_conn().cursor()645    command = convert_query("UPDATE Hosted_Files SET Active='No' WHERE ID=?")646    c.execute(command, (ID,))647    get_conn().commit()648def insert_hosted_file(URI, FilePath, ContentType, Base64, Active):649    c = get_conn().cursor()650    command = convert_query("INSERT INTO Hosted_Files (URI, FilePath, ContentType, Base64, Active) VALUES (?, ?, ?, ?, ?)")651    c.execute(command, (URI, FilePath, ContentType, Base64, Active))652    get_conn().commit()653def get_hosted_files():654    c = get_conn().cursor()655    c.execute("SELECT * FROM Hosted_Files")656    results = c.fetchall()657    hosted_files = []658    for result in results:659        hosted_files.append(HostedFile(result[0], result[1], result[2], result[3], result[4], result[5]))660    return hosted_files661def insert_opsec_event(date, owner, event, note):662    c = get_conn().cursor()663    command = convert_query("INSERT INTO OpSec_Entry (Date, Owner, Event, Note) VALUES (?, ?, ?, ?)")664    c.execute(command, (date, owner, event, note))665    get_conn().commit()666def del_opsec_event(OpsecID):667    c = get_conn().cursor()668    command = convert_query("DELETE FROM Opsec_Entry WHERE OpsecID=?")669    c.execute(command, (OpsecID,))670    get_conn().commit()671def get_opsec_events():672    c = get_conn().cursor()673    c.execute("SELECT * FROM Opsec_Entry")674    result = c.fetchall()675    if result:676        return result677    else:...
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.