ssg_test_suite.virt.determine_ip() - Code Metrics - Inspection of "Merge pull request #6299 from ggbecker/readthedocs..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 42467a...9d011f )

by Matěj

created 2020-11-11 16:04 UTC

ssg_test_suite.virt.determine_ip() D

↳ Parent: tests.ssg_test_suite.virt

Complexity

Conditions

Size

Total Lines	51
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	182

Importance

Changes

Metric	Value
cc	13
eloc	36
nop	1
dl	0
loc	51
ccs	0
cts	32
cp	0
crap	182
rs	4.2
c	0
b	0
f	0

How to fix Long Method Complexity

#!/usr/bin/env python2
from __future__ import print_function

import libvirt
import logging
import time
import xml.etree.ElementTree as ET
import sys
import socket

# Needed for compatibility as there is no TimeoutError in python2.
if sys.version_info[0] < 3:
    TimeoutException = socket.timeout
else:
    TimeoutException = TimeoutError

logging.getLogger(__name__).addHandler(logging.NullHandler())


class SnapshotStack(object):
    SNAPSHOT_BASE = ("<domainsnapshot>"
                     "  <name>{name}</name>"
                     "  <description>"
                     "     Full snapshot by SSG Test Suite"
                     "  </description>"
                     "</domainsnapshot>")
    CREATE_FLAGS = libvirt.VIR_DOMAIN_SNAPSHOT_CREATE_ATOMIC
    REVERT_FLAGS = libvirt.VIR_DOMAIN_SNAPSHOT_REVERT_FORCE

    def __init__(self, domain):
        self.snapshot_stack = []
        self.domain = domain

    def create(self, snapshot_name):
        logging.debug("Creating snapshot '{0}'".format(snapshot_name))
        snapshot_xml = self.SNAPSHOT_BASE.format(name=snapshot_name)
        snapshot = self.domain.snapshotCreateXML(snapshot_xml,
                                                 self.CREATE_FLAGS)

        self.snapshot_stack.append(snapshot)
        return snapshot

    def revert_forced(self, snapshot):
        snapshot_name = snapshot.getName()
        logging.debug("Forced revert of snapshot '{0}'".format(snapshot_name))
        self.domain.revertToSnapshot(snapshot,
                                     self.REVERT_FLAGS)
        snapshot.delete()
        self.snapshot_stack.remove(snapshot)
        logging.debug('Revert successful')

    def revert(self, delete=True):
        try:
            snapshot = self.snapshot_stack.pop()
        except IndexError:
            logging.error("No snapshot in stack anymore")
        else:
            self.domain.revertToSnapshot(snapshot,
                                         self.REVERT_FLAGS)
            if delete:
                logging.debug(("Hard revert of snapshot "
                               "'{0}' successful").format(snapshot.getName()))
                snapshot.delete()
            else:
                # this is soft revert - we are keeping the snapshot for
                # another use
                logging.debug(("Soft revert of snapshot "
                               "'{0}' successful").format(snapshot.getName()))
                self.snapshot_stack.append(snapshot)

    def delete(self, snapshot=None):
        # removing snapshot from the stack without doing a revert - use
        # coupled with revert without delete
        if snapshot:
            self.snapshot_stack.remove(snapshot)
        else:
            snapshot = self.snapshot_stack.pop()
        snapshot.delete()
        logging.debug(("Snapshot '{0}' deleted "
                       "successfully").format(snapshot.getName()))

    def clear(self):
        logging.debug('Reverting all created snapshots in reverse order')
        while self.snapshot_stack:
            snapshot = self.snapshot_stack.pop()
            snapshot_name = snapshot.getName()
            logging.debug("Reverting of snapshot '{0}'".format(snapshot_name))
            self.domain.revertToSnapshot(snapshot,
                                         self.REVERT_FLAGS)
            snapshot.delete()
            logging.debug('Revert successful')
        logging.info('All snapshots reverted successfully')


def connect_domain(hypervisor, domain_name):
    conn = libvirt.open(hypervisor)
    if conn is None:
        logging.error('Failed to open connection to the hypervisor')
        return None

    try:
        dom = conn.lookupByName(domain_name)
    except libvirt.libvirtError:
        logging.error("Failed to find domain '{0}'".format(domain_name))
        return None
    return dom


def determine_ip(domain):
    GUEST_AGENT_XML = ("<channel type='unix'>"
                       "  <source mode='bind'/>"
                       "  <target type='virtio'"
                       "          name='org.qemu.guest_agent.0'"
                       "          state='connected'/>"
                       "</channel>")

    # wait for machine until it gets to RUNNING state,
    # because it isn't possible to determine IP in e.g. PAUSED state
    must_end = time.time() + 120  # wait max. 2 minutes
    while time.time() < must_end:
        if domain.state()[0] == libvirt.VIR_DOMAIN_RUNNING:
            break
        time.sleep(1)

    domain_xml = ET.fromstring(domain.XMLDesc())
    for mac_node in domain_xml.iter('mac'):
        domain_mac = mac_node.attrib['address']
        break

    logging.debug('Fetching IP address of the domain')
    try:
        ifaces = domain.interfaceAddresses(
                            libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT,
                            0)
    except libvirt.libvirtError:
        # guest agent is not connected properly
        # let's try to reattach the guest-agent device
        guest_agent_xml_string = None
        domain_xml = ET.fromstring(domain.XMLDesc())
        for guest_agent_node in domain_xml.iter('channel'):
            if guest_agent_node.attrib['type'] == 'unix':
                guest_agent_xml_string = ET.tostring(guest_agent_node)
                break
        if guest_agent_xml_string:
            domain.detachDevice(guest_agent_xml_string)
        domain.attachDevice(GUEST_AGENT_XML)
        time.sleep(1)
        # now it should be ok
        ifaces = domain.interfaceAddresses(
                            libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT,
                            0)

    # get IPv4 address of the guest
    for (name, val) in ifaces.items():
        if val['hwaddr'] == domain_mac and val['addrs']:

            for ipaddr in val['addrs']:
                if ipaddr['type'] == libvirt.VIR_IP_ADDR_TYPE_IPV4:
                    logging.debug('IP address is {0}'.format(ipaddr['addr']))
                    return ipaddr['addr']


def start_domain(domain):
    if not domain.isActive():
        logging.debug("Starting domain '{0}'".format(domain.name()))
        domain.create()
        logging.debug('Waiting 30s for domain to start')
        time.sleep(30)


def reboot_domain(domain, domain_ip, ssh_port):
    timeout = 300           # Timeout for domain shutdown and boot.
    connection_timeout = 5  # Timeout on the socket before attempting to connect.

    logging.debug("Shutting down domain '{0}'".format(domain.name()))
    domain.shutdown()

    # Wait until domain shuts down.
    logging.debug("Waiting for domain to shutdown (max. {0}s)".format(timeout))
    end_time = time.time() + timeout
    while domain.isActive():
        time.sleep(1)
        if time.time() >= end_time:
            str_err = ("Timeout reached: '{0}' domain failed to shutdown. "
                       "Forcing the shutdown...".format(domain.name()))
            logging.warning(str_err)
            domain.destroy()

    logging.debug("Starting domain '{0}'".format(domain.name()))
    domain.create()

    # Wait until SSH (on ssh_port) starts accepting TCP connections.
    logging.debug("Waiting for domain to boot (max. {0}s)".format(timeout))
    end_time = time.time() + timeout
    while True:
        try:
            ssh_socket = socket.create_connection((domain_ip, ssh_port),
                                                  timeout=connection_timeout)
        except (OSError, socket.error):
            time.sleep(1)
            if time.time() >= end_time:
                str_err = ("Timeout reached: '{0}' ({1}:{2}) domain does not "
                           "accept connections.".format(domain.name(), domain_ip, ssh_port))
                logging.debug(str_err)
                raise TimeoutException(str_err)
        else:
            ssh_socket.close()
            break


1			#!/usr/bin/env python2
2			from __future__ import print_function
3
4			import libvirt
5			import logging
6			import time
7			import xml.etree.ElementTree as ET
8			import sys
9			import socket
10
11			# Needed for compatibility as there is no TimeoutError in python2.
12			if sys.version_info[0] < 3:
13			TimeoutException = socket.timeout
14			else:
15			TimeoutException = TimeoutError
16
17			logging.getLogger(__name__).addHandler(logging.NullHandler())
18
19
20			class SnapshotStack(object):
21			SNAPSHOT_BASE = ("<domainsnapshot>"
22			" <name>{name}</name>"
23			" <description>"
24			" Full snapshot by SSG Test Suite"
25			" </description>"
26			"</domainsnapshot>")
27			CREATE_FLAGS = libvirt.VIR_DOMAIN_SNAPSHOT_CREATE_ATOMIC
28			REVERT_FLAGS = libvirt.VIR_DOMAIN_SNAPSHOT_REVERT_FORCE
29
30			def __init__(self, domain):
31			self.snapshot_stack = []
32			self.domain = domain
33
34			def create(self, snapshot_name):
35			logging.debug("Creating snapshot '{0}'".format(snapshot_name))
36			snapshot_xml = self.SNAPSHOT_BASE.format(name=snapshot_name)
37			snapshot = self.domain.snapshotCreateXML(snapshot_xml,
38			self.CREATE_FLAGS)
39
40			self.snapshot_stack.append(snapshot)
41			return snapshot
42
43			def revert_forced(self, snapshot):
44			snapshot_name = snapshot.getName()
45			logging.debug("Forced revert of snapshot '{0}'".format(snapshot_name))
46			self.domain.revertToSnapshot(snapshot,
47			self.REVERT_FLAGS)
48			snapshot.delete()
49			self.snapshot_stack.remove(snapshot)
50			logging.debug('Revert successful')
51
52			def revert(self, delete=True):
53			try:
54			snapshot = self.snapshot_stack.pop()
55			except IndexError:
56			logging.error("No snapshot in stack anymore")
57			else:
58			self.domain.revertToSnapshot(snapshot,
59			self.REVERT_FLAGS)
60			if delete:
61			logging.debug(("Hard revert of snapshot "
62			"'{0}' successful").format(snapshot.getName()))
63			snapshot.delete()
64			else:
65			# this is soft revert - we are keeping the snapshot for
66			# another use
67			logging.debug(("Soft revert of snapshot "
68			"'{0}' successful").format(snapshot.getName()))
69			self.snapshot_stack.append(snapshot)
70
71			def delete(self, snapshot=None):
72			# removing snapshot from the stack without doing a revert - use
73			# coupled with revert without delete
74			if snapshot:
75			self.snapshot_stack.remove(snapshot)
76			else:
77			snapshot = self.snapshot_stack.pop()
78			snapshot.delete()
79			logging.debug(("Snapshot '{0}' deleted "
80			"successfully").format(snapshot.getName()))
81
82			def clear(self):
83			logging.debug('Reverting all created snapshots in reverse order')
84			while self.snapshot_stack:
85			snapshot = self.snapshot_stack.pop()
86			snapshot_name = snapshot.getName()
87			logging.debug("Reverting of snapshot '{0}'".format(snapshot_name))
88			self.domain.revertToSnapshot(snapshot,
89			self.REVERT_FLAGS)
90			snapshot.delete()
91			logging.debug('Revert successful')
92			logging.info('All snapshots reverted successfully')
93
94
95			def connect_domain(hypervisor, domain_name):
96			conn = libvirt.open(hypervisor)
97			if conn is None:
98			logging.error('Failed to open connection to the hypervisor')
99			return None
100
101			try:
102			dom = conn.lookupByName(domain_name)
103			except libvirt.libvirtError:
104			logging.error("Failed to find domain '{0}'".format(domain_name))
105			return None
106			return dom
107
108
109			def determine_ip(domain):
110			GUEST_AGENT_XML = ("<channel type='unix'>"
111			" <source mode='bind'/>"
112			" <target type='virtio'"
113			" name='org.qemu.guest_agent.0'"
114			" state='connected'/>"
115			"</channel>")
116
117			# wait for machine until it gets to RUNNING state,
118			# because it isn't possible to determine IP in e.g. PAUSED state
119			must_end = time.time() + 120 # wait max. 2 minutes
120			while time.time() < must_end:
121			if domain.state()[0] == libvirt.VIR_DOMAIN_RUNNING:
122			break
123			time.sleep(1)
124
125			domain_xml = ET.fromstring(domain.XMLDesc())
126			for mac_node in domain_xml.iter('mac'):
127			domain_mac = mac_node.attrib['address']
128			break
129
130			logging.debug('Fetching IP address of the domain')
131			try:
132			ifaces = domain.interfaceAddresses(
133			libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT,
134			0)
135			except libvirt.libvirtError:
136			# guest agent is not connected properly
137			# let's try to reattach the guest-agent device
138			guest_agent_xml_string = None
139			domain_xml = ET.fromstring(domain.XMLDesc())
140			for guest_agent_node in domain_xml.iter('channel'):
141			if guest_agent_node.attrib['type'] == 'unix':
142			guest_agent_xml_string = ET.tostring(guest_agent_node)
143			break
144			if guest_agent_xml_string:
145			domain.detachDevice(guest_agent_xml_string)
146			domain.attachDevice(GUEST_AGENT_XML)
147			time.sleep(1)
148			# now it should be ok
149			ifaces = domain.interfaceAddresses(
150			libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT,
151			0)
152
153			# get IPv4 address of the guest
154			for (name, val) in ifaces.items():
155			if val['hwaddr'] == domain_mac and val['addrs']:
			0 ignored issues – show introduced 2018-05-31 12:10 UTC by Report Bug Copy Issue Report The variable `domain_mac` does not seem to be defined for all execution paths. Loading history...
156			for ipaddr in val['addrs']:
157			if ipaddr['type'] == libvirt.VIR_IP_ADDR_TYPE_IPV4:
158			logging.debug('IP address is {0}'.format(ipaddr['addr']))
159			return ipaddr['addr']
160
161
162			def start_domain(domain):
163			if not domain.isActive():
164			logging.debug("Starting domain '{0}'".format(domain.name()))
165			domain.create()
166			logging.debug('Waiting 30s for domain to start')
167			time.sleep(30)
168
169
170			def reboot_domain(domain, domain_ip, ssh_port):
171			timeout = 300 # Timeout for domain shutdown and boot.
172			connection_timeout = 5 # Timeout on the socket before attempting to connect.
173
174			logging.debug("Shutting down domain '{0}'".format(domain.name()))
175			domain.shutdown()
176
177			# Wait until domain shuts down.
178			logging.debug("Waiting for domain to shutdown (max. {0}s)".format(timeout))
179			end_time = time.time() + timeout
180			while domain.isActive():
181			time.sleep(1)
182			if time.time() >= end_time:
183			str_err = ("Timeout reached: '{0}' domain failed to shutdown. "
184			"Forcing the shutdown...".format(domain.name()))
185			logging.warning(str_err)
186			domain.destroy()
187
188			logging.debug("Starting domain '{0}'".format(domain.name()))
189			domain.create()
190
191			# Wait until SSH (on ssh_port) starts accepting TCP connections.
192			logging.debug("Waiting for domain to boot (max. {0}s)".format(timeout))
193			end_time = time.time() + timeout
194			while True:
195			try:
196			ssh_socket = socket.create_connection((domain_ip, ssh_port),
197			timeout=connection_timeout)
198			except (OSError, socket.error):
199			time.sleep(1)
200			if time.time() >= end_time:
201			str_err = ("Timeout reached: '{0}' ({1}:{2}) domain does not "
202			"accept connections.".format(domain.name(), domain_ip, ssh_port))
203			logging.debug(str_err)
204			raise TimeoutException(str_err)
205			else:
206			ssh_socket.close()
207			break
208

ComplianceAsCode / content

Push — master ( 42467a...9d011f )

ssg_test_suite.virt.determine_ip() D

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like