generated from coulomb/repo-seed
tegwick
c0a9261cdc
Add profile.sandbox-canary, HostSnapshot/inventory/stale schemas, SSH collectors, before/after provision deltas, telemetry export to State Hub and local JSON, default `sandboxer create` self-deploy, inspect/reap-stale CLI, runbook, and CoulombCore verification (26 tests pass).
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""Telemetry parsing and introspection tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from sandboxer.defaults import resolve_create_defaults
|
|
from sandboxer.profiles.loader import load_profile
|
|
from sandboxer.telemetry.host_snapshot import (
|
|
parse_container_count,
|
|
parse_df_root,
|
|
parse_loadavg,
|
|
)
|
|
from sandboxer.telemetry.introspection import profile_wants_telemetry
|
|
from sandboxer.telemetry.models import HostSnapshot, SandboxInventory
|
|
|
|
|
|
def test_parse_loadavg() -> None:
|
|
assert parse_loadavg("0.52 0.48 0.45 1/234 999") == (0.52, 0.48, 0.45)
|
|
|
|
|
|
def test_parse_df_root() -> None:
|
|
line = "Filesystem Size Used Avail Use% Mounted\n/dev/sda1 100G 40G 55G 43% /"
|
|
used, avail = parse_df_root(line)
|
|
assert used == 43.0
|
|
assert avail == 55.0
|
|
|
|
|
|
def test_parse_container_count() -> None:
|
|
assert parse_container_count("abc\ndef\n") == 2
|
|
|
|
|
|
def test_profile_wants_telemetry_canary() -> None:
|
|
profile = load_profile("profile.sandbox-canary")
|
|
assert profile_wants_telemetry(profile) is True
|
|
|
|
|
|
def test_profile_wants_telemetry_compose_e2e() -> None:
|
|
profile = load_profile("profile.compose-e2e")
|
|
assert profile_wants_telemetry(profile) is False
|
|
|
|
|
|
def test_resolve_create_defaults_no_args() -> None:
|
|
profile, inputs = resolve_create_defaults(None, {})
|
|
assert profile == "profile.sandbox-canary"
|
|
assert "repo" in inputs
|
|
assert Path(inputs["repo"]).name == "sand-boxer"
|
|
|
|
|
|
def test_resolve_create_defaults_explicit_repo() -> None:
|
|
profile, inputs = resolve_create_defaults(None, {"repo": "/tmp/foo"})
|
|
assert profile == "profile.compose-e2e"
|
|
assert inputs["repo"] == "/tmp/foo"
|
|
|
|
|
|
def test_build_introspection_report_mocked(tmp_path: Path) -> None:
|
|
from sandboxer.lifecycle.store import SandboxStore
|
|
from sandboxer.telemetry.introspection import build_introspection_report
|
|
|
|
now = datetime.now(UTC)
|
|
snap = HostSnapshot(collected_at=now, host="h1", load_1m=1.0, mem_available_mb=1000)
|
|
snap2 = HostSnapshot(collected_at=now, host="h1", load_1m=1.5, mem_available_mb=900)
|
|
profile = load_profile("profile.sandbox-canary")
|
|
store = SandboxStore(path=tmp_path / "sandboxes.json")
|
|
|
|
with patch("sandboxer.telemetry.introspection.HostInventoryScanner") as scanner_cls:
|
|
scanner = scanner_cls.return_value
|
|
scanner.scan_inventory.return_value = SandboxInventory(
|
|
host="h1", base_dir="/tmp/sandboxer", collected_at=now, entries=[]
|
|
)
|
|
scanner.find_stale.return_value = []
|
|
report = build_introspection_report(
|
|
host="h1",
|
|
sandbox_id="abc",
|
|
profile=profile,
|
|
provision_before=snap,
|
|
provision_after=snap2,
|
|
store=store,
|
|
)
|
|
|
|
assert report.provision_delta is not None
|
|
assert report.provision_delta.load_1m_delta == pytest.approx(0.5)
|
|
assert report.provision_delta.mem_available_mb_delta == -100 |