Simone
4541dccf9c
_template_of_backup only stripped one "_"-chunk off the archive name, so real Template_<name>_%Y%m%d_%H%M%S.7z files (date *and* time) never matched their template -- backups silently failed to show up grouped under it. Replaced with a regex that strips the full optional date+time suffix and tolerates underscores in template names. Also adds delete_artifact/delete_backup (mirrored, containment-checked single-file deletion) and backups_by_template (groups list_backups() output by parsed template name, newest-first) for the web dashboard's per-item delete actions and the Templates panel's per-row backup list.
525 lines
16 KiB
Python
525 lines
16 KiB
Python
"""Tests for the Stage B (B3/B4) ``operations.inventory`` module.
|
|
|
|
Template discovery (``list_templates``) and containment-checked directory
|
|
listings (``list_dir``/``list_backups``/``list_artifacts``). No real ``7z``/
|
|
``vmrun`` involved -- everything is plain files under ``tmp_path``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import ci_orchestrator.operations.inventory as inventory_module
|
|
from ci_orchestrator.config import BackendConfig, Config, Paths
|
|
from ci_orchestrator.operations.inventory import (
|
|
FileEntry,
|
|
TemplateInfo,
|
|
list_artifacts,
|
|
list_backups,
|
|
list_dir,
|
|
list_templates,
|
|
)
|
|
|
|
|
|
def _make_config(root: Path) -> Config:
|
|
return Config(
|
|
paths=Paths(
|
|
root=root,
|
|
templates=root / "templates",
|
|
build_vms=root / "build-vms",
|
|
artifacts=root / "artifacts",
|
|
keys=root / "keys",
|
|
),
|
|
backend=BackendConfig(),
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_dir_bytes_cache() -> None:
|
|
"""The B7 ``dir_bytes`` cache is module-level -- start each test clean."""
|
|
inventory_module._dir_bytes_cache.clear()
|
|
|
|
|
|
# ── list_templates ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def _write_vmx(path: Path, guest_os: str) -> None:
|
|
path.write_text(f'.encoding = "UTF-8"\nguestOS = "{guest_os}"\n', encoding="utf-8")
|
|
|
|
|
|
def test_list_templates_missing_dir_is_empty(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path) # templates dir never created
|
|
|
|
assert list_templates(config) == []
|
|
|
|
|
|
def test_list_templates_parses_guest_os_and_transport_ssh(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "LinuxBuild2404"
|
|
tmpl_dir.mkdir(parents=True)
|
|
_write_vmx(tmpl_dir / "LinuxBuild2404.vmx", "ubuntu-64")
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.name == "LinuxBuild2404"
|
|
assert info.vmx == tmpl_dir / "LinuxBuild2404.vmx"
|
|
assert info.guest_os == "ubuntu-64"
|
|
assert info.transport == "ssh"
|
|
|
|
|
|
def test_list_templates_windows_guest_os_is_winrm_transport(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "WinBuild2025"
|
|
tmpl_dir.mkdir(parents=True)
|
|
_write_vmx(tmpl_dir / "WinBuild2025.vmx", "windows9srv-64")
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.transport == "winrm"
|
|
|
|
|
|
def test_list_templates_no_vmx_is_unknown_transport(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "Empty"
|
|
tmpl_dir.mkdir(parents=True)
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.vmx is None
|
|
assert info.guest_os is None
|
|
assert info.transport == "unknown"
|
|
|
|
|
|
def test_list_templates_has_vmem_true_when_present(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "WithVmem"
|
|
tmpl_dir.mkdir(parents=True)
|
|
_write_vmx(tmpl_dir / "WithVmem.vmx", "ubuntu-64")
|
|
(tmpl_dir / "WithVmem.vmem").write_bytes(b"\x00" * 16)
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.has_vmem is True
|
|
|
|
|
|
def test_list_templates_has_vmem_false_when_absent(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "Clean"
|
|
tmpl_dir.mkdir(parents=True)
|
|
_write_vmx(tmpl_dir / "Clean.vmx", "ubuntu-64")
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.has_vmem is False
|
|
|
|
|
|
def test_list_templates_dir_bytes_sums_file_sizes(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "Sized"
|
|
nested = tmpl_dir / "nested"
|
|
nested.mkdir(parents=True)
|
|
(tmpl_dir / "a.vmdk").write_bytes(b"x" * 100)
|
|
(nested / "b.vmdk").write_bytes(b"y" * 50)
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.dir_bytes == 150
|
|
|
|
|
|
def test_list_templates_multiple_entries_sorted_by_name(tmp_path: Path) -> None:
|
|
for name in ("WinBuild2022", "LinuxBuild2404", "WinBuild2025"):
|
|
(tmp_path / "templates" / name).mkdir(parents=True)
|
|
config = _make_config(tmp_path)
|
|
|
|
names = [t.name for t in list_templates(config)]
|
|
|
|
assert names == sorted(names)
|
|
|
|
|
|
def test_list_templates_last_backup_picks_newest_matching_archive(tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "LinuxBuild2404"
|
|
tmpl_dir.mkdir(parents=True)
|
|
backups_dir = tmp_path / "backups"
|
|
backups_dir.mkdir()
|
|
|
|
older = backups_dir / "Template_LinuxBuild2404_20260101_000000.7z"
|
|
newer = backups_dir / "Template_LinuxBuild2404_20260601_000000.7z"
|
|
unrelated = backups_dir / "Template_OtherTemplate_20260701_000000.7z"
|
|
older.write_bytes(b"old")
|
|
unrelated.write_bytes(b"other")
|
|
time.sleep(0.01)
|
|
newer.write_bytes(b"new")
|
|
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.last_backup is not None
|
|
assert info.last_backup.name == "Template_LinuxBuild2404_20260601_000000.7z"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("filename", "expected"),
|
|
[
|
|
# Real archive shape: ``Template_<name>_%Y%m%d_%H%M%S.7z`` -- both the
|
|
# date and the time chunk must be stripped (regression: only the time
|
|
# chunk was, leaving "LinuxBuild2404_20260722").
|
|
("Template_LinuxBuild2404_20260722_145001.7z", "LinuxBuild2404"),
|
|
("Template_WinBuild2025_20260101_000000.7z", "WinBuild2025"),
|
|
# Template names containing underscores survive intact.
|
|
("Template_Win_Build_2025_20260722_145001.7z", "Win_Build_2025"),
|
|
# Date-only shape (no time chunk) is still tolerated.
|
|
("Template_LinuxBuild2404_20260722.7z", "LinuxBuild2404"),
|
|
# Non-matching names parse to None.
|
|
("not-a-template-archive.7z", None),
|
|
("Template_no_timestamp.txt", None),
|
|
],
|
|
)
|
|
def test_template_of_backup_strips_full_timestamp(filename: str, expected: str | None) -> None:
|
|
assert inventory_module._template_of_backup(filename) == expected
|
|
|
|
|
|
def test_list_templates_last_backup_none_when_no_backups_dir(tmp_path: Path) -> None:
|
|
(tmp_path / "templates" / "LinuxBuild2404").mkdir(parents=True)
|
|
config = _make_config(tmp_path)
|
|
|
|
[info] = list_templates(config)
|
|
|
|
assert info.last_backup is None
|
|
|
|
|
|
def test_list_templates_skips_regular_files_directly_under_templates(tmp_path: Path) -> None:
|
|
templates_dir = tmp_path / "templates"
|
|
templates_dir.mkdir(parents=True)
|
|
(templates_dir / "README.txt").write_text("not a template", encoding="utf-8")
|
|
(templates_dir / "Real").mkdir()
|
|
config = _make_config(tmp_path)
|
|
|
|
names = [t.name for t in list_templates(config)]
|
|
|
|
assert names == ["Real"]
|
|
|
|
|
|
def test_list_templates_dir_bytes_is_cached(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
tmpl_dir = tmp_path / "templates" / "Cached"
|
|
tmpl_dir.mkdir(parents=True)
|
|
(tmpl_dir / "a.vmdk").write_bytes(b"x" * 10)
|
|
config = _make_config(tmp_path)
|
|
|
|
walk_calls: list[Path] = []
|
|
real_walk = os.walk
|
|
|
|
def _counting_walk(path: object, *a: object, **kw: object) -> object:
|
|
walk_calls.append(Path(str(path)))
|
|
return real_walk(path, *a, **kw) # type: ignore[arg-type]
|
|
|
|
monkeypatch.setattr(inventory_module.os, "walk", _counting_walk)
|
|
|
|
first = list_templates(config)[0].dir_bytes
|
|
second = list_templates(config)[0].dir_bytes
|
|
|
|
assert first == second == 10
|
|
assert len(walk_calls) == 1, "second call must hit the TTL cache, not os.walk again"
|
|
|
|
|
|
# ── list_dir / containment ───────────────────────────────────────────────────
|
|
|
|
|
|
def test_list_dir_accepts_root_itself(tmp_path: Path) -> None:
|
|
(tmp_path / "f1.txt").write_bytes(b"a")
|
|
|
|
entries = list_dir(tmp_path, root=tmp_path)
|
|
|
|
assert [e.name for e in entries] == ["f1.txt"]
|
|
|
|
|
|
def test_list_dir_accepts_real_subdir(tmp_path: Path) -> None:
|
|
sub = tmp_path / "backups"
|
|
sub.mkdir()
|
|
(sub / "f1.7z").write_bytes(b"a")
|
|
|
|
entries = list_dir(sub, root=tmp_path, pattern="*.7z")
|
|
|
|
assert [e.name for e in entries] == ["f1.7z"]
|
|
|
|
|
|
def test_list_dir_rejects_base_outside_root(tmp_path: Path) -> None:
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret.txt").write_bytes(b"nope")
|
|
|
|
assert list_dir(outside, root=root) == []
|
|
|
|
|
|
def test_list_dir_rejects_dotdot_escape(tmp_path: Path) -> None:
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret.txt").write_bytes(b"nope")
|
|
|
|
escaping = root / ".." / "outside"
|
|
|
|
assert list_dir(escaping, root=root) == []
|
|
|
|
|
|
def test_list_dir_rejects_symlink_escaping_root(tmp_path: Path) -> None:
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret.txt").write_bytes(b"nope")
|
|
|
|
link = root / "escape_link"
|
|
link.symlink_to(outside, target_is_directory=True)
|
|
|
|
assert list_dir(link, root=root) == []
|
|
|
|
|
|
def test_list_dir_missing_base_is_empty(tmp_path: Path) -> None:
|
|
assert list_dir(tmp_path / "does-not-exist", root=tmp_path) == []
|
|
|
|
|
|
def test_list_dir_sorted_newest_first(tmp_path: Path) -> None:
|
|
older = tmp_path / "a.txt"
|
|
newer = tmp_path / "b.txt"
|
|
older.write_bytes(b"old")
|
|
time.sleep(0.01)
|
|
newer.write_bytes(b"new")
|
|
|
|
entries = list_dir(tmp_path, root=tmp_path)
|
|
|
|
assert [e.name for e in entries] == ["b.txt", "a.txt"]
|
|
|
|
|
|
def test_list_dir_pattern_filters_extension(tmp_path: Path) -> None:
|
|
(tmp_path / "keep.7z").write_bytes(b"x")
|
|
(tmp_path / "skip.txt").write_bytes(b"y")
|
|
|
|
entries = list_dir(tmp_path, root=tmp_path, pattern="*.7z")
|
|
|
|
assert [e.name for e in entries] == ["keep.7z"]
|
|
|
|
|
|
def test_list_dir_entry_shape(tmp_path: Path) -> None:
|
|
f = tmp_path / "x.7z"
|
|
f.write_bytes(b"12345")
|
|
|
|
[entry] = list_dir(tmp_path, root=tmp_path)
|
|
|
|
assert isinstance(entry, FileEntry)
|
|
assert entry.size_bytes == 5
|
|
assert entry.mtime == pytest.approx(f.stat().st_mtime)
|
|
|
|
|
|
# ── list_backups / list_artifacts ────────────────────────────────────────────
|
|
|
|
|
|
def test_list_backups_reads_root_backups_dir(tmp_path: Path) -> None:
|
|
backups = tmp_path / "backups"
|
|
backups.mkdir()
|
|
(backups / "Template_X_20260101_000000.7z").write_bytes(b"x")
|
|
(backups / "notes.txt").write_bytes(b"ignored")
|
|
config = _make_config(tmp_path)
|
|
|
|
entries = list_backups(config)
|
|
|
|
assert [e.name for e in entries] == ["Template_X_20260101_000000.7z"]
|
|
|
|
|
|
def test_list_backups_missing_dir_is_empty(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
|
|
assert list_backups(config) == []
|
|
|
|
|
|
def test_list_artifacts_reads_configured_artifacts_dir(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
config.paths.artifacts.mkdir(parents=True)
|
|
(config.paths.artifacts / "build.log").write_bytes(b"log")
|
|
|
|
entries = list_artifacts(config)
|
|
|
|
assert [e.name for e in entries] == ["build.log"]
|
|
|
|
|
|
def test_list_artifacts_outside_root_is_rejected(tmp_path: Path) -> None:
|
|
"""``config.paths.artifacts`` pointed outside ``root`` must still 0-list.
|
|
|
|
``outside_artifacts`` sits beside ``root`` (both under ``tmp_path``) --
|
|
a sibling directory, not a descendant -- so containment must reject it.
|
|
"""
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside_artifacts = tmp_path / "artifacts-outside"
|
|
outside_artifacts.mkdir()
|
|
(outside_artifacts / "build.log").write_bytes(b"log")
|
|
|
|
config = Config(
|
|
paths=Paths(
|
|
root=root,
|
|
templates=root / "templates",
|
|
build_vms=root / "build-vms",
|
|
artifacts=outside_artifacts,
|
|
keys=root / "keys",
|
|
),
|
|
backend=BackendConfig(),
|
|
)
|
|
|
|
assert list_artifacts(config) == []
|
|
|
|
|
|
# ── defensive OSError degrade paths ("never raise") ─────────────────────────
|
|
|
|
|
|
class _BoomIterdir:
|
|
"""Duck-typed stand-in for a ``Path`` whose ``iterdir()`` blows up."""
|
|
|
|
def iterdir(self) -> list[Path]:
|
|
raise OSError("denied")
|
|
|
|
|
|
class _BoomGlob:
|
|
"""Duck-typed stand-in for a ``Path`` whose ``glob()`` blows up."""
|
|
|
|
def glob(self, _pattern: str) -> list[Path]:
|
|
raise OSError("denied")
|
|
|
|
|
|
class _BoomReadText:
|
|
"""Duck-typed stand-in for a ``Path`` whose ``read_text()`` blows up."""
|
|
|
|
def read_text(self, encoding: str = "utf-8", errors: str = "strict") -> str:
|
|
raise OSError("denied")
|
|
|
|
|
|
class _BoomResolve:
|
|
"""Duck-typed stand-in for a ``Path`` whose ``resolve()`` blows up."""
|
|
|
|
def resolve(self) -> Path:
|
|
raise OSError("denied")
|
|
|
|
|
|
def test_find_vmx_oserror_returns_none() -> None:
|
|
assert inventory_module._find_vmx(_BoomIterdir()) is None # type: ignore[arg-type]
|
|
|
|
|
|
def test_has_vmem_oserror_returns_false() -> None:
|
|
assert inventory_module._has_vmem(_BoomGlob()) is False # type: ignore[arg-type]
|
|
|
|
|
|
def test_read_guest_os_oserror_returns_empty_string() -> None:
|
|
assert inventory_module._read_guest_os(_BoomReadText()) == "" # type: ignore[arg-type]
|
|
|
|
|
|
def test_dir_bytes_cached_resolve_oserror_returns_zero() -> None:
|
|
assert inventory_module._dir_bytes_cached(_BoomResolve()) == 0 # type: ignore[arg-type]
|
|
|
|
|
|
def test_newest_backup_glob_oserror_returns_none(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
backups_dir = tmp_path / "backups"
|
|
backups_dir.mkdir()
|
|
monkeypatch.setattr(Path, "glob", lambda self, pattern: (_ for _ in ()).throw(OSError("denied")))
|
|
|
|
assert inventory_module._newest_backup(backups_dir, "X") is None
|
|
|
|
|
|
def test_dir_bytes_cached_walk_oserror_degrades_to_zero(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
tmpl_dir = tmp_path / "T"
|
|
tmpl_dir.mkdir()
|
|
(tmpl_dir / "a.txt").write_bytes(b"x" * 5)
|
|
|
|
def _boom_walk(_path: Path, *_a: object, **_kw: object) -> object:
|
|
raise OSError("walk exploded")
|
|
|
|
monkeypatch.setattr(inventory_module.os, "walk", _boom_walk)
|
|
|
|
assert inventory_module._dir_bytes_cached(tmpl_dir) == 0
|
|
|
|
|
|
def test_dir_bytes_cache_recomputes_after_ttl_expiry(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""B7: a cache entry older than the TTL must be recomputed, not reused forever."""
|
|
tmpl_dir = tmp_path / "templates" / "Stale"
|
|
tmpl_dir.mkdir(parents=True)
|
|
(tmpl_dir / "a.vmdk").write_bytes(b"x" * 10)
|
|
config = _make_config(tmp_path)
|
|
|
|
clock = {"t": 0.0}
|
|
monkeypatch.setattr(inventory_module.time, "monotonic", lambda: clock["t"])
|
|
walk_calls: list[Path] = []
|
|
real_walk = os.walk
|
|
|
|
def _counting_walk(path: object, *a: object, **kw: object) -> object:
|
|
walk_calls.append(Path(str(path)))
|
|
return real_walk(path, *a, **kw) # type: ignore[arg-type]
|
|
|
|
monkeypatch.setattr(inventory_module.os, "walk", _counting_walk)
|
|
|
|
first = list_templates(config)[0].dir_bytes
|
|
clock["t"] += inventory_module._DIR_BYTES_TTL_SECONDS + 1.0
|
|
second = list_templates(config)[0].dir_bytes
|
|
|
|
assert first == second == 10
|
|
assert len(walk_calls) == 2, "an expired cache entry must trigger a fresh os.walk"
|
|
|
|
|
|
def test_list_templates_iterdir_oserror_returns_empty(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
templates_dir = tmp_path / "templates"
|
|
templates_dir.mkdir()
|
|
config = _make_config(tmp_path)
|
|
original_iterdir = Path.iterdir
|
|
|
|
def _boom_iterdir(self: Path) -> object:
|
|
if self == templates_dir:
|
|
raise OSError("denied")
|
|
return original_iterdir(self)
|
|
|
|
monkeypatch.setattr(Path, "iterdir", _boom_iterdir)
|
|
|
|
assert list_templates(config) == []
|
|
|
|
|
|
def test_list_dir_resolve_oserror_returns_empty(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
monkeypatch.setattr(Path, "resolve", lambda self: (_ for _ in ()).throw(OSError("denied")))
|
|
|
|
assert list_dir(tmp_path, root=tmp_path) == []
|
|
|
|
|
|
def test_list_dir_skips_subdirectories(tmp_path: Path) -> None:
|
|
(tmp_path / "subdir").mkdir()
|
|
(tmp_path / "file.txt").write_bytes(b"x")
|
|
|
|
entries = list_dir(tmp_path, root=tmp_path)
|
|
|
|
assert [e.name for e in entries] == ["file.txt"]
|
|
|
|
|
|
def test_template_info_is_frozen_dataclass() -> None:
|
|
info = TemplateInfo(
|
|
name="X",
|
|
vmx=None,
|
|
guest_os=None,
|
|
transport="unknown",
|
|
dir_bytes=0,
|
|
has_vmem=False,
|
|
last_backup=None,
|
|
)
|
|
with pytest.raises(AttributeError):
|
|
info.name = "Y" # type: ignore[misc]
|