How to use sub_else_add method in autotest

Best Python code snippet using autotest_python
utils_conn.py
Source:utils_conn.py
...627        if not libvirt_version.version_compare(5, 6, 0, server_session):628            # edit the /etc/sysconfig/libvirtd to add --listen args in libvirtd629            pattern_to_repl = {r".*LIBVIRTD_ARGS\s*=\s*\"\s*--listen\s*\".*":630                               "LIBVIRTD_ARGS=\"--listen\""}631            self.remote_syslibvirtd.sub_else_add(pattern_to_repl)632            # edit the /etc/libvirt/libvirtd.conf633            # listen_tcp=1, tcp_port=$tcp_port, auth_tcp="none"634            # listen_tcp=1, tcp_port=$tcp_port, auth_tcp=$auth_tcp635            pattern_to_repl = {r".*listen_tls\s*=.*": 'listen_tls=0',636                               r".*listen_tcp\s*=.*": 'listen_tcp=1',637                               r".*tcp_port\s*=.*": 'tcp_port="%s"' % (tcp_port),638                               r".*auth_tcp\s*=.*": 'auth_tcp="%s"' % (auth_tcp)}639        else:640            # After libvirt 5.6.0, auth_tcp must be set to 'none'641            pattern_to_repl = {r".*auth_tcp\s*=.*": 'auth_tcp="none"'}642        # a whitelist of allowed SASL usernames, it's a list.643        # If the list is an empty, no client can connect644        if sasl_allowed_users:645            pattern_to_repl[r".*sasl_allowed_username_list\s*=.*"] = \646                'sasl_allowed_username_list=%s' % (sasl_allowed_users)647        if listen_addr:648            pattern_to_repl[r".*listen_addr\s*=.*"] = \649                "listen_addr='%s'" % (listen_addr)650        self.remote_libvirtdconf.sub_else_add(pattern_to_repl)651        # edit the /etc/sasl2/libvirt.conf to change sasl method652        if self.sasl_type == 'gssapi':653            keytab = "keytab: /etc/libvirt/krb5.tab"654        else:655            keytab = ""656        pattern_to_repl = {r".*mech_list\s*:\s*.*":657                           "mech_list: %s" % self.sasl_type,658                           r".*keytab\s*:\s*.*": keytab}659        self.remote_saslconf.sub_else_add(pattern_to_repl)660        if tcp_port != '16509' and libvirt_version.version_compare(5, 6, 0, server_session):661            pattern_to_repl = {r".*ListenStream\s*=.*": 'ListenStream=%s' % (tcp_port)}662            self.remote_libvirtd_tcp_socket.sub_else_add(pattern_to_repl)663        # restart libvirtd service on server664        try:665            session = remote.wait_for_login('ssh', server_ip, '22',666                                            server_user, server_pwd,667                                            r"[\#\$]\s*$")668            remote_runner = remote.RemoteRunner(session=session)669            remote_runner.run('iptables -F', ignore_status=True)670            libvirtd_service = utils_libvirtd.Libvirtd(session=session)671            # From libvirt 5.6, libvirtd is using systemd socket activation672            # by default673            if libvirt_version.version_compare(5, 6, 0, session):674                # Before start libvirtd-tcp.socket, user must stop libvirtd.675                # After libvirtd-tcp.socket is started, user mustn't start676                # libvirtd.677                libvirtd_service.stop()678                session.cmd("systemctl restart libvirtd-tcp.socket")679            else:680                libvirtd_service.restart()681        except (remote.LoginError, aexpect.ShellError) as detail:682            raise ConnServerRestartError(detail)683        logging.debug("TCP connection setup successfully.")684class TLSConnection(ConnectionBase):685    """686    Connection of TLS transport.687    Some specific variables for TLSConnection class.688    server_cn, client_cn, ca_cn: Info to build pki key.689    CERTOOL: tool to build key for TLS connection.690    pki_CA_dir: Dir to store CA key.691    libvirt_pki_dir, libvirt_pki_private_dir: Dir to store pki in libvirt.692    sysconfig_libvirtd_path, libvirtd_conf_path: Path of libvirt config file.693    hosts_path: /etc/hosts694    auth_tls, tls_port, listen_addr: custom TLS Auth, port and listen address695    tls_allowed_dn_list: DN's list are checked696    tls_verify_cert: disable verification, default is to always verify697    tls_sanity_cert: disable checks, default is to always run sanity checks698    custom_pki_path: custom pki path699    ca_cakey_path: CA certification path, sometimes need to reuse previous cert700    scp_new_cacert: copy new CA certification, default is to always copy701    restart_libvirtd: default is to restart libvirtd702    credential_dict: A dict for required file names in libvirt or qemu style703    qemu_tls: True for qemu native TLS support704    qemu_chardev_tls: True for config chardev tls in qemu conf705    """706    __slots__ = ('server_cn', 'client_cn', 'ca_cn', 'CERTTOOL', 'pki_CA_dir',707                 'libvirt_pki_dir', 'libvirt_pki_private_dir', 'client_hosts',708                 'server_libvirtdconf', 'server_syslibvirtd', 'auth_tls',709                 'tls_port', 'listen_addr', 'tls_allowed_dn_list', 'sasl_type',710                 'custom_pki_path', 'tls_verify_cert', 'tls_sanity_cert',711                 'ca_cakey_path', 'scp_new_cacert', 'restart_libvirtd',712                 'client_libvirtdconf', 'client_syslibvirtd', 'server_hosts',713                 'credential_dict', 'qemu_tls', 'qemu_chardev_tls',714                 'server_saslconf', 'server_qemuconf', 'client_qemuconf',715                 'server_libvirtd_tls_socket', 'client_libvirtd_tls_socket')716    def __init__(self, *args, **dargs):717        """718        Initialization of TLSConnection.719        (1).call the init func in ConnectionBase.720        (2).check and set CERTTOOL.721        (3).make a tmp directory as a workspace.722        (4).set values of pki related.723        """724        init_dict = dict(*args, **dargs)725        init_dict['server_cn'] = init_dict.get('server_cn', 'TLSServer')726        init_dict['client_cn'] = init_dict.get('client_cn', 'TLSClient')727        init_dict['ca_cn'] = init_dict.get('ca_cn', 'AUTOTEST.VIRT')728        init_dict['ca_cakey_path'] = init_dict.get('ca_cakey_path', None)729        init_dict['auth_tls'] = init_dict.get('auth_tls', 'none')730        init_dict['tls_port'] = init_dict.get('tls_port', '16514')731        init_dict['listen_addr'] = init_dict.get('listen_addr')732        init_dict['custom_pki_path'] = init_dict.get('custom_pki_path')733        init_dict['tls_verify_cert'] = init_dict.get('tls_verify_cert', 'yes')734        init_dict['tls_sanity_cert'] = init_dict.get('tls_sanity_cert', 'yes')735        init_dict['tls_allowed_dn_list'] = init_dict.get('tls_allowed_dn_list')736        init_dict['scp_new_cacert'] = init_dict.get('scp_new_cacert', 'yes')737        init_dict['sasl_type'] = init_dict.get('sasl_type', 'gssapi')738        init_dict['restart_libvirtd'] = init_dict.get(739            'restart_libvirtd', 'yes')740        super(TLSConnection, self).__init__(init_dict)741        # check and set CERTTOOL in slots742        try:743            CERTTOOL = path.find_command("certtool")744        except path.CmdNotFoundError:745            logging.warning("certtool executable not set or found on path, "746                            "TLS connection will not setup normally")747            CERTTOOL = '/bin/true'748        self.CERTTOOL = CERTTOOL749        self.qemu_tls = "yes" == init_dict.get('qemu_tls', 'no')750        self.qemu_chardev_tls = "yes" == init_dict.get('qemu_chardev_tls', 'no')751        delimeter = ''752        if self.qemu_tls or self.qemu_chardev_tls:753            delimeter = '-'754        self.credential_dict = {'cacert': 'ca%scert.pem' % delimeter,755                                'cakey': 'ca%skey.pem' % delimeter,756                                'servercert': 'server%scert.pem' % delimeter,757                                'serverkey': 'server%skey.pem' % delimeter,758                                'clientcert': 'client%scert.pem' % delimeter,759                                'clientkey': 'client%skey.pem' % delimeter}760        # set some pki related dir values761        if not self.custom_pki_path:762            self.pki_CA_dir = ('/etc/pki/CA/')763            self.libvirt_pki_dir = ('/etc/pki/libvirt/')764            self.libvirt_pki_private_dir = ('/etc/pki/libvirt/private/')765        else:766            # set custom certifications path767            dir_dict = {'CA': 'pki_CA_dir',768                        'libvirt': 'libvirt_pki_dir',769                        'libvirt/private': 'libvirt_pki_private_dir'}770            if not os.path.exists(self.custom_pki_path):771                os.makedirs(self.custom_pki_path)772            for dir_name in dir_dict:773                setattr(self, dir_dict[dir_name], self.custom_pki_path)774        self.server_qemuconf = remote.RemoteFile(775            address=self.server_ip,776            client='scp',777            username=self.server_user,778            password=self.server_pwd,779            port='22',780            remote_path='/etc/libvirt/qemu.conf')781        self.client_qemuconf = remote.RemoteFile(782            address=self.client_ip,783            client='scp',784            username=self.client_user,785            password=self.client_pwd,786            port='22',787            remote_path='/etc/libvirt/qemu.conf')788        self.client_hosts = remote.RemoteFile(789            address=self.client_ip,790            client='scp',791            username=self.client_user,792            password=self.client_pwd,793            port='22',794            remote_path='/etc/hosts')795        self.server_hosts = remote.RemoteFile(796            address=self.server_ip,797            client='scp',798            username=self.server_user,799            password=self.server_pwd,800            port='22',801            remote_path='/etc/hosts')802        self.server_syslibvirtd = remote.RemoteFile(803            address=self.server_ip,804            client='scp',805            username=self.server_user,806            password=self.server_pwd,807            port='22',808            remote_path='/etc/sysconfig/libvirtd')809        self.server_libvirtdconf = remote.RemoteFile(810            address=self.server_ip,811            client='scp',812            username=self.server_user,813            password=self.server_pwd,814            port='22',815            remote_path='/etc/libvirt/libvirtd.conf')816        self.server_saslconf = remote.RemoteFile(817            address=self.server_ip,818            client='scp',819            username=self.server_user,820            password=self.server_pwd,821            port='22',822            remote_path='/etc/sasl2/libvirt.conf')823        self.client_syslibvirtd = remote.RemoteFile(824            address=self.client_ip,825            client='scp',826            username=self.client_user,827            password=self.client_pwd,828            port='22',829            remote_path='/etc/sysconfig/libvirtd')830        self.client_libvirtdconf = remote.RemoteFile(831            address=self.client_ip,832            client='scp',833            username=self.client_user,834            password=self.client_pwd,835            port='22',836            remote_path='/etc/libvirt/libvirtd.conf')837        self.client_libvirtd_tls_socket = remote.RemoteFile(838            address=self.client_ip,839            client='scp',840            username=self.client_user,841            password=self.client_pwd,842            port='22',843            remote_path='/usr/lib/systemd/system/libvirtd-tls.socket')844        self.server_libvirtd_tls_socket = remote.RemoteFile(845            address=self.server_ip,846            client='scp',847            username=self.server_user,848            password=self.server_pwd,849            port='22',850            remote_path='/usr/lib/systemd/system/libvirtd-tls.socket')851    def conn_recover(self):852        """853        Do the clean up work.854        (1).initialize variables.855        (2).Delete remote file.856        (3).Restart libvirtd on server.857        """858        # clean up certifications firstly859        if self.auto_recover:860            self.cert_recover()861        # initialize variables862        server_ip = self.server_ip863        server_user = self.server_user864        server_pwd = self.server_pwd865        del self.client_hosts866        del self.server_syslibvirtd867        del self.server_libvirtdconf868        del self.server_qemuconf869        del self.server_hosts870        del self.server_saslconf871        del self.client_syslibvirtd872        del self.client_libvirtdconf873        del self.client_qemuconf874        del self.server_libvirtd_tls_socket875        del self.client_libvirtd_tls_socket876        # restart libvirtd service on server877        try:878            session = remote.wait_for_login('ssh', server_ip, '22',879                                            server_user, server_pwd,880                                            r"[\#\$]\s*$")881            libvirtd_service = utils_libvirtd.Libvirtd(session=session)882            if libvirt_version.version_compare(5, 6, 0, session):883                session.cmd("systemctl stop libvirtd-tls.socket")884                libvirtd_service.start()885            else:886                libvirtd_service.restart()887        except (remote.LoginError, aexpect.ShellError) as detail:888            raise ConnServerRestartError(detail)889        logging.debug("TLS connection recover successfully.")890    def cert_recover(self):891        """892        Do the clean up certifications work.893        (1).initialize variables.894        (2).Delete local and remote generated certifications file.895        """896        # initialize variables897        server_ip = self.server_ip898        server_user = self.server_user899        server_pwd = self.server_pwd900        cert_dict = {'CA': '%s*' % self.pki_CA_dir,901                     'cert': self.libvirt_pki_dir,902                     'key': self.libvirt_pki_private_dir}903        # remove local generated certifications file904        for cert in cert_dict:905            cert_path = cert_dict[cert]906            cmd = "rm -rf %s" % cert_path907            if os.path.exists(cert_path):908                shutil.rmtree(cert_path)909            else:910                status, output = process.getstatusoutput(cmd)911                if status:912                    raise ConnRmCertError(cert_path, output)913        # remove remote generated certifications file914        server_session = remote.wait_for_login('ssh', server_ip, '22',915                                               server_user, server_pwd,916                                               r"[\#\$]\s*$")917        for cert in cert_dict:918            cert_path = cert_dict[cert]919            cmd = "rm -rf %s" % cert_path920            status, output = server_session.cmd_status_output(cmd)921            if status:922                raise ConnRmCertError(cert_path, output)923        server_session.close()924        logging.debug("TLS certifications recover successfully.")925    def conn_setup(self, server_setup=True, client_setup=True,926                   server_setup_local=False):927        """928        setup a TLS connection between server and client.929        At first check the certtool needed to setup.930        Then call some setup functions to complete connection setup.931        :param server_setup: True to setup TLS server on target host,932                             False to not setup933        :param client_setup: True to setup TLS client on source host,934                             False to not setup935        :param server_setup_local: True to setup TLS server on source host,936                             False to not setup937        """938        if self.CERTTOOL == '/bin/true':939            raise ConnToolNotFoundError('certtool',940                                        "certtool executable not set or found on path.")941        # support build multiple CAs with different CA CN942        build_CA(self.tmp_dir, self.ca_cn,943                 self.ca_cakey_path, self.CERTTOOL,944                 self.credential_dict)945        # not always need to setup CA, client and server together946        if server_setup:947            self.server_setup()948        if client_setup:949            self.client_setup()950        if server_setup_local:951            self.server_setup(on_local=True)952        self.close_session()953        logging.debug("TLS connection setup successfully.")954    def server_setup(self, on_local=False):955        """956        setup private key and certificate file for server.957        (1).initialization for variables.958        (2).build server key.959        (3).copy files to server.960        (4).edit /etc/sysconfig/libvirtd on server.961        (5).edit /etc/libvirt/libvirtd.conf on server.962        (6).restart libvirtd service on server.963        :param on_local: True to setup TLS server on source host,964                         otherwise not.965        """966        # initialize variables967        tmp_dir = self.tmp_dir968        scp_new_cacert = self.scp_new_cacert969        # sometimes, need to reuse previous CA cert970        if self.ca_cakey_path and scp_new_cacert == 'no':971            cacert_path = os.path.join(self.ca_cakey_path, self.credential_dict['cacert'])972            cakey_path = os.path.join(self.ca_cakey_path, self.credential_dict['cakey'])973        else:974            cacert_path = os.path.join(tmp_dir, self.credential_dict['cacert'])975            cakey_path = os.path.join(tmp_dir, self.credential_dict['cakey'])976        serverkey_path = os.path.join(tmp_dir, self.credential_dict['serverkey'])977        servercert_path = os.path.join(tmp_dir, self.credential_dict['servercert'])978        # If need setup TLS server on source machine,979        # we need switch the machine information between source and target machines980        if on_local:981            server_ip = self.client_ip982            server_user = self.client_user983            server_pwd = self.client_pwd984            server_cn = self.client_cn985        else:986            server_ip = self.server_ip987            server_user = self.server_user988            server_pwd = self.server_pwd989            server_cn = self.server_cn990        auth_tls = self.auth_tls991        tls_port = self.tls_port992        listen_addr = self.listen_addr993        restart_libvirtd = self.restart_libvirtd994        tls_allowed_dn_list = self.tls_allowed_dn_list995        pki_path = self.custom_pki_path996        tls_verify_cert = self.tls_verify_cert997        tls_sanity_cert = self.tls_sanity_cert998        # build a server key.999        build_server_key(tmp_dir, self.ca_cakey_path,1000                         server_cn, self.CERTTOOL,1001                         self.credential_dict, on_local)1002        # scp cacert.pem, servercert.pem and serverkey.pem to server.1003        if on_local:1004            server_session = self.client_session1005        else:1006            server_session = self.server_session1007        if self.sasl_type == 'digest-md5':1008            utils_package.package_install('cyrus-sasl-md5', session=server_session)1009        cmd = "mkdir -p %s" % self.libvirt_pki_private_dir1010        status, output = server_session.cmd_status_output(cmd)1011        if status:1012            raise ConnMkdirError(self.libvirt_pki_private_dir, output)1013        scp_dict = {cacert_path: self.pki_CA_dir,1014                    cakey_path: self.pki_CA_dir,1015                    servercert_path: self.libvirt_pki_dir,1016                    serverkey_path: self.libvirt_pki_private_dir}1017        for key in scp_dict:1018            local_path = key1019            remote_path = scp_dict[key]1020            try:1021                remote.copy_files_to(server_ip, 'scp', server_user,1022                                     server_pwd, '22', local_path, remote_path)1023            except remote.SCPError as detail:1024                raise ConnSCPError('AdminHost', local_path,1025                                   server_ip, remote_path, detail)1026        # When qemu supports TLS, it needs not to modify below1027        # configuration files, so simply return1028        if self.qemu_tls:1029            return1030        # Ensure to use proper configuration objects1031        if on_local:1032            operate_libvirtdconf = self.client_libvirtdconf1033            operate_syslibvirtd = self.client_syslibvirtd1034            operate_libvirtd_tls_socket = self.client_libvirtd_tls_socket1035            operate_qemuconf = self.server_qemuconf1036        else:1037            operate_libvirtdconf = self.server_libvirtdconf1038            operate_syslibvirtd = self.server_syslibvirtd1039            operate_libvirtd_tls_socket = self.server_libvirtd_tls_socket1040            operate_qemuconf = self.client_qemuconf1041        # Change qemu conf file to support tls for chardev1042        if self.qemu_chardev_tls:1043            pattern2repl = {r".*chardev_tls\s*=\s*.*":1044                            "chardev_tls = 1"}1045            operate_qemuconf.sub_else_add(pattern2repl)1046            pattern2repl = {r".*chardev_tls_x509_cert_dir\s*="1047                            "\s*\"\/etc\/pki\/libvirt-chardev\s*\".*":1048                            "chardev_tls_x509_cert_dir="1049                            "\"/etc/pki/libvirt-chardev\""}1050            operate_qemuconf.sub_else_add(pattern2repl)1051        if not libvirt_version.version_compare(5, 6, 0, server_session):1052            # After libvirt 5.6.0, no need to set --listen for libvirt tls.1053            # Instead, libvirt use socket file on target host to handle1054            # the listen port.1055            # Before libvirt 5.6.0, edit the /etc/sysconfig/libvirtd to add1056            # --listen args in libvirtd1057            pattern_to_repl = {r".*LIBVIRTD_ARGS\s*=\s*\"\s*--listen\s*\".*":1058                               "LIBVIRTD_ARGS=\"--listen\""}1059            operate_syslibvirtd.sub_else_add(pattern_to_repl)1060            # edit the /etc/libvirt/libvirtd.conf to add listen_tls=11061            pattern_to_repl = {r".*listen_tls\s*=\s*.*": "listen_tls=1"}1062            operate_libvirtdconf.sub_else_add(pattern_to_repl)1063        # edit the /etc/libvirt/libvirtd.conf to add1064        # listen_addr=$listen_addr1065        if listen_addr:1066            pattern_to_repl = {r".*listen_addr\s*=.*":1067                               "listen_addr='%s'" % listen_addr}1068            operate_libvirtdconf.sub_else_add(pattern_to_repl)1069        # edit the /etc/libvirt/libvirtd.conf to add auth_tls=$auth_tls1070        if auth_tls != 'none':1071            pattern_to_repl = {r".*auth_tls\s*=\s*.*": 'auth_tls="%s"' % auth_tls}1072            operate_libvirtdconf.sub_else_add(pattern_to_repl)1073        elif libvirt_version.version_compare(5, 6, 0, server_session):1074            pattern_to_repl = {r".*auth_tls\s*=\s*.*": 'auth_tls="none"'}1075            operate_libvirtdconf.sub_else_add(pattern_to_repl)1076        # edit the /etc/libvirt/libvirtd.conf to add tls_port=$tls_port1077        if tls_port != '16514':1078            if libvirt_version.version_compare(5, 6, 0, server_session):1079                pattern_to_repl = {r".*ListenStream\s*=\s*.*": 'ListenStream=%s' % tls_port}1080                operate_libvirtd_tls_socket.sub_else_add(pattern_to_repl)1081            else:1082                pattern_to_repl = {r".*tls_port\s*=\s*.*": 'tls_port="%s"' % tls_port}1083                operate_libvirtdconf.sub_else_add(pattern_to_repl)1084        # edit the /etc/libvirt/libvirtd.conf to add1085        # tls_allowed_dn_list=$tls_allowed_dn_list1086        if isinstance(tls_allowed_dn_list, list):1087            pattern_to_repl = {r".*tls_allowed_dn_list\s*=\s*.*":1088                               'tls_allowed_dn_list=%s' % tls_allowed_dn_list}1089            operate_libvirtdconf.sub_else_add(pattern_to_repl)1090        # edit the /etc/libvirt/libvirtd.conf to override1091        # the default server certification file path1092        if pki_path:1093            cert_path_dict = {'ca_file': cacert_path,1094                              'key_file': serverkey_path,1095                              'cert_file': servercert_path}1096            pattern_to_repl = {}1097            for cert_name in cert_path_dict:1098                cert_file = os.path.basename(cert_path_dict[cert_name])1099                abs_cert_file = os.path.join(pki_path, cert_file)1100                pattern_to_repl[r".*%s\s*=.*" % (cert_name)] = \1101                    '%s="%s"' % (cert_name, abs_cert_file)1102            operate_libvirtdconf.sub_else_add(pattern_to_repl)1103        # edit the /etc/libvirt/libvirtd.conf to disable client verification1104        if tls_verify_cert == "no":1105            pattern_to_repl = {r".*tls_no_verify_certificate\s*=\s*.*":1106                               'tls_no_verify_certificate=1'}1107            operate_libvirtdconf.sub_else_add(pattern_to_repl)1108        # edit the /etc/libvirt/libvirtd.conf to disable server sanity checks1109        if tls_sanity_cert == "no":1110            pattern_to_repl = {r".*tls_no_sanity_certificate\s*=\s*.*":1111                               'tls_no_sanity_certificate=1'}1112            operate_libvirtdconf.sub_else_add(pattern_to_repl)1113        # edit the /etc/sasl2/libvirt.conf to change sasl method1114        if self.sasl_type == 'gssapi':1115            keytab = "keytab: /etc/libvirt/krb5.tab"1116        else:1117            keytab = ""1118        pattern_to_repl = {r".*mech_list\s*:\s*.*":1119                           "mech_list: %s" % self.sasl_type,1120                           r".*keytab\s*:\s*.*": keytab}1121        self.server_saslconf.sub_else_add(pattern_to_repl)1122        # restart libvirtd service on server1123        if restart_libvirtd == "yes":1124            if on_local:1125                libvirtd_service = utils_libvirtd.Libvirtd()1126                # From libvirt 5.6, libvirtd is using systemd socket activation1127                # by default1128                if libvirt_version.version_compare(5, 6, 0):1129                    process.run("systemctl stop libvirtd.socket")1130                    libvirtd_service.stop()1131                    process.run("systemctl start libvirtd.socket")1132                    process.run("systemctl restart libvirtd-tls.socket")1133                    libvirtd_service.start()1134                else:1135                    libvirtd_service.restart()1136            else:1137                try:1138                    session = remote.wait_for_login('ssh', server_ip, '22',1139                                                    server_user, server_pwd,1140                                                    r"[\#\$]\s*$")1141                    remote_runner = remote.RemoteRunner(session=session)1142                    remote_runner.run('iptables -F', ignore_status=True)1143                    libvirtd_service = utils_libvirtd.Libvirtd(session=session)1144                    if libvirt_version.version_compare(5, 6, 0, session):1145                        libvirtd_service.stop()1146                        session.cmd("systemctl restart libvirtd-tls.socket")1147                    else:1148                        libvirtd_service.restart()1149                except (remote.LoginError, aexpect.ShellError) as detail:1150                    raise ConnServerRestartError(detail)1151        # edit /etc/hosts on remote host in case of connecting1152        # from remote host to local host1153        if not on_local:1154            pattern_to_repl = {r".*%s.*" % self.client_cn:1155                               "%s %s" % (self.client_ip, self.client_cn)}1156            self.server_hosts.sub_else_add(pattern_to_repl)1157    def client_setup(self):1158        """1159        setup private key and certificate file for client.1160        (1).initialization for variables.1161        (2).build a key for client.1162        (3).copy files to client.1163        (4).edit /etc/hosts on client.1164        """1165        # initialize variables1166        tmp_dir = self.tmp_dir1167        cacert_path = os.path.join(tmp_dir, self.credential_dict['cacert'])1168        cakey_path = os.path.join(tmp_dir, self.credential_dict['cakey'])1169        clientkey_path = os.path.join(tmp_dir, self.credential_dict['clientkey'])1170        clientcert_path = os.path.join(tmp_dir, self.credential_dict['clientcert'])1171        client_ip = self.client_ip1172        client_user = self.client_user1173        client_pwd = self.client_pwd1174        # build a client key.1175        build_client_key(tmp_dir, self.client_cn, self.CERTTOOL,1176                         self.credential_dict)1177        # scp cacert.pem, clientcert.pem and clientkey.pem to client.1178        client_session = self.client_session1179        if self.sasl_type == 'digest-md5':1180            utils_package.package_install('cyrus-sasl-md5', session=client_session)1181        for target_dir in [self.pki_CA_dir, self.libvirt_pki_private_dir]:1182            if not os.path.exists(target_dir):1183                cmd = "mkdir -p %s" % target_dir1184                status, output = client_session.cmd_status_output(cmd)1185                if status:1186                    raise ConnMkdirError(target_dir, output)1187        scp_dict = {cacert_path: self.pki_CA_dir,1188                    cakey_path: self.pki_CA_dir,1189                    clientcert_path: self.libvirt_pki_dir,1190                    clientkey_path: self.libvirt_pki_private_dir}1191        for key in scp_dict:1192            local_path = key1193            remote_path = scp_dict[key]1194            try:1195                remote.copy_files_to(client_ip, 'scp', client_user,1196                                     client_pwd, '22', local_path, remote_path)1197            except remote.SCPError as detail:1198                raise ConnSCPError('AdminHost', local_path,1199                                   client_ip, remote_path, detail)1200        # edit /etc/hosts on client1201        pattern_to_repl = {r".*%s.*" % self.server_cn:1202                           "%s %s" % (self.server_ip, self.server_cn)}1203        self.client_hosts.sub_else_add(pattern_to_repl)1204def build_client_key(tmp_dir, client_cn="TLSClient", certtool="certtool",1205                     credential_dict=None):1206    """1207    (1).initialization for variables.1208    (2).make a private key with certtool command.1209    (3).prepare a info file.1210    (4).make a certificate file with certtool command.1211    :param client_cn: cn for client info1212    :param certtool: cert command1213    :param credential_dict: A dict for credential files' names1214    """1215    # Initialize variables1216    cakey_path = os.path.join(tmp_dir, credential_dict['cakey'])1217    cacert_path = os.path.join(tmp_dir, credential_dict['cacert'])1218    clientkey_path = os.path.join(tmp_dir, credential_dict['clientkey'])1219    clientcert_path = os.path.join(tmp_dir, credential_dict['clientcert'])1220    clientinfo_path = os.path.join(tmp_dir, 'client.info')1221    # make a private key.1222    cmd = "%s --generate-privkey > %s" % (certtool, clientkey_path)1223    CmdResult = process.run(cmd, ignore_status=True, shell=True)1224    if CmdResult.exit_status:1225        raise ConnPrivKeyError(clientkey_path, results_stderr_52lts(CmdResult))1226    # prepare a info file to build clientcert.1227    clientinfo_file = open(clientinfo_path, "w")1228    clientinfo_file.write("organization = AUTOTEST.VIRT\n")1229    clientinfo_file.write("cn = %s\n" % (client_cn))1230    clientinfo_file.write("tls_www_client\n")1231    clientinfo_file.write("encryption_key\n")1232    clientinfo_file.write("signing_key\n")1233    clientinfo_file.close()1234    # make a client certificate file and a client key file.1235    cmd = ("%s --generate-certificate --load-privkey %s \1236           --load-ca-certificate %s --load-ca-privkey %s \1237           --template %s --outfile %s" %1238           (certtool, clientkey_path, cacert_path,1239            cakey_path, clientinfo_path, clientcert_path))1240    CmdResult = process.run(cmd, ignore_status=True)1241    if CmdResult.exit_status:1242        raise ConnCertError(clientinfo_path, results_stderr_52lts(CmdResult))1243def build_server_key(tmp_dir, ca_cakey_path=None,1244                     server_cn="TLSServer", certtool="certtool",1245                     credential_dict=None, on_local=False):1246    """1247    (1).initialization for variables.1248    (2).make a private key with certtool command.1249    (3).prepare a info file.1250    (4).make a certificate file with certtool command.1251    :param client_cn: cn for client info1252    :param certtool: cert command1253    :param credential_dict: A dict for credential files' names1254    :param on_local: True to clean up old server key on source host1255    """1256    # initialize variables1257    # sometimes, need to reuse previous CA cert1258    if not ca_cakey_path:1259        cakey_path = os.path.join(tmp_dir, credential_dict['cakey'])1260        cacert_path = os.path.join(tmp_dir, credential_dict['cacert'])1261    else:1262        cakey_path = os.path.join(ca_cakey_path, credential_dict['cakey'])1263        cacert_path = os.path.join(ca_cakey_path, credential_dict['cacert'])1264    serverkey_path = os.path.join(tmp_dir, credential_dict['serverkey'])1265    servercert_path = os.path.join(tmp_dir, credential_dict['servercert'])1266    serverinfo_path = os.path.join(tmp_dir, 'server.info')1267    if on_local:1268        # delete serverkey.pem, servercert.pem and server.info1269        # already created for remote host1270        if os.path.exists(serverkey_path):1271            os.remove(serverkey_path)1272        if os.path.exists(servercert_path):1273            os.remove(servercert_path)1274        if os.path.exists(serverinfo_path):1275            os.remove(serverinfo_path)1276    # make a private key1277    cmd = "%s --generate-privkey > %s" % (certtool, serverkey_path)1278    cmd_result = process.run(cmd, ignore_status=True, shell=True)1279    if cmd_result.exit_status:1280        raise ConnPrivKeyError(serverkey_path, results_stderr_52lts(cmd_result))1281    # prepare a info file to build servercert and serverkey1282    serverinfo_file = open(serverinfo_path, "w")1283    serverinfo_file.write("organization = AUTOTEST.VIRT\n")1284    serverinfo_file.write("cn = %s\n" % (server_cn))1285    serverinfo_file.write("tls_www_server\n")1286    serverinfo_file.write("encryption_key\n")1287    serverinfo_file.write("signing_key\n")1288    serverinfo_file.close()1289    # make a server certificate file and a server key file1290    cmd = ("%s --generate-certificate --load-privkey %s \1291           --load-ca-certificate %s --load-ca-privkey %s \1292           --template %s --outfile %s" %1293           (certtool, serverkey_path, cacert_path,1294            cakey_path, serverinfo_path, servercert_path))1295    CmdResult = process.run(cmd, ignore_status=True)1296    if CmdResult.exit_status:1297        raise ConnCertError(serverinfo_path, results_stderr_52lts(CmdResult))1298def build_CA(tmp_dir, cn="AUTOTEST.VIRT", ca_cakey_path=None,1299             certtool="certtool", credential_dict=None):1300    """1301    setup private key and certificate file which are needed to build.1302    certificate file for client and server.1303    (1).initialization for variables.1304    (2).make a private key with certtool command.1305    (3).prepare a info file.1306    (4).make a certificate file with certtool command.1307    :param tmp_dir: temp directory to store credentail files in1308    :param cn: cn for CA info1309    :param ca_cakey_path: path of CA key file1310    :param certtool: cert command1311    :param credential_dict: A dict for credential files' names1312    """1313    # initialize variables1314    if not ca_cakey_path:1315        cakey_path = os.path.join(tmp_dir, credential_dict['cakey'])1316    else:1317        cakey_path = os.path.join(ca_cakey_path, credential_dict['cakey'])1318    cainfo_path = os.path.join(tmp_dir, 'ca.info')1319    cacert_path = os.path.join(tmp_dir, credential_dict['cacert'])1320    # make a private key1321    # sometimes, may reuse previous CA cert, so don't always need to1322    # generate private key1323    if not ca_cakey_path:1324        cmd = "%s --generate-privkey > %s " % (certtool, cakey_path)1325        cmd_result = process.run(cmd, ignore_status=True, timeout=10, shell=True)1326        if cmd_result.exit_status:1327            raise ConnPrivKeyError(cakey_path, results_stderr_52lts(cmd_result))1328    # prepare a info file to build certificate file1329    cainfo_file = open(cainfo_path, "w")1330    cainfo_file.write("cn = %s\n" % cn)1331    cainfo_file.write("ca\n")1332    cainfo_file.write("cert_signing_key\n")1333    cainfo_file.close()1334    # make a certificate file to build clientcert and servercert1335    cmd = ("%s --generate-self-signed --load-privkey %s\1336           --template %s --outfile %s" %1337           (certtool, cakey_path, cainfo_path, cacert_path))1338    CmdResult = process.run(cmd, ignore_status=True)1339    if CmdResult.exit_status:1340        raise ConnCertError(cainfo_path, results_stderr_52lts(CmdResult))1341class UNIXConnection(ConnectionBase):1342    """1343    Connection class for UNIX transport.1344    Some specific variables for UNIXConnection class.1345    """1346    __slots__ = ('auth_unix_ro', 'auth_unix_rw', 'unix_sock_dir',1347                 'unix_sock_group', 'unix_sock_ro_perms',1348                 'unix_sock_rw_perms', 'access_drivers',1349                 'client_ip', 'client_user', 'client_pwd',1350                 'client_libvirtdconf', 'restart_libvirtd',1351                 'client_saslconf', 'client_hosts', 'sasl_type', 'libvirt_ver',1352                 'sasl_allowed_username_list', 'client_libvirtd_socket')1353    def __init__(self, *args, **dargs):1354        """1355        init params for UNIX connection.1356        :param auth_unix_ro: UNIX R/O sockets, default is 'none'.1357        :param auth_unix_rw: UNIX R/W sockets, default is 'none'.1358        :param unix_sock_group: UNIX domain socket group ownership,1359                                default is 'libvirt'.1360        :param access_drivers: access control restrictions,1361                               default is '["polkit"]'.1362        :param unix_sock_ro_perms: UNIX socket permissions for the1363                                   R/O socket, default is '0777'.1364        :param unix_sock_rw_perms: UNIX socket permissions for the1365                                   R/W socket, default is '0770'.1366        :param client_libvirtdconf: Path of client libvirtd.conf, default is1367                                  '/etc/libvirt/libvirtd.conf'.1368        :param restart_libvirtd: default is to restart libvirtd.1369        """1370        init_dict = dict(*args, **dargs)1371        init_dict['auth_unix_ro'] = init_dict.get('auth_unix_ro', 'none')1372        init_dict['auth_unix_rw'] = init_dict.get('auth_unix_rw', 'none')1373        init_dict['sasl_type'] = init_dict.get('sasl_type', 'gssapi')1374        init_dict['unix_sock_dir'] = init_dict.get(1375            'unix_sock_dir', '/var/run/libvirt')1376        init_dict['unix_sock_group'] = init_dict.get(1377            'unix_sock_group', 'libvirt')1378        init_dict['access_drivers'] = init_dict.get(1379            'access_drivers', ["polkit"])1380        init_dict['unix_sock_ro_perms'] = init_dict.get(1381            'unix_sock_ro_perms', '0777')1382        init_dict['unix_sock_rw_perms'] = init_dict.get(1383            'unix_sock_rw_perms', '0770')1384        init_dict['restart_libvirtd'] = init_dict.get(1385            'restart_libvirtd', 'yes')1386        init_dict['sasl_allowed_username_list'] = init_dict.get(1387            'sasl_allowed_username_list', '["root/admin" ]')1388        super(UNIXConnection, self).__init__(init_dict)1389        # Unable to get libvirt verion via libvirt_version.version_compare1390        # once UNIX connection is setup, so set the value here.1391        client_session = self.client_session1392        self.libvirt_ver = libvirt_version.version_compare(5, 6, 0, client_session)1393        self.client_libvirtdconf = remote.RemoteFile(1394            address=self.client_ip,1395            client='scp',1396            username=self.client_user,1397            password=self.client_pwd,1398            port='22',1399            remote_path='/etc/libvirt/libvirtd.conf')1400        self.client_libvirtd_socket = remote.RemoteFile(1401            address=self.client_ip,1402            client='scp',1403            username=self.client_user,1404            password=self.client_pwd,1405            port='22',1406            remote_path='/usr/lib/systemd/system/libvirtd.socket')1407        self.client_saslconf = remote.RemoteFile(1408            address=self.client_ip,1409            client='scp',1410            username=self.client_user,1411            password=self.client_pwd,1412            port='22',1413            remote_path='/etc/sasl2/libvirt.conf')1414        self.client_hosts = remote.RemoteFile(1415            address=self.client_ip,1416            client='scp',1417            username=self.client_user,1418            password=self.client_pwd,1419            port='22',1420            remote_path='/etc/hosts')1421    def conn_recover(self):1422        """1423        Do the clean up work.1424        (1).Delete remote file.1425        (2).Restart libvirtd on server.1426        """1427        del self.client_libvirtdconf1428        del self.client_saslconf1429        del self.client_hosts1430        del self.client_libvirtd_socket1431        # restart libvirtd service on server1432        client_session = self.client_session1433        try:1434            libvirtd_service = utils_libvirtd.Libvirtd(session=client_session)1435            if self.libvirt_ver:1436                process.run("systemctl daemon-reload")1437                process.run("systemctl stop libvirtd.socket")1438                libvirtd_service.stop()1439                process.run("systemctl start libvirtd.socket")1440                libvirtd_service.start()1441            else:1442                libvirtd_service.restart()1443        except (remote.LoginError, aexpect.ShellError,1444                process.CmdError) as detail:1445            raise ConnServerRestartError(detail)1446        logging.debug("UNIX connection recover successfully.")1447    def conn_setup(self):1448        """1449        Setup a UNIX connection.1450        (1).Initialize variables.1451        (2).Update libvirtd.conf configuration.1452        (3).Update libvirtd.socket for libvirt >= 5.6.1453        (4).Restart libvirtd on client.1454        """1455        # initialize variables1456        auth_unix_ro = self.auth_unix_ro1457        auth_unix_rw = self.auth_unix_rw1458        unix_sock_group = self.unix_sock_group1459        unix_sock_dir = self.unix_sock_dir1460        unix_sock_ro_perms = self.unix_sock_ro_perms1461        unix_sock_rw_perms = self.unix_sock_rw_perms1462        access_drivers = self.access_drivers1463        restart_libvirtd = self.restart_libvirtd1464        client_session = self.client_session1465        sasl_allowed_username_list = self.sasl_allowed_username_list1466        # edit the /etc/libvirt/libvirtd.conf to add auth_unix_ro arg1467        if auth_unix_ro:1468            pattern_to_repl = {r".*auth_unix_ro\s*=.*":1469                               'auth_unix_ro="%s"' % auth_unix_ro}1470            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1471        # edit the /etc/libvirt/libvirtd.conf to add auth_unix_rw arg1472        if auth_unix_rw:1473            pattern_to_repl = {r".*auth_unix_rw\s*=.*":1474                               'auth_unix_rw="%s"' % auth_unix_rw}1475            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1476        # edit the /etc/libvirt/libvirtd.conf to add unix_sock_group arg1477        if unix_sock_group != 'libvirt':1478            pattern_to_repl = {r".*unix_sock_group\s*=.*":1479                               'unix_sock_group="%s"' % unix_sock_group}1480            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1481        # edit the /etc/libvirt/libvirtd.conf to add unix_sock_dir arg1482        if unix_sock_dir != '/var/run/libvirt':1483            if self.libvirt_ver:1484                pattern_to_repl = {r".*ListenStream\s*=.*":1485                                   'ListenStream=%s/libvirt-sock' % unix_sock_dir}1486                self.client_libvirtd_socket.sub_else_add(pattern_to_repl)1487            else:1488                pattern_to_repl = {r".*unix_sock_dir\s*=.*":1489                                   'unix_sock_dir="%s"' % unix_sock_dir}1490                self.client_libvirtdconf.sub_else_add(pattern_to_repl)1491        # edit the /etc/libvirt/libvirtd.conf to add access_drivers arg1492        if access_drivers != ["polkit"]:1493            pattern_to_repl = {r".*access_drivers\s*=.*":1494                               'access_drivers="%s"' % access_drivers}1495            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1496        if auth_unix_rw == 'sasl':1497            pattern_to_repl = {r".*access_drivers\s*=.*":1498                               '#access_drivers="%s"' % access_drivers}1499            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1500        # edit the /etc/libvirt/libvirtd.conf to add unix_sock_ro_perms arg1501        if unix_sock_ro_perms:1502            pattern_to_repl = {r".*unix_sock_ro_perms\s*=.*":1503                               'unix_sock_ro_perms="%s"' % unix_sock_ro_perms}1504            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1505        # edit the /etc/libvirt/libvirtd.conf to add unix_sock_rw_perms arg1506        if unix_sock_rw_perms:1507            if self.libvirt_ver:1508                pattern_to_repl = {r".*SocketMode\s*=.*":1509                                   'SocketMode=%s' % unix_sock_rw_perms}1510                self.client_libvirtd_socket.sub_else_add(pattern_to_repl)1511            else:1512                pattern_to_repl = {r".*unix_sock_rw_perms\s*=.*":1513                                   'unix_sock_rw_perms="%s"' % unix_sock_rw_perms}1514                self.client_libvirtdconf.sub_else_add(pattern_to_repl)1515        if self.sasl_type == 'digest-md5':1516            utils_package.package_install('cyrus-sasl-md5', session=client_session)1517        # edit the /etc/sasl2/libvirt.conf to change sasl method and1518        # edit the /etc/hosts to add the host1519        if self.sasl_type == 'gssapi' and auth_unix_rw == 'sasl':1520            keytab = "keytab: /etc/libvirt/krb5.tab"1521            sasldb = ""1522            remote_runner = remote.RemoteRunner(session=client_session)1523            hostname = remote_runner.run('hostname', ignore_status=True).stdout.strip()1524            pattern_to_repl = {r".*127.0.0.1\s*.*":1525                               "127.0.0.1    %s localhost localhost.localdomain "1526                               "localhost4 localhost4.localdomain6" % hostname,1527                               r".*::1\s*.*":1528                               "::1    %s localhost localhost.localdomain "1529                               "localhost6 localhost6.localdomain6" % hostname1530                               }1531            self.client_hosts.sub_else_add(pattern_to_repl)1532            pattern_to_repl = {r".*sasl_allowed_username_list\s*=.*":1533                               'sasl_allowed_username_list=%s' % sasl_allowed_username_list}1534            self.client_libvirtdconf.sub_else_add(pattern_to_repl)1535        else:1536            keytab = ""1537            sasldb = "sasldb_path: /etc/libvirt/passwd.db"1538        pattern_to_repl = {r".*mech_list\s*:\s*.*":1539                           "mech_list: %s" % self.sasl_type,1540                           r".*keytab\s*:\s*.*": keytab,1541                           r".*sasldb_path\s*:\s*.*": sasldb}1542        self.client_saslconf.sub_else_add(pattern_to_repl)1543        # restart libvirtd service on server1544        if restart_libvirtd == "yes":1545            try:1546                libvirtd_service = utils_libvirtd.Libvirtd(1547                    session=client_session)1548                if self.libvirt_ver:1549                    process.run("systemctl stop libvirtd.socket")1550                    libvirtd_service.stop()1551                    process.run("systemctl daemon-reload")1552                    process.run("systemctl start libvirtd.socket")1553                    libvirtd_service.start()1554                else:1555                    libvirtd_service.restart()1556            except (remote.LoginError, aexpect.ShellError,...
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.