From d9ad1c2d059f271bbd8f2ed548bb3c9782d8833e Mon Sep 17 00:00:00 2001 From: Tony Asleson Date: Tue, 28 Apr 2026 13:18:43 -0500 Subject: [PATCH] tests: replace hardcoded test pool names with unique checksummed names Test resources previously used generic names (test_vg0, pool0, pool1) that could theoretically collide with real resources and made it impossible to verify ownership before cleanup. Replace with dynamically generated names in the format: snapm___<8 random hex>_<4 hex SHA-256 checksum> The embedded checksum allows the cleanup script to validate that a resource was created by the test suite before destroying it. A standalone bash CLI tool (snapm-test-name) is provided for developers to manually generate and validate names. Resolves: #978 Assisted-by: Claude --- tests/_util.py | 23 ++++++++++-- tests/bin/cleanup.sh | 63 ++++++++++++++++++++++++--------- tests/bin/snapm-test-name | 73 +++++++++++++++++++++++++++++++++++++++ tests/test_boot.py | 4 +-- tests/test_lvm2.py | 22 ++++++------ tests/test_manager.py | 4 +-- tests/test_mounts.py | 6 ++-- tests/test_stratis.py | 36 +++++++++---------- 8 files changed, 175 insertions(+), 56 deletions(-) create mode 100755 tests/bin/snapm-test-name diff --git a/tests/_util.py b/tests/_util.py index b3c0d5f2..217bf1f8 100644 --- a/tests/_util.py +++ b/tests/_util.py @@ -1,3 +1,4 @@ +import hashlib import tempfile import subprocess import logging @@ -6,6 +7,22 @@ log = logging.getLogger() + +def generate_test_name(subsystem, context): + random_hex = os.urandom(4).hex() + prefix = f"snapm_{subsystem}_{context}_{random_hex}" + checksum = hashlib.sha256(prefix.encode("ascii")).hexdigest()[:4] + return f"{prefix}_{checksum}" + + +def validate_test_name(name): + if len(name) < 6 or name[-5] != "_": + return False + prefix = name[:-5] + checksum = name[-4:] + return hashlib.sha256(prefix.encode("ascii")).hexdigest()[:4] == checksum + + _LOSETUP_CMD = "losetup" _UDEVADM = "udevadm" @@ -27,9 +44,9 @@ _MKFS_EXT4_CMD = "mkfs.ext4" # LVM2 -_VG_NAME = "test_vg0" +_VG_NAME = generate_test_name("lm", "vg") -_THIN_POOL_NAME = "pool0" +_THIN_POOL_NAME = generate_test_name("lm", "pool") # 1GiB _LV_SIZE = 1024**3 @@ -52,7 +69,7 @@ _NAME_ARG = "--name" _SIZE_ARG = "--size" -_POOL_NAME = "pool1" +_POOL_NAME = generate_test_name("st", "pool") _FS_SIZE = "1GiB" diff --git a/tests/bin/cleanup.sh b/tests/bin/cleanup.sh index 9c7202ba..076bd034 100755 --- a/tests/bin/cleanup.sh +++ b/tests/bin/cleanup.sh @@ -22,25 +22,54 @@ if [ -f /tmp/fstab ]; then rm -f /tmp/fstab fi -if [ -d /dev/test_vg0 ]; then - export LVM_SYSTEM_DIR="$PWD/tests/lvm" - vgremove --force --yes test_vg0 &> /dev/null || echo Failed to clean up test_vg0 || true -fi +_validate_snapm_name() { + local name="$1" + if [ ${#name} -lt 6 ]; then + return 1 + fi + local sep="${name: -5:1}" + if [ "$sep" != "_" ]; then + return 1 + fi + local prefix="${name:0:${#name}-5}" + local embedded="${name: -4}" + local computed + computed=$(printf '%s' "$prefix" | sha256sum | cut -c1-4) + [ "$embedded" = "$computed" ] +} -if [ -d /dev/stratis/pool1 ]; then - STRATIS_FILESYSTEMS=$(stratis filesystem list pool1 | awk '/pool1/{print $2}') - for FS in $STRATIS_FILESYSTEMS; do - if ! FAIL=$(stratis filesystem destroy pool1 "$FS" 2>&1); then - printf 'Failed to clean up pool1 %s: %s\n' "$FS" "$FAIL" - fi - done -fi +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" -if stratis pool list | grep -qE '^pool1([[:space:]]|$)'; then - if ! FAIL=$(stratis pool destroy pool1 2>&1); then - printf 'Failed to clean up pool1: %s\n' "$FAIL" - fi -fi +# Clean up LVM VGs matching snapm_lm_* +export LVM_SYSTEM_DIR="$REPO_ROOT/tests/lvm" +for vg_dir in /dev/snapm_lm_*; do + [ -d "$vg_dir" ] || continue + vg_name=$(basename "$vg_dir") + if _validate_snapm_name "$vg_name"; then + vgremove --force --yes "$vg_name" &> /dev/null \ + || printf 'Failed to clean up %s\n' "$vg_name" + fi +done + +# Clean up Stratis pools matching snapm_st_* +for pool_dir in /dev/stratis/snapm_st_*; do + [ -d "$pool_dir" ] || continue + pool_name=$(basename "$pool_dir") + if _validate_snapm_name "$pool_name"; then + STRATIS_FILESYSTEMS=$(stratis filesystem list "$pool_name" 2>/dev/null \ + | awk -v pool="$pool_name" '$0 ~ pool {print $2}') + for FS in $STRATIS_FILESYSTEMS; do + if ! FAIL=$(stratis filesystem destroy "$pool_name" "$FS" 2>&1); then + printf 'Failed to clean up %s %s: %s\n' "$pool_name" "$FS" "$FAIL" + fi + done + if stratis pool list 2>/dev/null | grep -qE "^${pool_name}([[:space:]]|$)"; then + if ! FAIL=$(stratis pool destroy "$pool_name" 2>&1); then + printf 'Failed to clean up %s: %s\n' "$pool_name" "$FAIL" + fi + fi + fi +done LOOP_DEVICES=$(losetup --noheadings --list -Oname,back-file | awk '/_snapm_loop_back/{print $1}') for loop in $LOOP_DEVICES; do diff --git a/tests/bin/snapm-test-name b/tests/bin/snapm-test-name new file mode 100755 index 00000000..e4d21672 --- /dev/null +++ b/tests/bin/snapm-test-name @@ -0,0 +1,73 @@ +#!/bin/bash +# snapm-test-name - generate and validate snapm test resource names +# +# Format: snapm___<8 hex random>_<4 hex checksum> +# +# Usage: +# snapm-test-name generate +# snapm-test-name validate +# +# Subsystem: lm (LVM), st (Stratis) +# Context: vg, pool, lv, vol + +set -euo pipefail + +_generate() { + local subsystem="$1" + local context="$2" + local random_hex + random_hex=$(od -An -tx1 -N4 /dev/urandom | tr -d ' \n') + local prefix="snapm_${subsystem}_${context}_${random_hex}" + local checksum + checksum=$(printf '%s' "$prefix" | sha256sum | cut -c1-4) + printf '%s_%s\n' "$prefix" "$checksum" +} + +_validate() { + local name="$1" + if [ ${#name} -lt 6 ]; then + echo "INVALID: name too short" + return 1 + fi + local sep="${name: -5:1}" + if [ "$sep" != "_" ]; then + echo "INVALID: missing checksum separator" + return 1 + fi + local prefix="${name:0:${#name}-5}" + local embedded="${name: -4}" + local computed + computed=$(printf '%s' "$prefix" | sha256sum | cut -c1-4) + if [ "$embedded" = "$computed" ]; then + echo "VALID: $name" + return 0 + else + echo "INVALID: checksum mismatch (expected $computed, got $embedded)" + return 1 + fi +} + +if [ $# -lt 2 ]; then + echo "Usage: $0 generate " >&2 + echo " $0 validate " >&2 + exit 2 +fi + +case "$1" in + generate) + if [ $# -ne 3 ]; then + echo "Usage: $0 generate " >&2 + exit 2 + fi + _generate "$2" "$3" + ;; + validate) + _validate "$2" + ;; + *) + echo "Unknown command: $1" >&2 + echo "Usage: $0 generate " >&2 + echo " $0 validate " >&2 + exit 2 + ;; +esac diff --git a/tests/test_boot.py b/tests/test_boot.py index cd8cf249..7c8ed65f 100644 --- a/tests/test_boot.py +++ b/tests/test_boot.py @@ -22,7 +22,7 @@ import boom from tests import have_root, is_redhat, BOOT_ROOT_TEST -from ._util import LvmLoopBacked +from ._util import LvmLoopBacked, _VG_NAME ETC_FSTAB = "/etc/fstab" TMP_FSTAB = "/tmp/fstab" @@ -100,7 +100,7 @@ def _set_fstab(self): with open(TMP_FSTAB, "w", encoding="utf8") as file: file.write("# Test fstab\n") for origin, mp in self.boot_volumes: - file.write(f"/dev/test_vg0/{origin}\t{mp}\text4\tdefaults 0 0\n") + file.write(f"/dev/{_VG_NAME}/{origin}\t{mp}\text4\tdefaults 0 0\n") run(["mount", "--bind", TMP_FSTAB, ETC_FSTAB], check=True) def _clear_fstab(self): diff --git a/tests/test_lvm2.py b/tests/test_lvm2.py index 9d763473..9efd60b5 100644 --- a/tests/test_lvm2.py +++ b/tests/test_lvm2.py @@ -18,7 +18,7 @@ from snapm import SnapmCalloutError from tests import have_root -from ._util import LvmLoopBacked +from ._util import LvmLoopBacked, _VG_NAME, _THIN_POOL_NAME class Lvm2TestsSimple(unittest.TestCase): @@ -232,17 +232,17 @@ def test_lvm2cow_create_delete_origin_accounting(self): # Create snapshot via Plugin.create_snapshot() lvm2cow_plugin.start_transaction() - lvm2cow_plugin.check_create_snapshot("test_vg0/root", "test", 1721136677, "/", "1%SIZE") - lvm2cow_plugin.create_snapshot("test_vg0/root", "test", 1721136677, "/", "1%SIZE") + lvm2cow_plugin.check_create_snapshot(f"{_VG_NAME}/root", "test", 1721136677, "/", "1%SIZE") + lvm2cow_plugin.create_snapshot(f"{_VG_NAME}/root", "test", 1721136677, "/", "1%SIZE") lvm2cow_plugin.end_transaction() # Verify counter set to one - origin_count = lvm2cow_plugin.origins["/dev/test_vg0/root"] + origin_count = lvm2cow_plugin.origins[f"/dev/{_VG_NAME}/root"] self.assertEqual(origin_count, 1) # Delete snapshot & verify decrement - lvm2cow_plugin.delete_snapshot("test_vg0/root-snapset_test_1721136677_-") - origin_count = lvm2cow_plugin.origins["/dev/test_vg0/root"] + lvm2cow_plugin.delete_snapshot(f"{_VG_NAME}/root-snapset_test_1721136677_-") + origin_count = lvm2cow_plugin.origins[f"/dev/{_VG_NAME}/root"] self.assertEqual(origin_count, 0) def test_lvm2thin_create_delete_pool_accounting(self): @@ -255,15 +255,15 @@ def test_lvm2thin_create_delete_pool_accounting(self): # Create snapshot via Plugin.create_snapshot() lvm2thin_plugin.start_transaction() - lvm2thin_plugin.check_create_snapshot("test_vg0/opt", "test", 1721136677, "/opt", "1%SIZE") - lvm2thin_plugin.create_snapshot("test_vg0/opt", "test", 1721136677, "/opt", "1%SIZE") + lvm2thin_plugin.check_create_snapshot(f"{_VG_NAME}/opt", "test", 1721136677, "/opt", "1%SIZE") + lvm2thin_plugin.create_snapshot(f"{_VG_NAME}/opt", "test", 1721136677, "/opt", "1%SIZE") lvm2thin_plugin.end_transaction() # Verify counter incremented - pool0_count = lvm2thin_plugin.pools["test_vg0/pool0"] + pool0_count = lvm2thin_plugin.pools[f"{_VG_NAME}/{_THIN_POOL_NAME}"] self.assertEqual(pool0_count, 1) # Delete snapshot & verify decrement - lvm2thin_plugin.delete_snapshot("test_vg0/opt-snapset_test_1721136677_-opt") - pool0_count = lvm2thin_plugin.pools["test_vg0/pool0"] + lvm2thin_plugin.delete_snapshot(f"{_VG_NAME}/opt-snapset_test_1721136677_-opt") + pool0_count = lvm2thin_plugin.pools[f"{_VG_NAME}/{_THIN_POOL_NAME}"] self.assertEqual(pool0_count, 0) diff --git a/tests/test_manager.py b/tests/test_manager.py index 0bc5eec2..19105e88 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -22,7 +22,7 @@ import boom from tests import have_root, BOOT_ROOT_TEST, MockPlugin -from ._util import LvmLoopBacked, StratisLoopBacked +from ._util import LvmLoopBacked, StratisLoopBacked, _VG_NAME log = logging.getLogger() @@ -636,7 +636,7 @@ def test_activate_deactivate_snapsets(self): self.assertEqual(snap.status, snapm.SnapStatus.ACTIVE) self.manager.deactivate_snapshot_sets(selection=s) for snap in sets[0].snapshots: - if snap.origin.removeprefix("test_vg0/") in self.thin_volumes: + if snap.origin.removeprefix(f"{_VG_NAME}/") in self.thin_volumes: self.assertEqual(snap.status, snapm.SnapStatus.INACTIVE) self.manager.delete_snapshot_sets(snapm.Selection(name="testset0")) diff --git a/tests/test_mounts.py b/tests/test_mounts.py index 336f9842..9d16ba89 100644 --- a/tests/test_mounts.py +++ b/tests/test_mounts.py @@ -25,7 +25,7 @@ from snapm.manager.plugins import format_snapshot_name, encode_mount_point from tests import have_root, is_redhat, in_rh_ci -from ._util import LoopBackDevices, LvmLoopBacked +from ._util import LoopBackDevices, LvmLoopBacked, _VG_NAME ETC_FSTAB = "/etc/fstab" @@ -171,7 +171,7 @@ def _set_fstab(self): with open(self._tmp_fstab, "w", encoding="utf8") as file: file.write("# Test fstab\n") for origin, mp in self.mount_volumes: - file.write(f"/dev/test_vg0/{origin}\t{mp}\text4\tdefaults 0 0\n") + file.write(f"/dev/{_VG_NAME}/{origin}\t{mp}\text4\tdefaults 0 0\n") run(["mount", "--bind", self._tmp_fstab, ETC_FSTAB], check=True) def _clear_fstab(self): @@ -232,7 +232,7 @@ def setUp(self): # ./~ Make it cheap & keep it shallow. Fill it up and make it hollow. ./~ with tempfile.TemporaryDirectory(prefix="snapm_root_prep_") as tempdir: - with mounted("/dev/test_vg0/root", tempdir): + with mounted(f"/dev/{_VG_NAME}/root", tempdir): for dirpath in _root_dirs: os.makedirs(os.path.join(tempdir, dirpath)) diff --git a/tests/test_stratis.py b/tests/test_stratis.py index 16a6a636..7e758f40 100644 --- a/tests/test_stratis.py +++ b/tests/test_stratis.py @@ -28,7 +28,7 @@ from snapm.manager.plugins import device_from_mount_point from tests import have_root -from ._util import StratisLoopBacked +from ._util import StratisLoopBacked, _POOL_NAME def _systemctl_args(command): @@ -137,8 +137,8 @@ def test__get_pool_filesystem(self): proxy = get_object(TOP_OBJECT) managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) - (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, "pool1", "fs1") - self.assertTrue(str(pool.Name()) == "pool1") + (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, _POOL_NAME, "fs1") + self.assertTrue(str(pool.Name()) == _POOL_NAME) self.assertTrue(str(filesystem.Name()) == "fs1") def test__get_pool_filesystem_bad_pool(self): @@ -146,21 +146,21 @@ def test__get_pool_filesystem_bad_pool(self): managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) with self.assertRaises(DbusClientUniqueResultError) as cm: - (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, "nosuchpool1", "fs1") + (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, "nosuchpool", "fs1") def test__get_pool_filesystem_bad_fs(self): proxy = get_object(TOP_OBJECT) managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) with self.assertRaises(DbusClientUniqueResultError) as cm: - (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, "pool1", "nosuchfs1") + (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, _POOL_NAME, "nosuchfs1") def test_pool_free_space_bytes(self): proxy = get_object(TOP_OBJECT) managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) - (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, "pool1", "fs1") - free_bytes = stratis._pool_free_space_bytes(managed_objects, "pool1") + (pool, filesystem) = stratis._get_pool_filesystem(managed_objects, _POOL_NAME, "fs1") + free_bytes = stratis._pool_free_space_bytes(managed_objects, _POOL_NAME) self.assertEqual( int(pool.TotalPhysicalSize()) - int(pool.TotalPhysicalUsed()[1]) if pool.TotalPhysicalUsed()[0] else 0, free_bytes @@ -170,13 +170,13 @@ def test_fs_size_bytes(self): proxy = get_object(TOP_OBJECT) managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) - size_bytes = stratis._fs_size_bytes(managed_objects, "pool1", "fs1") + size_bytes = stratis._fs_size_bytes(managed_objects, _POOL_NAME, "fs1") self.assertEqual(size_bytes, 2**30) def test_pool_fs_from_device_path(self): devpath = device_from_mount_point(self._stratis.mount_points()[0]) (pool, fs) = stratis.pool_fs_from_device_path(devpath) - self.assertEqual(pool, "pool1") + self.assertEqual(pool, _POOL_NAME) self.assertEqual(fs, "fs1") def test_stratis_discover_snapshots(self): @@ -193,20 +193,20 @@ def test_stratis_create_delete_pool_accounting(self): # Plugin setup stratis_plugin = stratis.Stratis(log, ConfigParser()) snapshots = stratis_plugin.discover_snapshots() - pool1_count = stratis_plugin.pools["pool1"] + pool_count = stratis_plugin.pools[_POOL_NAME] # Create snapshot via Plugin.create_snapshot() stratis_plugin.start_transaction() - stratis_plugin.check_create_snapshot("pool1/fs1", "test", 1721136677, "/opt", "1%SIZE") - stratis_plugin.create_snapshot("pool1/fs1", "test", 1721136677, "/opt", "1%SIZE") + stratis_plugin.check_create_snapshot(f"{_POOL_NAME}/fs1", "test", 1721136677, "/opt", "1%SIZE") + stratis_plugin.create_snapshot(f"{_POOL_NAME}/fs1", "test", 1721136677, "/opt", "1%SIZE") stratis_plugin.end_transaction() # Verify counter incremented - self.assertEqual(stratis_plugin.pools["pool1"], pool1_count + 1) + self.assertEqual(stratis_plugin.pools[_POOL_NAME], pool_count + 1) # Delete snapshot & verify decrement - stratis_plugin.delete_snapshot("pool1/fs1-snapset_test_1721136677_-opt") - self.assertEqual(stratis_plugin.pools["pool1"], pool1_count) + stratis_plugin.delete_snapshot(f"{_POOL_NAME}/fs1-snapset_test_1721136677_-opt") + self.assertEqual(stratis_plugin.pools[_POOL_NAME], pool_count) def test_stratis_can_snapshot_no_stratisd(self): systemctl_stop_args = _systemctl_args("stop") @@ -224,7 +224,7 @@ def test__origin_uuid_to_fs_name(self): proxy = get_object(TOP_OBJECT) managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) - (pool, fs) = stratis._get_pool_filesystem(managed_objects, "pool1", "fs1-snapset_test_1721136677_-opt") + (pool, fs) = stratis._get_pool_filesystem(managed_objects, _POOL_NAME, "fs1-snapset_test_1721136677_-opt") managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {}) @@ -242,7 +242,7 @@ def test_filter_stratis_snapshot_snapshot(self): stratis_plugin = stratis.Stratis(log, ConfigParser()) - (pool, fs) = stratis._get_pool_filesystem(managed_objects, "pool1", "fs1-snapset_test_1721136677_-opt") + (pool, fs) = stratis._get_pool_filesystem(managed_objects, _POOL_NAME, "fs1-snapset_test_1721136677_-opt") self.assertEqual(True, stratis_plugin._filter_stratis_snapshot(fs)) def test_filter_stratis_snapshot_nonsnapshot(self): @@ -251,5 +251,5 @@ def test_filter_stratis_snapshot_nonsnapshot(self): stratis_plugin = stratis.Stratis(log, ConfigParser()) - (pool, fs) = stratis._get_pool_filesystem(managed_objects, "pool1", "fs1") + (pool, fs) = stratis._get_pool_filesystem(managed_objects, _POOL_NAME, "fs1") self.assertEqual(False, stratis_plugin._filter_stratis_snapshot(fs))