Best Python code snippet using lisa_python
virl_setup.py
Source:virl_setup.py  
...222    print('restarting docker registry')223    cmd = 'docker restart registry'224    success_msg = 'docker registry restarted'225    run_command(cmd, success_msg)226def restart_service(name):227    print('')228    print('restarting {}'.format(name))229    cmd = 'systemctl restart {}'.format(name)230    success_msg = '{} restarted'.format(name)231    run_command(cmd, success_msg)232def show_status(name):233    print('')234    print('Status of service {}'.format(name))235    cmd = "systemctl --lines=0 --output=short status {} | grep 'Active:\|Memory:\|CPU:'".format(name)236    run_command(cmd)237def handle_start():238    current_state = ''239    print('****** Main menu ******')240    print('')241    print('1. Network configuration')242    print('2. Maintenance')243    print('3. Diagnostic')244    print('')245    print('0. Exit')246    return read_next_state(current_state)247def handle_0():248    sys.exit()249def handle_1():250    current_state = '1'251    print('***** Network configuration *****')252    print('')253    print('*** WARNING: It is recommended that primary interface  ***')254    print('*** WARNING: configuration is performed on the console ***')255    print('*** WARNING: and not over an SSH connection.           ***')256    print('')257    print('1. Switch primary interface')258    print('2. Run DHCP on primary interface')259    print('3. Static IP configuration on primary interface')260    print('4. DNS configuration')261    print('5. NTP server')262    print('')263    print('0. Back')264    return read_next_state(current_state)265def handle_1_1():266    config = Config(path=VINSTALL_CFG)267    config.user_input(268        field='public_port',269        prompt="Interface",270        default='eth0'271    )272    config.write()273    return press_return_to_continue('1')274def handle_1_2():275    config = Config(VINSTALL_CFG)276    config.set(field='using_dhcp_on_the_public_port', value='True')277    config.write()278    run_command('sudo vinstall salt')279    run_salt_state('virl.host')280    run_salt_state('virl.hostname')281    run_salt_state('virl.network.system')282    return press_return_to_continue('1')283def handle_1_3():284    config = Config(VINSTALL_CFG)285    config.set(field='using_dhcp_on_the_public_port', value='False')286    # static ip287    static_ip = config.user_input(288        field='Static_IP',289        prompt='Static IP',290        default='172.16.6.250',291        validator=Validators.is_ipv4_addr292    )293    # public_netmask294    netmask = config.user_input(295        field='public_netmask',296        prompt='Netmask or netmask bits',297        default='24',298        validator=Validators.is_ipv4_netmask299    )300    # public network301    netmask_bits = netaddr.IPAddress(netmask).netmask_bits()302    network = netaddr.IPNetwork(303        "{network}/{netmask_bits}".format(304            network=static_ip,305            netmask_bits=netmask_bits)306    )307    config.set(field="public_network", value=str(network.network))308    # public_gateway:309    gateway = config.user_input(310        field='public_gateway',311        prompt='Public gateway',312        default='172.16.6.1',313        validator=Validators.is_ipv4_addr,314    )315    if not Validators.is_in_network(316        addr=netaddr.IPAddress(static_ip),317        network=network,318    ):319        print("Incorrect settings: IP address {ip} is not valid for the network {net}/{bits}".format(ip=static_ip, net=str(network.network), bits=netmask_bits))320        return press_return_to_continue('1')321    if not Validators.is_in_network(322        addr=netaddr.IPAddress(gateway),323        network=network,324    ):325        print("Incorrect settings: Gateway IP address {ip} is not valid for the network {net}/{bits}".format(ip=gateway, net=str(network), bits=netmask_bits))326        return press_return_to_continue('1')327    if Validators.is_broadcast(328        addr=netaddr.IPAddress(static_ip),329        network=network,330    ):331        print("Incorrect settings: IP address {ip} is broadcast address of the network {net}/{bits}".format(ip=static_ip, net=str(network), bits=netmask_bits))332        return press_return_to_continue('1')333    if Validators.is_broadcast(334        addr=netaddr.IPAddress(gateway),335        network=network,336    ):337        print("Incorrect settings: Gateway IP address {ip} is broadcast address of the network {net}/{bits}".format(ip=gateway, net=str(network), bits=netmask_bits))338        return press_return_to_continue('1')339    config.write()340    run_command('vinstall salt')341    run_salt_state('virl.host')342    run_salt_state('virl.hostname')343    run_salt_state('virl.network.system')344    return press_return_to_continue('1')345def handle_1_4():346    config = Config(VINSTALL_CFG)347    # first nameserver348    config.user_input(349        field='first_nameserver',350        prompt='First nameserver',351        default='8.8.8.8',352        validator=Validators.is_ipv4_addr,353    )354    # second nameserver355    config.user_input(356        field='second_nameserver',357        prompt='Second nameserver',358        default='8.8.4.4',359        validator=Validators.is_ipv4_addr360    )361    config.write()362    run_command('vinstall salt')363    run_salt_state('virl.host')364    run_salt_state('virl.network.system')365    return press_return_to_continue('1')366def handle_1_5():367    config = Config(VINSTALL_CFG)368    config.user_input(369        field='ntp_server',370        prompt='NTP Server',371        default='ntp.ubuntu.com'372    )373    config.write()374    run_command('vinstall salt')375    run_salt_state('virl.ntp')376    return press_return_to_continue('1')377def handle_2():378    current_state = '2'379    print('***** Maintenance ******')380    print('')381    print('1. Restart VIRL services')382    print('2. Restart OpenStack services')383    print('3. Reset OpenStack worker pools')384    if uksm_enabled_kernel():385        if uksm_enabled():386            print('4. Enable UKSM')387        else:388            print('4. Disable UKSM')389    else:390        print('4. ! A UKSM-enabled kernel is not running')391    print('')392    print('0. Back')393    return read_next_state(current_state)394def handle_2_1():395    current_state = '2.1'396    print('***** Maintenance - VIRL services restart *****')397    print('')398    print('1. All VIRL services')399    print('2. STD services (includes redis)')400    print('3. UWM services (includes webmux)')401    print('4. ANK services')402    print('5. Live visualisation services')403    print('6. Docker registry')404    print('')405    print('0. Back')406    return read_next_state(current_state)407def handle_2_1_1():408    print('***** Restarting all VIRL services *****')409    for service in VIRL_SERVICES:410        restart_service(service)411    restart_docker()412    return press_return_to_continue('2.1')413def handle_2_1_2():414    print('***** Restarting STD services *****')415    restart_service('virl-std.service')416    restart_service('virl-tap-counter.service')417    restart_service('redis.service')418    restart_service('redis-server.service')419    return press_return_to_continue('2.1')420def handle_2_1_3():421    print('***** Restarting UWM services ******')422    restart_service('virl-uwm.service')423    restart_service('virl-webmux.service')424    return press_return_to_continue('2.1')425def handle_2_1_4():426    print('***** Restarting ANK services *****')427    restart_service('ank-cisco-webserver.service')428    return press_return_to_continue('2.1')429def handle_2_1_5():430    print('***** Restarting live visualisation services *****')431    restart_service('virl-vis-mux.service')432    restart_service('virl-vis-processor.service')433    restart_service('virl-vis-webserver.service')434    return press_return_to_continue('2.1')435def handle_2_1_6():436    print('***** Restarting docker registry *****')437    print('')438    restart_docker()439    return press_return_to_continue('2.1')440def handle_2_2():441    current_state = '2.2'442    print('***** Maintenance - OpenStack services restart *****')443    print('')444    print('1. All OpenStack and infrastructure services')445    print('2. All Compute Services')446    print('3. All Networking Services')447    print('4. All Image Services')448    print('5. Identity service')449    print('6. Infrastructure services (MySql, RabbitMQ)')450    print('')451    print('0. Back')452    return read_next_state(current_state)453def handle_2_2_1():454    print('***** Restarting OpenStack services *****')455    for service in OPENSTACK_SERVICES:456        restart_service(service)457    return press_return_to_continue('2.2')458def handle_2_2_2():459    print('***** Restarting OpenStack Compute services *****')460    restart_service('nova-api.service')461    restart_service('nova-compute.service')462    restart_service('nova-consoleauth.service')463    restart_service('nova-cert.service')464    restart_service('nova-conductor.service')465    restart_service('nova-novncproxy.service')466    restart_service('nova-serialproxy.service')467    return press_return_to_continue('2.2')468def handle_2_2_3():469    print('***** Restarting OpenStack Networking services *****')470    restart_service('neutron-dhcp-agent.service')471    restart_service('neutron-linuxbridge-cleanup.service ')472    restart_service('neutron-server.service')473    restart_service('neutron-l3-agent.service')474    restart_service('neutron-metadata-agent.service')475    restart_service('neutron-linuxbridge-agent.service')476    restart_service('neutron-ovs-cleanup.service')477    return press_return_to_continue('2.2')478def handle_2_2_4():479    print('***** Restarting OpenStack Image services *****')480    restart_service('glance-api.service')481    restart_service('glance-registry.service')482    return press_return_to_continue('2.2')483def handle_2_2_5():484    print('***** Restarting OpenStack Identity services *****')485    restart_service('keystone.service')486    return press_return_to_continue('2.2')487def handle_2_2_6():488    print('***** Restarting OpenStack Infrastructure services *****')489    restart_service('mysql.service ')490    restart_service('rabbitmq-server.service')491    return press_return_to_continue('2.2')492def handle_2_3():493    print('***** Reseting OpenStack worker pools *****')494    print('')495    print('This may take some time')496    print('')497    print('Restarting OpenStack workers')498    run_salt_state('openstack.worker_pool')499    return press_return_to_continue('2')500def handle_2_4():501    print('***** UKSM kernel changes *****')502    if uksm_enabled_kernel():503        if uksm_enabled():504            run_command('echo 1 > {}'.format(UKSM_KERNEL_PATH))...default_nsx_impl.py
Source:default_nsx_impl.py  
...72        substitution = 'export GLOG_v=3'73        Linux.replace_in_file(74            client_object=client_object, path=cls.NSXAGENT_PATH, regex=regex,75            substitution=substitution)76        DefaultServiceImpl.restart_service(77            client_object, service_name=cls.NSX_AGENT)78    @classmethod79    def _set_nsxmpa_log_level(cls, client_object, log_level=None):80        """81        Sets the log level of nsx-mpa.82        Method would look for this pattern:83            $ModLoad imuxsock84        and would add following lines after it:85            $SystemLogRateLimitInterval 086            $SystemLogRateLimitBurst 087        @type client_object: BaseClient88        @param client_object: Used to pass commands to the host.89        @type log_level: str90        @param log_level: Specifies whether the log level is being increased or91            decreased.92        """93        restart_service = False94        rate_interval_exists = DefaultFileOpsImpl.file_find_context(95            client_object, file_name=cls.RSYSLOG_PATH,96            start_str="\$ModLoad imuxsock",97            end_str="\$SystemLogRateLimitInterval 0")98        if common.status_codes.FAILURE == rate_interval_exists:99            regex = r'^\$ModLoad imuxsock.*'100            substitution = (r"$ModLoad imuxsock\%s"101                            "$SystemLogRateLimitInterval 0" % os.linesep)102            Linux.replace_in_file(103                client_object=client_object, path=cls.RSYSLOG_PATH,104                regex=regex, substitution=substitution)105            restart_service = True106        rate_burst_exists = DefaultFileOpsImpl.file_find_context(107            client_object, file_name=cls.RSYSLOG_PATH,108            start_str="\$SystemLogRateLimitInterval 0",109            end_str="\$SystemLogRateLimitBurst 0")110        if common.status_codes.FAILURE == rate_burst_exists:111            regex = r"^\$SystemLogRateLimitInterval 0$"112            substitution = (r"$SystemLogRateLimitInterval 0\%s"113                            "$SystemLogRateLimitBurst 0" % os.linesep)114            Linux.replace_in_file(115                client_object=client_object, path=cls.RSYSLOG_PATH,116                regex=regex, substitution=substitution)117            restart_service = True118        if restart_service:119            DefaultServiceImpl.restart_service(120                client_object, service_name=cls.RSYSLOG)121            DefaultServiceImpl.restart_service(122                client_object, service_name=cls.NSX_MPA)123    @classmethod124    def _set_nsxa_log_level(cls, client_object, log_level=None):125        """126        Sets the log level of nsxa log.127        @type client_object: BaseClient128        @param client_object: Used to pass commands to the host.129        @type log_level: str130        @param log_level: Specifies whether the log level is being increased or131            decreased.132        """133        regex = (r'\*\.info;mail\.none;authpriv\.none;cron\.none')134        substitution = ("*.info;*.debug;mail.none;authpriv.none;"135                        "cron.none")136        Linux.replace_in_file(137            client_object=client_object, path=cls.RSYSLOG_PATH, regex=regex,138            substitution=substitution)139        DefaultServiceImpl.restart_service(140            client_object, service_name=cls.RSYSLOG)141        DefaultServiceImpl.restart_service(...main.py
Source:main.py  
...32    win32serviceutil.WaitForServiceStatus(name, status, secs)33def query_service(name: str) -> None:34    status = win32serviceutil.QueryServiceStatus(name)35    return status36def restart_service(name: str) -> None:37    logger.info("============= Restart Service Strat =============")38    try:39        #ìë¹ì¤ê° ì¤ë¨ëê³  ìì ë40        if query_service(name)[1] == 2:41            wait_service(name, 4, 10)42        # ìë¹ì¤ê° ì¤íëê³  ìì ë43        elif query_service(name)[1] == 3:44            wait_service(name, 1, 10)45        # ìë¹ì¤ê° ì´ë¯¸ ì¤íì¤ì¼ë46        elif query_service(name)[1] == 4:47            stop_service(name)48            wait_service(name, 1, 10)49    except:50        logger.info("Service dose not started or stoped")51        restart_service(name)52    if query_service(name)[1] != 4:53        try:54            while True:55                memory_usage_dict = dict(psutil.virtual_memory()._asdict())56                memory_usage_percent = memory_usage_dict['percent']57                # ë©ëª¨ë¦¬ê° ì¶©ë¶í ë°í ë íì ììí  ì ìë ë¡ì§58                if(int(memory_usage_percent) < int(MEMORY_UNDER)):59                    start_service(name)60                    wait_service(name, 4, 10)61                    break62                time.sleep(2)63        except:64            logger.info("Service dose not started")65            restart_service(name)66    logger.info("============= Restart Service Complete =============")67def checkMem() -> None:68    memory_usage_dict = dict(psutil.virtual_memory()._asdict())69    memory_usage_percent = memory_usage_dict['percent']70    if(int(MEMORY_LIMIT) < int(memory_usage_percent)):71        restart_service(SERVICE_NAME)72def job() -> None:73    restart_service(SERVICE_NAME)74if __name__ == '__main__':75    logger.info("============= SDRekindeler Started =============")76    # 5ì´ì íë²ì© ë©ëª¨ë¦¬ë¥¼ ì ê²íëë¡ ì¤ì 77    #schedule.every(5).seconds.do(checkMem)78    # ë§¤ì¼ 10:30 ì ì¤í79    schedule.every().day.at(RESTART_TIME).do(job)80    while True:81        schedule.run_pending()...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
