summaryrefslogtreecommitdiff
path: root/tests/integration_tests/instances.py
diff options
context:
space:
mode:
authorJames Falcon <james.falcon@canonical.com>2021-12-15 20:16:38 -0600
committerGitHub <noreply@github.com>2021-12-15 19:16:38 -0700
commitbae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf (patch)
tree1fbb3269fc87e39832e3286ef42eefd2b23fcd44 /tests/integration_tests/instances.py
parent2bcf4fa972fde686c2e3141c58e640640b44dd00 (diff)
downloadvyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.tar.gz
vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.zip
Adopt Black and isort (SC-700) (#1157)
Applied Black and isort, fixed any linting issues, updated tox.ini and CI.
Diffstat (limited to 'tests/integration_tests/instances.py')
-rw-r--r--tests/integration_tests/instances.py83
1 files changed, 47 insertions, 36 deletions
diff --git a/tests/integration_tests/instances.py b/tests/integration_tests/instances.py
index 8f66bf43..793f729e 100644
--- a/tests/integration_tests/instances.py
+++ b/tests/integration_tests/instances.py
@@ -1,8 +1,8 @@
# This file is part of cloud-init. See LICENSE file for license information.
-from enum import Enum
import logging
import os
import uuid
+from enum import Enum
from tempfile import NamedTemporaryFile
from pycloudlib.instance import BaseInstance
@@ -13,20 +13,21 @@ from tests.integration_tests.util import retry
try:
from typing import TYPE_CHECKING
+
if TYPE_CHECKING:
from tests.integration_tests.clouds import ( # noqa: F401
- IntegrationCloud
+ IntegrationCloud,
)
except ImportError:
pass
-log = logging.getLogger('integration_testing')
+log = logging.getLogger("integration_testing")
def _get_tmp_path():
tmp_filename = str(uuid.uuid4())
- return '/var/tmp/{}.tmp'.format(tmp_filename)
+ return "/var/tmp/{}.tmp".format(tmp_filename)
class CloudInitSource(Enum):
@@ -37,6 +38,7 @@ class CloudInitSource(Enum):
explanation of these values. If the value set there can't be parsed into
one of these values, an exception will be raised
"""
+
NONE = 1
IN_PLACE = 2
PROPOSED = 3
@@ -51,8 +53,12 @@ class CloudInitSource(Enum):
class IntegrationInstance:
- def __init__(self, cloud: 'IntegrationCloud', instance: BaseInstance,
- settings=integration_settings):
+ def __init__(
+ self,
+ cloud: "IntegrationCloud",
+ instance: BaseInstance,
+ settings=integration_settings,
+ ):
self.cloud = cloud
self.instance = instance
self.settings = settings
@@ -69,41 +75,44 @@ class IntegrationInstance:
self.instance.restart()
def execute(self, command, *, use_sudo=True) -> Result:
- if self.instance.username == 'root' and use_sudo is False:
- raise Exception('Root user cannot run unprivileged')
+ if self.instance.username == "root" and use_sudo is False:
+ raise Exception("Root user cannot run unprivileged")
return self.instance.execute(command, use_sudo=use_sudo)
def pull_file(self, remote_path, local_path):
# First copy to a temporary directory because of permissions issues
tmp_path = _get_tmp_path()
- self.instance.execute('cp {} {}'.format(str(remote_path), tmp_path))
+ self.instance.execute("cp {} {}".format(str(remote_path), tmp_path))
self.instance.pull_file(tmp_path, str(local_path))
def push_file(self, local_path, remote_path):
# First push to a temporary directory because of permissions issues
tmp_path = _get_tmp_path()
self.instance.push_file(str(local_path), tmp_path)
- self.execute('mv {} {}'.format(tmp_path, str(remote_path)))
+ self.execute("mv {} {}".format(tmp_path, str(remote_path)))
def read_from_file(self, remote_path) -> str:
- result = self.execute('cat {}'.format(remote_path))
+ result = self.execute("cat {}".format(remote_path))
if result.failed:
# TODO: Raise here whatever pycloudlib raises when it has
# a consistent error response
raise IOError(
- 'Failed reading remote file via cat: {}\n'
- 'Return code: {}\n'
- 'Stderr: {}\n'
- 'Stdout: {}'.format(
- remote_path, result.return_code,
- result.stderr, result.stdout)
+ "Failed reading remote file via cat: {}\n"
+ "Return code: {}\n"
+ "Stderr: {}\n"
+ "Stdout: {}".format(
+ remote_path,
+ result.return_code,
+ result.stderr,
+ result.stdout,
+ )
)
return result.stdout
def write_to_file(self, remote_path, contents: str):
# Writes file locally and then pushes it rather
# than writing the file directly on the instance
- with NamedTemporaryFile('w', delete=False) as tmp_file:
+ with NamedTemporaryFile("w", delete=False) as tmp_file:
tmp_file.write(contents)
try:
@@ -113,7 +122,7 @@ class IntegrationInstance:
def snapshot(self):
image_id = self.cloud.snapshot(self.instance)
- log.info('Created new image: %s', image_id)
+ log.info("Created new image: %s", image_id)
return image_id
def install_new_cloud_init(
@@ -133,10 +142,11 @@ class IntegrationInstance:
else:
raise Exception(
"Specified to install {} which isn't supported here".format(
- source)
+ source
+ )
)
- version = self.execute('cloud-init -v').split()[-1]
- log.info('Installed cloud-init version: %s', version)
+ version = self.execute("cloud-init -v").split()[-1]
+ log.info("Installed cloud-init version: %s", version)
if clean:
self.instance.clean()
if take_snapshot:
@@ -149,38 +159,39 @@ class IntegrationInstance:
@retry(tries=30, delay=1)
def install_proposed_image(self):
- log.info('Installing proposed image')
+ log.info("Installing proposed image")
assert self.execute(
'echo deb "http://archive.ubuntu.com/ubuntu '
'$(lsb_release -sc)-proposed main" >> '
- '/etc/apt/sources.list.d/proposed.list'
+ "/etc/apt/sources.list.d/proposed.list"
).ok
- assert self.execute('apt-get update -q').ok
- assert self.execute('apt-get install -qy cloud-init').ok
+ assert self.execute("apt-get update -q").ok
+ assert self.execute("apt-get install -qy cloud-init").ok
@retry(tries=30, delay=1)
def install_ppa(self):
- log.info('Installing PPA')
- assert self.execute('add-apt-repository {} -y'.format(
- self.settings.CLOUD_INIT_SOURCE)
+ log.info("Installing PPA")
+ assert self.execute(
+ "add-apt-repository {} -y".format(self.settings.CLOUD_INIT_SOURCE)
).ok
- assert self.execute('apt-get update -q').ok
- assert self.execute('apt-get install -qy cloud-init').ok
+ assert self.execute("apt-get update -q").ok
+ assert self.execute("apt-get install -qy cloud-init").ok
@retry(tries=30, delay=1)
def install_deb(self):
- log.info('Installing deb package')
+ log.info("Installing deb package")
deb_path = integration_settings.CLOUD_INIT_SOURCE
deb_name = os.path.basename(deb_path)
- remote_path = '/var/tmp/{}'.format(deb_name)
+ remote_path = "/var/tmp/{}".format(deb_name)
self.push_file(
local_path=integration_settings.CLOUD_INIT_SOURCE,
- remote_path=remote_path)
- assert self.execute('dpkg -i {path}'.format(path=remote_path)).ok
+ remote_path=remote_path,
+ )
+ assert self.execute("dpkg -i {path}".format(path=remote_path)).ok
@retry(tries=30, delay=1)
def upgrade_cloud_init(self):
- log.info('Upgrading cloud-init to latest version in archive')
+ log.info("Upgrading cloud-init to latest version in archive")
assert self.execute("apt-get update -q").ok
assert self.execute("apt-get install -qy cloud-init").ok