Simone
3da992119b
uptime_sec came from the clone directory's own mtime, but a running VM's hypervisor can rewrite lock files under that same directory at any point, resetting the mtime and pinning "up" near 0s for the VM's entire life -- worst exactly while a build is actively running, since that's when the VM is busiest. The clone name already carries its own creation timestamp (_make_clone_name, parsed back via parse_clone_name), set once and never disturbed after -- prefer it, falling back to the old mtime read only for a name that doesn't parse.
592 lines
20 KiB
Python
592 lines
20 KiB
Python
"""Tests for the pure ``operations.vms`` layer (W1 of the web-UI plan).
|
|
|
|
Migrated from the direct-call tests that used to live in
|
|
``test_commands_vm.py`` (``_list_orphans`` / ``_find_vmx`` /
|
|
``cleanup_orphans``), adapted to the new dataclass return types. Adds
|
|
coverage for ``list_live_vms`` and ``remove_vm``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import ci_orchestrator.operations.vms as vms_module
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
from ci_orchestrator.commands.job import _make_clone_name
|
|
from ci_orchestrator.config import BackendConfig, Config, Paths
|
|
from ci_orchestrator.operations.vms import (
|
|
Orphan,
|
|
ParsedClone,
|
|
RemovalResult,
|
|
cleanup_orphans,
|
|
list_live_vms,
|
|
list_orphans,
|
|
parse_clone_name,
|
|
remove_vm,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(time, "sleep", lambda _s: None)
|
|
monkeypatch.setattr(vms_module.time, "sleep", lambda _s: None)
|
|
|
|
|
|
def _age_dir(path: Path, hours: float) -> None:
|
|
ts = (datetime.now() - timedelta(hours=hours)).timestamp()
|
|
import os
|
|
|
|
os.utime(path, (ts, ts))
|
|
|
|
|
|
def _make_clone_dir(tmp_path: Path, name: str = "clone") -> tuple[Path, Path]:
|
|
clone_dir = tmp_path / name
|
|
clone_dir.mkdir()
|
|
vmx = clone_dir / f"{name}.vmx"
|
|
vmx.write_text('config.version = "8"', encoding="utf-8")
|
|
return clone_dir, vmx
|
|
|
|
|
|
def _make_config(build_vms: Path) -> Config:
|
|
return Config(
|
|
paths=Paths(
|
|
root=build_vms.parent,
|
|
templates=build_vms.parent / "templates",
|
|
build_vms=build_vms,
|
|
artifacts=build_vms.parent / "artifacts",
|
|
keys=build_vms.parent / "keys",
|
|
),
|
|
backend=BackendConfig(),
|
|
)
|
|
|
|
|
|
class _FakeBackend:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
running_after_stop: bool = False,
|
|
delete_fails_first: int = 0,
|
|
) -> None:
|
|
self.calls: list[tuple[str, Any]] = []
|
|
self._running = True
|
|
self._running_after_stop = running_after_stop
|
|
self._delete_fails_remaining = delete_fails_first
|
|
|
|
def stop(self, handle: VmHandle, hard: bool = False) -> None:
|
|
self.calls.append(("stop", hard))
|
|
if not hard or not self._running_after_stop:
|
|
self._running = False
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
return self._running
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
self.calls.append(("delete", None))
|
|
if self._delete_fails_remaining > 0:
|
|
self._delete_fails_remaining -= 1
|
|
raise BackendOperationFailed("deleteVM", 1, "transient")
|
|
|
|
|
|
# ── list_orphans ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_list_orphans_non_dir_base(tmp_path: Path) -> None:
|
|
f = tmp_path / "file.txt"
|
|
f.write_text("x")
|
|
assert list_orphans(f, 4) == []
|
|
|
|
|
|
def test_list_orphans_skips_non_dir_and_oserror(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
(base / "a-file").write_text("not a dir")
|
|
bad_dir = base / "bad"
|
|
bad_dir.mkdir()
|
|
_age_dir(bad_dir, hours=10)
|
|
|
|
real_stat = Path.stat
|
|
|
|
def _stat(self: Path, *a: Any, **kw: Any) -> Any:
|
|
if self == bad_dir:
|
|
raise OSError("nope")
|
|
return real_stat(self, *a, **kw)
|
|
|
|
monkeypatch.setattr(Path, "stat", _stat)
|
|
assert list_orphans(base, 4) == []
|
|
|
|
|
|
def test_list_orphans_returns_orphan_dataclass(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
old, vmx = _make_clone_dir(base, "old")
|
|
_age_dir(old, hours=10)
|
|
|
|
orphans = list_orphans(base, 4)
|
|
assert orphans == [Orphan(path=old, vmx=vmx, age_hours=10)]
|
|
|
|
|
|
# ── _find_vmx ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_find_vmx_returns_none(tmp_path: Path) -> None:
|
|
d = tmp_path / "novmx"
|
|
d.mkdir()
|
|
(d / "readme.txt").write_text("x")
|
|
assert vms_module._find_vmx(d) is None
|
|
|
|
|
|
# ── cleanup_orphans ────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_cleanup_orphans_dir_without_vmx_logs_and_removes(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
no_vmx = base / "no-vmx"
|
|
no_vmx.mkdir()
|
|
_age_dir(no_vmx, hours=10)
|
|
results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=None)
|
|
assert sum(r.dir_removed for r in results) == 1
|
|
assert not no_vmx.exists()
|
|
|
|
|
|
def test_cleanup_orphans_backend_stop_error_swallowed(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
old, _ = _make_clone_dir(base, "old1")
|
|
_age_dir(old, hours=10)
|
|
|
|
class _B:
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
raise BackendOperationFailed("stop", 1, "no")
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
raise BackendOperationFailed("delete", 1, "no")
|
|
|
|
results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=_B())
|
|
assert sum(r.dir_removed for r in results) == 1
|
|
assert results[0].stopped is False
|
|
assert results[0].deleted is False
|
|
assert results[0].error is not None
|
|
|
|
|
|
def test_cleanup_orphans_dir_removal_failure_logs(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
old, _ = _make_clone_dir(base, "stuck")
|
|
_age_dir(old, hours=10)
|
|
monkeypatch.setattr(vms_module, "_try_remove_dir", lambda _p: False)
|
|
results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=None)
|
|
assert sum(r.dir_removed for r in results) == 0
|
|
assert len(results) == 1
|
|
|
|
|
|
def test_cleanup_orphans_lock_unlink_oserror(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
lock = tmp_path / "stale.lock"
|
|
lock.write_text("x")
|
|
_age_dir(lock, hours=1)
|
|
|
|
real_unlink = Path.unlink
|
|
|
|
def _boom(self: Path, *a: Any, **kw: Any) -> Any:
|
|
if self == lock:
|
|
raise OSError("locked")
|
|
return real_unlink(self, *a, **kw)
|
|
|
|
monkeypatch.setattr(Path, "unlink", _boom)
|
|
# Must not raise even though unlink() fails.
|
|
cleanup_orphans(clone_base=base, max_age_hours=4, backend=None, lock_file=lock)
|
|
|
|
|
|
def test_cleanup_orphans_dry_run_no_destroy(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
old, vmx = _make_clone_dir(base, "old")
|
|
_age_dir(old, hours=10)
|
|
backend = _FakeBackend()
|
|
results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=backend, dry_run=True)
|
|
assert results == [RemovalResult(target=str(vmx), stopped=False, deleted=False, dir_removed=False, error=None)]
|
|
assert backend.calls == []
|
|
assert old.exists()
|
|
|
|
|
|
def test_cleanup_orphans_no_orphans_returns_empty(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
assert cleanup_orphans(clone_base=base, max_age_hours=4, backend=None) == []
|
|
|
|
|
|
def test_cleanup_orphans_lock_respects_dry_run(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
lock = tmp_path / "stale.lock"
|
|
lock.write_text("x")
|
|
_age_dir(lock, hours=1)
|
|
cleanup_orphans(clone_base=base, max_age_hours=4, backend=None, dry_run=True, lock_file=lock)
|
|
assert lock.exists() # dry run never touches the lock file
|
|
|
|
|
|
# ── list_live_vms ────────────────────────────────────────────────────────
|
|
|
|
|
|
class _FakeLiveBackend:
|
|
def __init__(
|
|
self,
|
|
handles: list[VmHandle],
|
|
*,
|
|
ip: str | None = "192.168.1.50",
|
|
ip_raises: bool = False,
|
|
list_running_raises: bool = False,
|
|
) -> None:
|
|
self._handles = handles
|
|
self._ip = ip
|
|
self._ip_raises = ip_raises
|
|
self._list_running_raises = list_running_raises
|
|
|
|
def list_running(self) -> list[VmHandle]:
|
|
if self._list_running_raises:
|
|
raise BackendOperationFailed("list", 1, "boom")
|
|
return self._handles
|
|
|
|
def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None:
|
|
if self._ip_raises:
|
|
raise BackendOperationFailed("getGuestIPAddress", 1, "no tools")
|
|
return self._ip
|
|
|
|
|
|
def test_list_live_vms_orphan_flag_on_old_clone(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
clone_dir, vmx = _make_clone_dir(build_vms, "old")
|
|
vmx.write_text('guestOS = "ubuntu-64"', encoding="utf-8")
|
|
_age_dir(clone_dir, hours=10)
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name="old")])
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert len(result) == 1
|
|
vm = result[0]
|
|
assert vm.name == "old"
|
|
assert vm.vmx == str(vmx)
|
|
assert vm.ip == "192.168.1.50"
|
|
assert vm.guest_os == "ubuntu-64"
|
|
assert vm.uptime_sec is not None and vm.uptime_sec >= 9 * 3600
|
|
assert vm.is_orphan is True
|
|
|
|
|
|
def test_list_live_vms_fresh_clone_not_orphan(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
clone_dir, vmx = _make_clone_dir(build_vms, "fresh")
|
|
_age_dir(clone_dir, hours=0)
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name=None)])
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert len(result) == 1
|
|
vm = result[0]
|
|
assert vm.name == vmx.stem # name fell back to vmx stem
|
|
assert vm.is_orphan is False
|
|
|
|
|
|
def test_list_live_vms_outside_build_vms_not_orphan(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
(tmp_path / "elsewhere").mkdir()
|
|
elsewhere, vmx = _make_clone_dir(tmp_path / "elsewhere", "other")
|
|
_age_dir(elsewhere, hours=10)
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name="other")])
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert len(result) == 1
|
|
vm = result[0]
|
|
assert vm.is_orphan is False
|
|
assert vm.uptime_sec is None
|
|
|
|
|
|
def test_list_live_vms_uptime_uses_clone_name_timestamp_not_directory_mtime(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The clone directory's own mtime can be disturbed by the hypervisor
|
|
while the VM is running (VMware rewrites lock files under it) -- the
|
|
timestamp already baked into the clone's own name at creation time can't
|
|
be, so it must win over the filesystem mtime."""
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
two_hours_ago = datetime.now(UTC) - timedelta(hours=2)
|
|
name = f"Clone_412-1_{two_hours_ago:%Y%m%d_%H%M%S}_linux"
|
|
clone_dir, vmx = _make_clone_dir(build_vms, name)
|
|
# Something touched the directory moments ago -- its mtime says "just
|
|
# now" even though the clone itself is 2h old.
|
|
_age_dir(clone_dir, hours=0)
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name=name)])
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert len(result) == 1
|
|
vm = result[0]
|
|
assert vm.uptime_sec is not None and vm.uptime_sec >= 7000
|
|
|
|
|
|
def test_list_live_vms_get_ip_none(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
_, vmx = _make_clone_dir(build_vms, "noip")
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(vmx))], ip=None)
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert result[0].ip is None
|
|
|
|
|
|
def test_list_live_vms_get_ip_raises_is_tolerated(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
_, vmx = _make_clone_dir(build_vms, "iperr")
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(vmx))], ip_raises=True)
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert result[0].ip is None
|
|
|
|
|
|
def test_list_live_vms_list_running_raises_returns_empty(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
backend = _FakeLiveBackend([], list_running_raises=True)
|
|
config = _make_config(build_vms)
|
|
assert list_live_vms(backend, config) == [] # type: ignore[arg-type]
|
|
|
|
|
|
def test_list_live_vms_missing_vmx_file_guest_os_empty(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
missing_vmx = build_vms / "ghost" / "ghost.vmx" # parent dir doesn't even exist
|
|
|
|
backend = _FakeLiveBackend([VmHandle(identifier=str(missing_vmx))])
|
|
config = _make_config(build_vms)
|
|
|
|
result = list_live_vms(backend, config) # type: ignore[arg-type]
|
|
assert result[0].guest_os == ""
|
|
|
|
|
|
def test_list_live_vms_empty_when_nothing_running(tmp_path: Path) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
build_vms.mkdir()
|
|
backend = _FakeLiveBackend([])
|
|
config = _make_config(build_vms)
|
|
assert list_live_vms(backend, config) == [] # type: ignore[arg-type]
|
|
|
|
|
|
# ── remove_vm ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_remove_vm_happy_path(tmp_path: Path) -> None:
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "live")
|
|
backend = _FakeBackend()
|
|
result = remove_vm(backend, vmx) # type: ignore[arg-type]
|
|
assert result == RemovalResult(
|
|
target=str(vmx), stopped=True, deleted=True, dir_removed=True, error=None
|
|
)
|
|
assert not clone_dir.exists()
|
|
assert ("delete", None) in backend.calls
|
|
|
|
|
|
def test_remove_vm_dry_run_no_op(tmp_path: Path) -> None:
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "dry")
|
|
backend = _FakeBackend()
|
|
result = remove_vm(backend, vmx, dry_run=True) # type: ignore[arg-type]
|
|
assert result == RemovalResult(
|
|
target=str(vmx), stopped=False, deleted=False, dir_removed=False, error=None
|
|
)
|
|
assert backend.calls == []
|
|
assert clone_dir.exists()
|
|
|
|
|
|
def test_remove_vm_delete_retry_succeeds(tmp_path: Path) -> None:
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "retry")
|
|
backend = _FakeBackend(delete_fails_first=2)
|
|
result = remove_vm(backend, vmx) # type: ignore[arg-type]
|
|
delete_calls = [c for c in backend.calls if c[0] == "delete"]
|
|
assert len(delete_calls) == 3 # 2 failures + 1 success
|
|
assert result.deleted is True
|
|
assert not clone_dir.exists()
|
|
|
|
|
|
def test_remove_vm_delete_fails_all_retries(tmp_path: Path) -> None:
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "neverdeletes")
|
|
|
|
class _B:
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
pass
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
raise BackendOperationFailed("delete", 1, "always")
|
|
|
|
result = remove_vm(_B(), vmx) # type: ignore[arg-type]
|
|
assert result.deleted is False
|
|
assert result.error is not None
|
|
# Directory removal is still attempted even when delete fails.
|
|
assert not clone_dir.exists()
|
|
assert result.dir_removed is True
|
|
|
|
|
|
def test_remove_vm_stop_error_is_swallowed(tmp_path: Path) -> None:
|
|
_, vmx = _make_clone_dir(tmp_path, "stoperr")
|
|
|
|
class _B:
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
raise BackendOperationFailed("stop", 1, "no")
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
pass
|
|
|
|
result = remove_vm(_B(), vmx) # type: ignore[arg-type]
|
|
assert result.stopped is False
|
|
assert result.deleted is True
|
|
assert result.error is not None
|
|
|
|
|
|
def test_remove_vm_dir_removal_failure(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
_, vmx = _make_clone_dir(tmp_path, "stubborn")
|
|
backend = _FakeBackend()
|
|
monkeypatch.setattr(vms_module, "_try_remove_dir", lambda _p: False)
|
|
result = remove_vm(backend, vmx) # type: ignore[arg-type]
|
|
assert result.dir_removed is False
|
|
|
|
|
|
# ── reconcile_ip_leases (TODO §2.8) ───────────────────────────────────────────
|
|
|
|
|
|
def _make_pool(tmp_path: Path) -> Any:
|
|
from ci_orchestrator.ip_pool import IpSlotPool
|
|
|
|
return IpSlotPool(
|
|
addresses=["10.0.0.1", "10.0.0.2"],
|
|
netmask="255.255.255.0",
|
|
gateway="10.0.0.254",
|
|
lease_file=tmp_path / "ip-pool.json",
|
|
)
|
|
|
|
|
|
def test_reconcile_ip_leases_releases_dead_clone(tmp_path: Path) -> None:
|
|
from ci_orchestrator.operations.vms import reconcile_ip_leases
|
|
|
|
base = tmp_path / "build-vms"
|
|
base.mkdir()
|
|
(base / "clone-alive").mkdir() # survives on disk
|
|
pool = _make_pool(tmp_path)
|
|
pool.acquire("clone-alive") # 10.0.0.1
|
|
pool.acquire("clone-dead") # 10.0.0.2 — no dir on disk
|
|
|
|
released = reconcile_ip_leases(base, pool)
|
|
|
|
assert released == ["10.0.0.2"]
|
|
|
|
|
|
def test_reconcile_ip_leases_dry_run_no_write(tmp_path: Path) -> None:
|
|
from ci_orchestrator.operations.vms import reconcile_ip_leases
|
|
|
|
base = tmp_path / "build-vms"
|
|
base.mkdir()
|
|
pool = _make_pool(tmp_path)
|
|
pool.acquire("clone-dead") # 10.0.0.1, dir absent
|
|
|
|
released = reconcile_ip_leases(base, pool, dry_run=True)
|
|
|
|
assert released == ["10.0.0.1"]
|
|
import json
|
|
|
|
leases = json.loads((tmp_path / "ip-pool.json").read_text())
|
|
assert leases["10.0.0.1"] == "clone-dead" # not written
|
|
|
|
|
|
def test_reconcile_ip_leases_unreadable_base_is_noop(tmp_path: Path) -> None:
|
|
from ci_orchestrator.operations.vms import reconcile_ip_leases
|
|
|
|
missing = tmp_path / "does-not-exist"
|
|
pool = _make_pool(tmp_path)
|
|
pool.acquire("clone-dead")
|
|
|
|
# clone_base cannot be enumerated → must NOT release anything.
|
|
assert reconcile_ip_leases(missing, pool) == []
|
|
import json
|
|
|
|
leases = json.loads((tmp_path / "ip-pool.json").read_text())
|
|
assert leases["10.0.0.1"] == "clone-dead"
|
|
|
|
|
|
# ── parse_clone_name (Stage C1) ───────────────────────────────────────────
|
|
|
|
|
|
def test_parse_clone_name_round_trips_real_make_clone_name() -> None:
|
|
"""A real ``_make_clone_name`` output must invert cleanly."""
|
|
before = datetime.now(tz=UTC).replace(microsecond=0)
|
|
name = _make_clone_name("build-42")
|
|
after = datetime.now(tz=UTC).replace(microsecond=0)
|
|
|
|
parsed = parse_clone_name(name)
|
|
|
|
assert parsed.job_id == "build-42"
|
|
assert parsed.started is not None
|
|
# started is a naive datetime built from UTC wall-clock digits -- compare
|
|
# against naive bounds taken from an aware UTC "now" either side of the call.
|
|
assert before.replace(tzinfo=None) <= parsed.started <= after.replace(tzinfo=None) + timedelta(seconds=1)
|
|
|
|
|
|
def test_parse_clone_name_job_id_containing_underscore() -> None:
|
|
parsed = parse_clone_name("Clone_my_job_42_20250101_123456_abc123")
|
|
|
|
assert parsed.job_id == "my_job_42"
|
|
assert parsed.started == datetime(2025, 1, 1, 12, 34, 56)
|
|
|
|
|
|
def test_parse_clone_name_non_clone_prefix_degrades_both_to_none() -> None:
|
|
assert parse_clone_name("SomeOtherVMName") == ParsedClone(job_id=None, started=None)
|
|
|
|
|
|
def test_parse_clone_name_too_few_parts_degrades_both_to_none() -> None:
|
|
# No room for date + time + suffix at all.
|
|
assert parse_clone_name("Clone_short") == ParsedClone(job_id=None, started=None)
|
|
|
|
|
|
def test_parse_clone_name_garbage_timestamp_degrades_started_only() -> None:
|
|
parsed = parse_clone_name("Clone_myjob_notadate_notatime_abc123")
|
|
|
|
assert parsed.job_id == "myjob"
|
|
assert parsed.started is None
|
|
|
|
|
|
def test_parse_clone_name_empty_job_id_degrades_to_none() -> None:
|
|
# No text between "Clone_" and the timestamp.
|
|
parsed = parse_clone_name("Clone_20250101_123456_abc123")
|
|
|
|
assert parsed.job_id is None
|
|
assert parsed.started == datetime(2025, 1, 1, 12, 34, 56)
|
|
|
|
|
|
def test_parse_clone_name_wrong_length_date_degrades_started_only() -> None:
|
|
parsed = parse_clone_name("Clone_job1_2025010_123456_abc123")
|
|
|
|
assert parsed.started is None
|