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

How to use check_time_sync method in lisa

Best Python code snippet using lisa_python

start_NTP_tests.py

Source:start_NTP_tests.py

1#!/usr/bin/env python32# Breaking: Disable NTP on gw3import pexpect4import sys5DEBUG = False6BREAKING = False7def debug(to_print):8    if DEBUG:9        print(" DEBUG: " + to_print)10def establish_ssh_and_setup(ip, setup_files, ssh_client = None):11    # Copy the needed files12    # If provided with a already set up ssh connection,13    # Copy the needed files through the provided connection.14    if ssh_client == None:15        debug(" Setting up, using scp, on " + ip)16        child = pexpect.spawn("scp -r -P 2220 " + setup_files + " root@" + ip + ":~/")17    else:18        debug(" Setting up, using scp, on " + ip + " through provided ssh-connection")19        child = ssh_client20        child.sendline("scp -r " + setup_files + " root@" + ip + ":~/")21    child.expect("root@" + ip + "'s password:" or pexpect.TIMEOUT)22    child.sendline("password")23    # Establish ssh, if there is already a connection,24    # establish a new one through the provided one25    if ssh_client == None:26        child.expect(pexpect.EOF or pexpect.TIMEOUT)27        child.close()28        29        debug(" Establishing ssh connection to " + ip)30        child = pexpect.spawn("ssh -p 2220 root@" + ip)31    else:32        child.expect("root@.*:~# " or pexpect.TIMEOUT)33        34        debug(" Establishing ssh connection to " + ip + " through provided ssh-connection")35        child.sendline("ssh root@" + ip)36    37    child.expect("root@" + ip + "'s password:" or pexpect.TIMEOUT) 38    child.sendline("password")39    child.expect("root@.*:~# " or pexpect.TIMEOUT)40    return child41if __name__ == "__main__":42    if len(sys.argv) >= 2:43        DEBUG = sys.argv[1].upper() == 'DEBUG'44    if len(sys.argv) == 3:45        BREAKING = sys.argv[2].upper() == "BREAKING"46    breaking = ""47    passed_result = "PASSED without breaking"48    if BREAKING:49        breaking = "break"50        passed_result = "PASSED"51    # Establish ssh to router and run tests52    debug("Setting up tests on gw")53    ssh_client = establish_ssh_and_setup("127.0.0.1", "check_time_sync.py")54    debug("DONE\n")55    # Establish ssh to server and run tests56    debug("Running Tests for server")57    ssh_client = establish_ssh_and_setup("10.0.0.2", "check_time_sync.py", ssh_client)58    debug("     Running test: \"check_time_sync.py\"")59    ssh_client.sendline("python3 check_time_sync.py " + breaking)60    ssh_client.expect("PASSED" or pexpect.TIMEOUT)61    debug(passed_result + "\n")62    # Cleanup on server and exit back to router63    ssh_client.sendline("rm check_time_sync.py")64    ssh_client.sendline("exit")65    ssh_client.expect("root@gw:~# " or pexpect.TIMEOUT)66    # Establish ssh to client-1 and run tests67    debug("Running Tests for client-1")68    ssh_client = establish_ssh_and_setup("10.0.0.3", "check_time_sync.py", ssh_client)69    debug("     Running test: \"check_time_sync.py\"")70    ssh_client.sendline("python3 check_time_sync.py " + breaking)71    ssh_client.expect("PASSED" or pexpect.TIMEOUT)72    debug(passed_result + "\n")73    # Cleanup on client-1 and exit back to router74    ssh_client.sendline("rm check_time_sync.py")75    ssh_client.sendline("exit")76    ssh_client.expect("root@gw:~# " or pexpect.TIMEOUT)77    78    # Establish ssh to client-2 and run tests79    debug("Running Tests for client-2")80    ssh_client = establish_ssh_and_setup("10.0.0.4", "check_time_sync.py", ssh_client)81    debug("     Running test: \"check_time_sync.py\"")82    ssh_client.sendline("python3 check_time_sync.py " + breaking)83    ssh_client.expect("PASSED" or pexpect.TIMEOUT)84    debug(passed_result + "\n")85    # Cleanup on client-2 and exit back to router86    ssh_client.sendline("rm check_time_sync.py")87    ssh_client.sendline("exit")88    ssh_client.expect("root@gw:~# " or pexpect.TIMEOUT)89    90    # Cleanup on router91    ssh_client.sendline("rm check_time_sync.py")92    ssh_client.sendline("exit")93    #END...

timesync_checker.py

Source:timesync_checker.py

...12        self.ros_rate = rospy.Rate(self.config["ros_rate"])13        self.diag_publisher = rospy.Publisher('/diagnostics', DiagnosticArray, queue_size=1)14        time.sleep(2)15    def loop(self):16        timesync_info = self.check_time_sync()17        self.publish_diagnostic_msg(timesync_info)18        self.ros_rate.sleep()19    def check_time_sync(self):20        if self.config["server_hostname"] is not None:21            server_con_info = self.config["server_hostname"]22        else:23            server_con_info = self.config["server_ip"]24        bashCommand = "ntpdate -q {}".format(server_con_info)25        process = subprocess.Popen(bashCommand.split(), stdout=subprocess.PIPE)26        output, error = process.communicate()27        split_msg = output.split()28        info_dict = {"server": split_msg[1], "stratum": split_msg[3], "offset": split_msg[5], "delay": split_msg[7]}29        return info_dict30    def publish_diagnostic_msg(self, timesync_info):31        arr = DiagnosticArray()32        txt_msg = "Server: {}, stratum: {}, offset: {}, delay: {}".format(timesync_info["server"],33                                                                          timesync_info["stratum"],...

function.py

Source:function.py

1import os2import sys3import time4import re5import threading6import webbrowser7import win32com.shell.shell as shell8def extractIp(ip):9    result = re.findall( r'[0-9]+(?:\.[0-9]+){3}', ip ) #IP ì¶ì¶ ì ê·ì10    return result11# ëë©ì¸ ip12def get_domain_ip(url):13    ip = os.popen('ping -n 1 ' + url).read()14    getIp = extractIp(ip) #ì ê·ìì¼ë¡ ipë§ ë½ê¸°.15        16    return getIp[1]17# ëê¸°í ê²°ê³¼18def check_sync():19    check_time_sync = os.popen('w32tm /query /configuration').read()20    return check_time_sync21# ìê° ëê¸°í22def time_sync(url):23    target_url = get_domain_ip(url)24    25    if shell.IsUserAnAdmin():26        os.system('net start w32time')27        os.system('w32tm /config /manualpeerlist:' + target_url + ' /syncfromflags:manual /update')28        return target_url29    else:30        return False31def target_url_open(target_time, target_url):32    global flag33    flag = threading.Event()34    while(not flag.is_set()):35        if(time.strftime('%H:%M') == target_time):36            webbrowser.open(target_url)37            break38        39        time.sleep(0.02)...

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.