From bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf Mon Sep 17 00:00:00 2001 From: James Falcon Date: Wed, 15 Dec 2021 20:16:38 -0600 Subject: Adopt Black and isort (SC-700) (#1157) Applied Black and isort, fixed any linting issues, updated tox.ini and CI. --- tests/integration_tests/instances.py | 83 ++++++++++++++++++++---------------- 1 file changed, 47 insertions(+), 36 deletions(-) (limited to 'tests/integration_tests/instances.py') 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 -- cgit v1.2.3