tests.install_vm.parse_args() - Code Metrics - Inspection of "Merge pull request #7780 from matusmarhefka/instal..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 666d5f...4395a0 )

by Matěj

created 2021-11-04 12:48 UTC

tests.install_vm.parse_args() B

↳ Parent: tests.install_vm

Complexity

Conditions

Size

Total Lines	102
Code Lines	84

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	84
nop	0
dl	0
loc	102
rs	7.5236
c	0
b	0
f	0

How to fix Long Method

#!/usr/bin/env python

import argparse
import os
import sys
import subprocess
import time


def parse_args():
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--libvirt",
        dest="libvirt",
        default="qemu:///session",
        help="What hypervisor should be used when installing VM."
    )
    parser.add_argument(
        "--kickstart",
        dest="kickstart",
        default="kickstarts/test_suite.cfg",
        help="Path to a kickstart file for installation of a VM."
    )
    parser.add_argument(
        "--distro",
        dest="distro",
        required=True,
        choices=['fedora', 'rhel7', 'centos7', 'rhel8', 'rhel9'],
        help="What distribution to install."
    )
    parser.add_argument(
        "--domain",
        dest="domain",
        required=True,
        help="What name should the new domain have."
    )
    parser.add_argument(
        "--disk-dir",
        dest="disk_dir",
        default=None,
        help="Location of the VM qcow2 file."
    )
    parser.add_argument(
        "--ram",
        dest="ram",
        default=3072,
        type=int,
        help="Amount of RAM configured for the VM."
    )
    parser.add_argument(
        "--cpu",
        dest="cpu",
        default=2,
        type=int,
        help="Number of CPU cores configured for the VM."
    )
    parser.add_argument(
        "--network",
        dest="network",
        help="Network type/spec, ie. bridge=br0 or network=name."
    )
    parser.add_argument(
        "--disk",
        dest="disk",
        help="Disk type/spec, ie. pool=MyPool,bus=sata,cache=unsafe."
    )
    parser.add_argument(
        "--url",
        dest="url",
        default=None,
        help="URL to an installation tree on a remote server."
    )
    parser.add_argument(
        "--extra-repo",
        dest="extra_repo",
        default=None,
        help="URL to an extra repository to be used during installation (e.g. AppStream)."
    )
    parser.add_argument(
        "--dry",
        dest="dry",
        action="store_true",
        help="Print command line instead of triggering command."
    )
    parser.add_argument(
        "--ssh-pubkey",
        dest="ssh_pubkey",
        default=None,
        help="Path to an SSH public key which will be used to access the VM."
    )
    parser.add_argument(
        "--uefi",
        dest="uefi",
        choices=['secureboot', 'normal'],
        help="Perform UEFI based installation, optionally with secure boot support."
    )
    parser.add_argument(
        "--install-gui",
        dest="install_gui",
        action='store_true',
        help="Perform a GUI installation (default is installation without GUI)."
    )
    parser.add_argument(
        "--console",
        dest="console",
        action='store_true',
        help="Connect to a serial console of the VM (to monitor installation progress)."
    )

    return parser.parse_args()


def wait_vm_not_running(domain):
    timeout = 300

    print("Waiting for {0} VM to shutdown (max. {1}s)".format(domain, timeout))
    end_time = time.time() + timeout
    try:
        while True:
            time.sleep(5)
            if subprocess.getoutput("virsh domstate {0}".format(domain)).rstrip() != "running":
                return
            if time.time() >= end_time:
                print("Timeout reached: {0} VM failed to shutdown, cancelling wait."
                      .format(domain))
                return
    except KeyboardInterrupt:
        print("Interrupted, cancelling wait.")
        return


def main():
    data = parse_args()
    username = ""
    try:
        username = os.environ["SUDO_USER"]
    except KeyError:
        pass
    home_dir = os.path.expanduser('~' + username)

    if not data.url:
        if data.distro == "fedora":
            data.url = "https://download.fedoraproject.org/pub/fedora/linux/releases/34/Everything/x86_64/os"
        elif data.distro == "centos7":
            data.url = "http://mirror.centos.org/centos/7/os/x86_64"
    if not data.url:
        sys.stderr.write("For the '{}' distro the `--url` option needs to be provided.\n".format(data.distro))
        return 1

    if not data.ssh_pubkey:
        data.ssh_pubkey = home_dir + "/.ssh/id_rsa.pub"
    if not os.path.isfile(data.ssh_pubkey):
        sys.stderr.write("Error: SSH public key not found at {0}\n".format(data.ssh_pubkey))
        sys.stderr.write("You can use the `--ssh-pubkey` to specify which key should be used.\n")
        return 1
    with open(data.ssh_pubkey) as f:
        pub_key = f.readline().rstrip()
    print("Using SSH public key from file: {0}".format(data.ssh_pubkey))
    print("Using hypervisor: {0}".format(data.libvirt))

    if data.disk:
        data.disk_spec = data.disk
    elif data.disk_dir:
        disk_path = os.path.join(data.disk_dir, data.domain) + ".qcow2"
        print("Location of VM disk: {0}".format(disk_path))
        data.disk_spec = "path={0},format=qcow2,size=20".format(disk_path)
    else:
        data.disk_spec = "size=20,format=qcow2"

    data.ks_basename = os.path.basename(data.kickstart)

    tmp_kickstart = "/tmp/" + data.ks_basename
    with open(data.kickstart) as infile, open(tmp_kickstart, "w") as outfile:
        content = infile.read()
        content = content.replace("&&HOST_PUBLIC_KEY&&", pub_key)
        if not data.distro == "fedora":
            content = content.replace("&&YUM_REPO_URL&&", data.url)
        if data.extra_repo:
            # extra repository
            repo_cmd = "repo --name=extra-repository --baseurl={}".format(data.extra_repo)
            content = content.replace("&&YUM_EXTRA_REPO&&", repo_cmd)
            content = content.replace("&&YUM_EXTRA_REPO_URL&&", data.extra_repo)
        else:
            content = content.replace("&&YUM_EXTRA_REPO&&", "")
        if data.uefi:
            content = content.replace(
                "part /boot --fstype=xfs --size=512",
                "part /boot --fstype=xfs --size=312\npart /boot/efi --fstype=efi --size=200"
            )
        if data.install_gui:
            gui_group="\n%packages\n@^graphical-server-environment\n"
            if data.distro == "fedora":
                gui_group="\n%packages\n@^Fedora Workstation\n"
            content = content.replace("\n%packages\n", gui_group)
            data.graphics_opt = "vnc"
            data.inst_opt = "inst.graphical"
        else:
            data.graphics_opt = "none"
            data.inst_opt = "inst.cmdline"
        outfile.write(content)
    data.kickstart = tmp_kickstart
    print("Using kickstart file: {0}".format(data.kickstart))

    if not data.network:
        if data.libvirt == "qemu:///system":
            data.network = "network=default"
        else:
            data.network = "bridge=virbr0"
    if data.console:
        data.wait_opt = 0
    else:
        data.wait_opt = -1

    # The kernel option 'net.ifnames=0' is used to disable predictable network
    # interface names, for more details see:
    # https://www.freedesktop.org/wiki/Software/systemd/PredictableNetworkInterfaceNames/
    command = 'virt-install --connect={libvirt} --name={domain} --memory={ram} --vcpus={cpu} --network {network} --disk {disk_spec} --initrd-inject={kickstart} --extra-args="inst.ks=file:/{ks_basename} {inst_opt} ksdevice=eth0 net.ifnames=0 console=ttyS0,115200" --serial pty --graphics={graphics_opt} --noautoconsole --rng /dev/random --wait={wait_opt} --location={url}'.format(**data.__dict__)
    if data.uefi == "normal":
        command = command+" --boot uefi"
    if data.uefi == "secureboot":
        command = command + " --boot uefi,loader_secure=yes,\
loader=/usr/share/edk2/ovmf/OVMF_CODE.secboot.fd,\
nvram_template=/usr/share/edk2/ovmf/OVMF_VARS.secboot.fd --features smm=on"

    if data.dry:
        print("\nThe following command would be used for the VM installation:")
        print(command)
    else:
        os.system(command)
        if data.console:
            os.system("unbuffer virsh console {0}".format(data.domain))
            wait_vm_not_running(data.domain)
            os.system("virsh start {0}".format(data.domain))

    print("\nTo determine the IP address of the {0} VM use:".format(data.domain))
    if data.libvirt == "qemu:///system":
        print("  sudo virsh domifaddr {0}\n".format(data.domain))
    else:
        # command evaluation in fish shell is simply surrounded by
        # parenthesis for example: (echo foo). In other shells you
        # need to prepend the $ symbol as: $(echo foo)
        from os import environ
        print("  arp -n | grep {0}(virsh -q domiflist {1} | awk '{{print $5}}')\n".format('' if 'fish' == environ['SHELL'][-4:] else '$', data.domain))

    print("To connect to the {0} VM use:\n  ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@IP\n".format(data.domain))
    print("To connect to the VM serial console, use:\n  virsh console {0}\n".format(data.domain))
    print("If you have used the `--ssh-pubkey` also add '-o IdentityFile=PATH_TO_PRIVATE_KEY' option to your ssh command and export the SSH_ADDITIONAL_OPTIONS='-o IdentityFile=PATH_TO_PRIVATE_KEY' before running the SSG Test Suite.")

    if data.libvirt == "qemu:///system":
        print("\nIMPORTANT: When running SSG Test Suite use `sudo -E` to make sure that your SSH key is used.")


if __name__ == '__main__':
    main()


1			#!/usr/bin/env python
2
3			import argparse
4			import os
5			import sys
6			import subprocess
7			import time
8
9
10			def parse_args():
11			parser = argparse.ArgumentParser()
12
13			parser.add_argument(
14			"--libvirt",
15			dest="libvirt",
16			default="qemu:///session",
17			help="What hypervisor should be used when installing VM."
18			)
19			parser.add_argument(
20			"--kickstart",
21			dest="kickstart",
22			default="kickstarts/test_suite.cfg",
23			help="Path to a kickstart file for installation of a VM."
24			)
25			parser.add_argument(
26			"--distro",
27			dest="distro",
28			required=True,
29			choices=['fedora', 'rhel7', 'centos7', 'rhel8', 'rhel9'],
30			help="What distribution to install."
31			)
32			parser.add_argument(
33			"--domain",
34			dest="domain",
35			required=True,
36			help="What name should the new domain have."
37			)
38			parser.add_argument(
39			"--disk-dir",
40			dest="disk_dir",
41			default=None,
42			help="Location of the VM qcow2 file."
43			)
44			parser.add_argument(
45			"--ram",
46			dest="ram",
47			default=3072,
48			type=int,
49			help="Amount of RAM configured for the VM."
50			)
51			parser.add_argument(
52			"--cpu",
53			dest="cpu",
54			default=2,
55			type=int,
56			help="Number of CPU cores configured for the VM."
57			)
58			parser.add_argument(
59			"--network",
60			dest="network",
61			help="Network type/spec, ie. bridge=br0 or network=name."
62			)
63			parser.add_argument(
64			"--disk",
65			dest="disk",
66			help="Disk type/spec, ie. pool=MyPool,bus=sata,cache=unsafe."
67			)
68			parser.add_argument(
69			"--url",
70			dest="url",
71			default=None,
72			help="URL to an installation tree on a remote server."
73			)
74			parser.add_argument(
75			"--extra-repo",
76			dest="extra_repo",
77			default=None,
78			help="URL to an extra repository to be used during installation (e.g. AppStream)."
79			)
80			parser.add_argument(
81			"--dry",
82			dest="dry",
83			action="store_true",
84			help="Print command line instead of triggering command."
85			)
86			parser.add_argument(
87			"--ssh-pubkey",
88			dest="ssh_pubkey",
89			default=None,
90			help="Path to an SSH public key which will be used to access the VM."
91			)
92			parser.add_argument(
93			"--uefi",
94			dest="uefi",
95			choices=['secureboot', 'normal'],
96			help="Perform UEFI based installation, optionally with secure boot support."
97			)
98			parser.add_argument(
99			"--install-gui",
100			dest="install_gui",
101			action='store_true',
102			help="Perform a GUI installation (default is installation without GUI)."
103			)
104			parser.add_argument(
105			"--console",
106			dest="console",
107			action='store_true',
108			help="Connect to a serial console of the VM (to monitor installation progress)."
109			)
110
111			return parser.parse_args()
112
113
114			def wait_vm_not_running(domain):
115			timeout = 300
116
117			print("Waiting for {0} VM to shutdown (max. {1}s)".format(domain, timeout))
118			end_time = time.time() + timeout
119			try:
120			while True:
121			time.sleep(5)
122			if subprocess.getoutput("virsh domstate {0}".format(domain)).rstrip() != "running":
123			return
124			if time.time() >= end_time:
125			print("Timeout reached: {0} VM failed to shutdown, cancelling wait."
126			.format(domain))
127			return
128			except KeyboardInterrupt:
129			print("Interrupted, cancelling wait.")
130			return
131
132
133			def main():
134			data = parse_args()
135			username = ""
136			try:
137			username = os.environ["SUDO_USER"]
138			except KeyError:
139			pass
140			home_dir = os.path.expanduser('~' + username)
141
142			if not data.url:
143			if data.distro == "fedora":
144			data.url = "https://download.fedoraproject.org/pub/fedora/linux/releases/34/Everything/x86_64/os"
145			elif data.distro == "centos7":
146			data.url = "http://mirror.centos.org/centos/7/os/x86_64"
147			if not data.url:
148			sys.stderr.write("For the '{}' distro the `--url` option needs to be provided.\n".format(data.distro))
149			return 1
150
151			if not data.ssh_pubkey:
152			data.ssh_pubkey = home_dir + "/.ssh/id_rsa.pub"
153			if not os.path.isfile(data.ssh_pubkey):
154			sys.stderr.write("Error: SSH public key not found at {0}\n".format(data.ssh_pubkey))
155			sys.stderr.write("You can use the `--ssh-pubkey` to specify which key should be used.\n")
156			return 1
157			with open(data.ssh_pubkey) as f:
158			pub_key = f.readline().rstrip()
159			print("Using SSH public key from file: {0}".format(data.ssh_pubkey))
160			print("Using hypervisor: {0}".format(data.libvirt))
161
162			if data.disk:
163			data.disk_spec = data.disk
164			elif data.disk_dir:
165			disk_path = os.path.join(data.disk_dir, data.domain) + ".qcow2"
166			print("Location of VM disk: {0}".format(disk_path))
167			data.disk_spec = "path={0},format=qcow2,size=20".format(disk_path)
168			else:
169			data.disk_spec = "size=20,format=qcow2"
170
171			data.ks_basename = os.path.basename(data.kickstart)
172
173			tmp_kickstart = "/tmp/" + data.ks_basename
174			with open(data.kickstart) as infile, open(tmp_kickstart, "w") as outfile:
175			content = infile.read()
176			content = content.replace("&&HOST_PUBLIC_KEY&&", pub_key)
177			if not data.distro == "fedora":
178			content = content.replace("&&YUM_REPO_URL&&", data.url)
179			if data.extra_repo:
180			# extra repository
181			repo_cmd = "repo --name=extra-repository --baseurl={}".format(data.extra_repo)
182			content = content.replace("&&YUM_EXTRA_REPO&&", repo_cmd)
183			content = content.replace("&&YUM_EXTRA_REPO_URL&&", data.extra_repo)
184			else:
185			content = content.replace("&&YUM_EXTRA_REPO&&", "")
186			if data.uefi:
187			content = content.replace(
188			"part /boot --fstype=xfs --size=512",
189			"part /boot --fstype=xfs --size=312\npart /boot/efi --fstype=efi --size=200"
190			)
191			if data.install_gui:
192			gui_group="\n%packages\n@^graphical-server-environment\n"
193			if data.distro == "fedora":
194			gui_group="\n%packages\n@^Fedora Workstation\n"
195			content = content.replace("\n%packages\n", gui_group)
196			data.graphics_opt = "vnc"
197			data.inst_opt = "inst.graphical"
198			else:
199			data.graphics_opt = "none"
200			data.inst_opt = "inst.cmdline"
201			outfile.write(content)
202			data.kickstart = tmp_kickstart
203			print("Using kickstart file: {0}".format(data.kickstart))
204
205			if not data.network:
206			if data.libvirt == "qemu:///system":
207			data.network = "network=default"
208			else:
209			data.network = "bridge=virbr0"
210			if data.console:
211			data.wait_opt = 0
212			else:
213			data.wait_opt = -1
214
215			# The kernel option 'net.ifnames=0' is used to disable predictable network
216			# interface names, for more details see:
217			# https://www.freedesktop.org/wiki/Software/systemd/PredictableNetworkInterfaceNames/
218			command = 'virt-install --connect={libvirt} --name={domain} --memory={ram} --vcpus={cpu} --network {network} --disk {disk_spec} --initrd-inject={kickstart} --extra-args="inst.ks=file:/{ks_basename} {inst_opt} ksdevice=eth0 net.ifnames=0 console=ttyS0,115200" --serial pty --graphics={graphics_opt} --noautoconsole --rng /dev/random --wait={wait_opt} --location={url}'.format(**data.__dict__)
219			if data.uefi == "normal":
220			command = command+" --boot uefi"
221			if data.uefi == "secureboot":
222			command = command + " --boot uefi,loader_secure=yes,\
223			loader=/usr/share/edk2/ovmf/OVMF_CODE.secboot.fd,\
224			nvram_template=/usr/share/edk2/ovmf/OVMF_VARS.secboot.fd --features smm=on"
225
226			if data.dry:
227			print("\nThe following command would be used for the VM installation:")
228			print(command)
229			else:
230			os.system(command)
231			if data.console:
232			os.system("unbuffer virsh console {0}".format(data.domain))
233			wait_vm_not_running(data.domain)
234			os.system("virsh start {0}".format(data.domain))
235
236			print("\nTo determine the IP address of the {0} VM use:".format(data.domain))
237			if data.libvirt == "qemu:///system":
238			print(" sudo virsh domifaddr {0}\n".format(data.domain))
239			else:
240			# command evaluation in fish shell is simply surrounded by
241			# parenthesis for example: (echo foo). In other shells you
242			# need to prepend the $ symbol as: $(echo foo)
243			from os import environ
244			print(" arp -n \| grep {0}(virsh -q domiflist {1} \| awk '{{print $5}}')\n".format('' if 'fish' == environ['SHELL'][-4:] else '$', data.domain))
245
246			print("To connect to the {0} VM use:\n ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@IP\n".format(data.domain))
247			print("To connect to the VM serial console, use:\n virsh console {0}\n".format(data.domain))
248			print("If you have used the `--ssh-pubkey` also add '-o IdentityFile=PATH_TO_PRIVATE_KEY' option to your ssh command and export the SSH_ADDITIONAL_OPTIONS='-o IdentityFile=PATH_TO_PRIVATE_KEY' before running the SSG Test Suite.")
249
250			if data.libvirt == "qemu:///system":
251			print("\nIMPORTANT: When running SSG Test Suite use `sudo -E` to make sure that your SSH key is used.")
252
253
254			if __name__ == '__main__':
255			main()
256

ComplianceAsCode / content

Push — master ( 666d5f...4395a0 )

tests.install_vm.parse_args() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like