summaryrefslogtreecommitdiff
path: root/tests/integration_tests/modules/test_disk_setup.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/modules/test_disk_setup.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/modules/test_disk_setup.py')
-rw-r--r--tests/integration_tests/modules/test_disk_setup.py76
1 files changed, 40 insertions, 36 deletions
diff --git a/tests/integration_tests/modules/test_disk_setup.py b/tests/integration_tests/modules/test_disk_setup.py
index 9c9edc46..22277331 100644
--- a/tests/integration_tests/modules/test_disk_setup.py
+++ b/tests/integration_tests/modules/test_disk_setup.py
@@ -1,25 +1,29 @@
import json
import os
-import pytest
from uuid import uuid4
+
+import pytest
from pycloudlib.lxd.instance import LXDInstance
from cloudinit.subp import subp
from tests.integration_tests.instances import IntegrationInstance
from tests.integration_tests.util import verify_clean_log
-DISK_PATH = '/tmp/test_disk_setup_{}'.format(uuid4())
+DISK_PATH = "/tmp/test_disk_setup_{}".format(uuid4())
def setup_and_mount_lxd_disk(instance: LXDInstance):
- subp('lxc config device add {} test-disk-setup-disk disk source={}'.format(
- instance.name, DISK_PATH).split())
+ subp(
+ "lxc config device add {} test-disk-setup-disk disk source={}".format(
+ instance.name, DISK_PATH
+ ).split()
+ )
@pytest.yield_fixture
def create_disk():
# 640k should be enough for anybody
- subp('dd if=/dev/zero of={} bs=1k count=640'.format(DISK_PATH).split())
+ subp("dd if=/dev/zero of={} bs=1k count=640".format(DISK_PATH).split())
yield
os.remove(DISK_PATH)
@@ -54,21 +58,21 @@ class TestDeviceAliases:
"""Test devices aliases work on disk setup/mount"""
def test_device_alias(self, create_disk, client: IntegrationInstance):
- log = client.read_from_file('/var/log/cloud-init.log')
+ log = client.read_from_file("/var/log/cloud-init.log")
assert (
- "updated disk_setup device entry 'my_alias' to '/dev/sdb'"
- ) in log
- assert 'changed my_alias.1 => /dev/sdb1' in log
- assert 'changed my_alias.2 => /dev/sdb2' in log
+ "updated disk_setup device entry 'my_alias' to '/dev/sdb'" in log
+ )
+ assert "changed my_alias.1 => /dev/sdb1" in log
+ assert "changed my_alias.2 => /dev/sdb2" in log
verify_clean_log(log)
- lsblk = json.loads(client.execute('lsblk --json'))
- sdb = [x for x in lsblk['blockdevices'] if x['name'] == 'sdb'][0]
- assert len(sdb['children']) == 2
- assert sdb['children'][0]['name'] == 'sdb1'
- assert sdb['children'][0]['mountpoint'] == '/mnt1'
- assert sdb['children'][1]['name'] == 'sdb2'
- assert sdb['children'][1]['mountpoint'] == '/mnt2'
+ lsblk = json.loads(client.execute("lsblk --json"))
+ sdb = [x for x in lsblk["blockdevices"] if x["name"] == "sdb"][0]
+ assert len(sdb["children"]) == 2
+ assert sdb["children"][0]["name"] == "sdb1"
+ assert sdb["children"][0]["mountpoint"] == "/mnt1"
+ assert sdb["children"][1]["name"] == "sdb2"
+ assert sdb["children"][1]["mountpoint"] == "/mnt2"
PARTPROBE_USERDATA = """\
@@ -121,13 +125,13 @@ class TestPartProbeAvailability:
def _verify_first_disk_setup(self, client, log):
verify_clean_log(log)
- lsblk = json.loads(client.execute('lsblk --json'))
- sdb = [x for x in lsblk['blockdevices'] if x['name'] == 'sdb'][0]
- assert len(sdb['children']) == 2
- assert sdb['children'][0]['name'] == 'sdb1'
- assert sdb['children'][0]['mountpoint'] == '/mnt1'
- assert sdb['children'][1]['name'] == 'sdb2'
- assert sdb['children'][1]['mountpoint'] == '/mnt2'
+ lsblk = json.loads(client.execute("lsblk --json"))
+ sdb = [x for x in lsblk["blockdevices"] if x["name"] == "sdb"][0]
+ assert len(sdb["children"]) == 2
+ assert sdb["children"][0]["name"] == "sdb1"
+ assert sdb["children"][0]["mountpoint"] == "/mnt1"
+ assert sdb["children"][1]["name"] == "sdb2"
+ assert sdb["children"][1]["mountpoint"] == "/mnt2"
# Not bionic or xenial because the LXD agent gets in the way of us
# changing the userdata
@@ -148,13 +152,13 @@ class TestPartProbeAvailability:
with a warning and a traceback. When partprobe is in use, everything
should work successfully.
"""
- log = client.read_from_file('/var/log/cloud-init.log')
+ log = client.read_from_file("/var/log/cloud-init.log")
self._verify_first_disk_setup(client, log)
# Update our userdata and cloud.cfg to mount then perform new disk
# setup
client.write_to_file(
- '/var/lib/cloud/seed/nocloud-net/user-data',
+ "/var/lib/cloud/seed/nocloud-net/user-data",
UPDATED_PARTPROBE_USERDATA,
)
client.execute(
@@ -162,17 +166,17 @@ class TestPartProbeAvailability:
"/etc/cloud/cloud.cfg"
)
- client.execute('cloud-init clean --logs')
+ client.execute("cloud-init clean --logs")
client.restart()
# Assert new setup works as expected
verify_clean_log(log)
- lsblk = json.loads(client.execute('lsblk --json'))
- sdb = [x for x in lsblk['blockdevices'] if x['name'] == 'sdb'][0]
- assert len(sdb['children']) == 1
- assert sdb['children'][0]['name'] == 'sdb1'
- assert sdb['children'][0]['mountpoint'] == '/mnt3'
+ lsblk = json.loads(client.execute("lsblk --json"))
+ sdb = [x for x in lsblk["blockdevices"] if x["name"] == "sdb"][0]
+ assert len(sdb["children"]) == 1
+ assert sdb["children"][0]["name"] == "sdb1"
+ assert sdb["children"][0]["mountpoint"] == "/mnt3"
def test_disk_setup_no_partprobe(
self, create_disk, client: IntegrationInstance
@@ -180,11 +184,11 @@ class TestPartProbeAvailability:
"""Ensure disk setup still works as expected without partprobe."""
# We can't do this part in a bootcmd because the path has already
# been found by the time we get to the bootcmd
- client.execute('rm $(which partprobe)')
- client.execute('cloud-init clean --logs')
+ client.execute("rm $(which partprobe)")
+ client.execute("cloud-init clean --logs")
client.restart()
- log = client.read_from_file('/var/log/cloud-init.log')
+ log = client.read_from_file("/var/log/cloud-init.log")
self._verify_first_disk_setup(client, log)
- assert 'partprobe' not in log
+ assert "partprobe" not in log