258 lines
9.1 KiB
Python
258 lines
9.1 KiB
Python
"""Bound pending-publication admission before device acquisition."""
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from fenris import collector
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self, now):
|
|
self.now = now
|
|
|
|
def utcnow(self):
|
|
return self.now
|
|
|
|
|
|
@pytest.fixture
|
|
def sysfs_controller(tmp_path):
|
|
controller_path = tmp_path / "sys" / "class" / "nvme" / "nvme0"
|
|
controller_path.mkdir(parents=True)
|
|
(controller_path / "subsysnqn").write_text("nqn.test:drive\n")
|
|
(controller_path / "model").write_text("Test NVMe\n")
|
|
(controller_path / "serial").write_text("test-serial\n")
|
|
(controller_path / "firmware_rev").write_text("1.0\n")
|
|
transport = controller_path / "transport"
|
|
transport.mkdir()
|
|
(transport / "trstring").write_text("pcie\n")
|
|
return controller_path
|
|
|
|
|
|
def _smartctl(units_written):
|
|
return {
|
|
"nvme_smart_health_information_log": {
|
|
"critical_warning": 0,
|
|
"temperature": 35,
|
|
"available_spare": 100,
|
|
"percentage_used": 5,
|
|
"data_units_written": units_written,
|
|
"data_units_read": 500,
|
|
"power_on_hours": 100,
|
|
},
|
|
"user_capacity": {"bytes": 1_024_000_000_000},
|
|
"model_name": "Test NVMe",
|
|
"serial_number": "test-serial",
|
|
"firmware_version": "1.0",
|
|
}
|
|
|
|
|
|
def _collect(config, controller, now, units_written):
|
|
return collector.run_collection(
|
|
config=config,
|
|
clock=FakeClock(now),
|
|
acquire=lambda: (_smartctl(units_written), controller),
|
|
)
|
|
|
|
|
|
def test_partial_recovery_frees_admission_and_keeps_pending_order(
|
|
tmp_path, sysfs_controller, monkeypatch,
|
|
):
|
|
"""A freed slot admits one observation behind older pending evidence."""
|
|
assert collector.PENDING_PUBLICATION_LIMIT == 6_720
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
monkeypatch.setattr(collector, "PENDING_PUBLICATION_LIMIT", 2)
|
|
|
|
store_path = tmp_path / "observations.db"
|
|
config = {"device": "/dev/nvme0", "store_path": str(store_path)}
|
|
old_time = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
|
controller = sysfs_controller
|
|
|
|
assert _collect(config, controller, old_time, 1000)["ok"] is True
|
|
|
|
original_derive = collector.derive_hours_from_interval
|
|
|
|
def unavailable_derivation(*_args):
|
|
raise RuntimeError("derivation unavailable")
|
|
|
|
monkeypatch.setattr(collector, "derive_hours_from_interval", unavailable_derivation)
|
|
first_pending = _collect(
|
|
config, controller, old_time + timedelta(minutes=5), 1010,
|
|
)
|
|
assert first_pending["ok"] is False
|
|
|
|
second_pending = _collect(
|
|
config, controller, old_time + timedelta(minutes=10), 1020,
|
|
)
|
|
assert second_pending["ok"] is False
|
|
with sqlite3.connect(store_path) as conn:
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM pending_publications"
|
|
).fetchone()[0] == 2
|
|
|
|
called = False
|
|
|
|
def must_not_acquire():
|
|
nonlocal called
|
|
called = True
|
|
return _smartctl(1030), controller
|
|
|
|
full = collector.run_collection(
|
|
config=config,
|
|
clock=FakeClock(old_time + timedelta(days=40)),
|
|
acquire=must_not_acquire,
|
|
)
|
|
assert full["ok"] is False
|
|
assert "capacity full (2 observations)" in full["error"]
|
|
assert called is False
|
|
with sqlite3.connect(store_path) as conn:
|
|
assert conn.execute(
|
|
"SELECT COUNT(*), MIN(sample_ts) FROM pending_publications"
|
|
).fetchone() == (
|
|
2,
|
|
(old_time + timedelta(minutes=5)).isoformat(),
|
|
)
|
|
assert conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0] == 1
|
|
|
|
def fail_second_pending(conn, previous, current):
|
|
if current["data_units_written"] == 1020:
|
|
raise RuntimeError("second observation still blocked")
|
|
return original_derive(conn, previous, current)
|
|
|
|
monkeypatch.setattr(collector, "derive_hours_from_interval", fail_second_pending)
|
|
acquired = False
|
|
|
|
def acquire_after_partial_recovery():
|
|
nonlocal acquired
|
|
acquired = True
|
|
return _smartctl(1030), controller
|
|
|
|
partial = collector.run_collection(
|
|
config=config,
|
|
clock=FakeClock(old_time + timedelta(days=40, minutes=5)),
|
|
acquire=acquire_after_partial_recovery,
|
|
)
|
|
assert partial["ok"] is False
|
|
assert acquired is True, partial
|
|
assert "no new observation acquired" not in partial["error"]
|
|
|
|
with sqlite3.connect(store_path) as conn:
|
|
assert conn.execute(
|
|
"SELECT data_units_written FROM samples ORDER BY id"
|
|
).fetchall() == [(1000,), (1010,)]
|
|
queued_units = [
|
|
json.loads(row[0])["sample"]["data_units_written"]
|
|
for row in conn.execute(
|
|
"SELECT payload FROM pending_publications ORDER BY id"
|
|
)
|
|
]
|
|
assert queued_units == [1020, 1030]
|
|
assert conn.execute("SELECT COUNT(*) FROM monitoring_periods").fetchone()[0] == 1
|
|
|
|
recovered_in_order = []
|
|
|
|
def record_recovery_order(conn, previous, current):
|
|
recovered_in_order.append(current["data_units_written"])
|
|
return original_derive(conn, previous, current)
|
|
|
|
monkeypatch.setattr(collector, "derive_hours_from_interval", record_recovery_order)
|
|
recovered = collector.run_collection(
|
|
config=config,
|
|
clock=FakeClock(old_time + timedelta(days=40, minutes=10)),
|
|
acquire=lambda: (_smartctl(1040), controller),
|
|
)
|
|
assert recovered["ok"] is True, recovered
|
|
assert recovered_in_order == [1020, 1030, 1040]
|
|
|
|
with sqlite3.connect(store_path) as conn:
|
|
assert conn.execute(
|
|
"SELECT data_units_written FROM samples ORDER BY id DESC LIMIT 1"
|
|
).fetchone() == (1040,)
|
|
assert conn.execute("SELECT COUNT(*) FROM pending_publications").fetchone()[0] == 0
|
|
|
|
|
|
def test_actual_pending_limit_fails_before_acquisition_and_keeps_old_evidence(
|
|
tmp_path, sysfs_controller, monkeypatch,
|
|
):
|
|
"""The production limit retains every old row and refuses device acquisition."""
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
store_path = tmp_path / "full-observations.db"
|
|
config = {"device": "/dev/nvme0", "store_path": str(store_path)}
|
|
controller = sysfs_controller
|
|
old_time = datetime(2026, 7, 1, tzinfo=timezone.utc)
|
|
assert _collect(config, controller, old_time, 1000)["ok"] is True
|
|
|
|
def unavailable_derivation(*_args):
|
|
raise RuntimeError("derivation unavailable")
|
|
|
|
monkeypatch.setattr(collector, "derive_hours_from_interval", unavailable_derivation)
|
|
failed = _collect(
|
|
config, controller, old_time + timedelta(minutes=5), 1010,
|
|
)
|
|
assert failed["ok"] is False
|
|
|
|
with sqlite3.connect(store_path) as conn:
|
|
payload_row = conn.execute(
|
|
"SELECT payload FROM pending_publications ORDER BY id LIMIT 1"
|
|
).fetchone()
|
|
assert payload_row is not None
|
|
base = json.loads(payload_row[0])
|
|
base_observed_at = datetime.fromisoformat(base["sample"]["ts"])
|
|
base_units = base["sample"]["data_units_written"]
|
|
rows = []
|
|
for offset in range(1, collector.PENDING_PUBLICATION_LIMIT):
|
|
observation = json.loads(payload_row[0])
|
|
units_written = base_units + offset
|
|
observed_at = base_observed_at + timedelta(minutes=5 * offset)
|
|
observation["sample"]["ts"] = observed_at.isoformat()
|
|
observation["sample"]["data_units_written"] = units_written
|
|
observation["sample"]["bytes_written"] = units_written * 512_000
|
|
rows.append((observed_at.isoformat(), json.dumps(observation)))
|
|
conn.executemany(
|
|
"INSERT INTO pending_publications (sample_ts, payload) VALUES (?, ?)",
|
|
rows,
|
|
)
|
|
conn.commit()
|
|
|
|
called = False
|
|
|
|
def must_not_acquire():
|
|
nonlocal called
|
|
called = True
|
|
return _smartctl(9000), controller
|
|
|
|
result = collector.run_collection(
|
|
config=config,
|
|
clock=FakeClock(old_time + timedelta(days=40)),
|
|
acquire=must_not_acquire,
|
|
)
|
|
assert result["ok"] is False
|
|
assert f"capacity full ({collector.PENDING_PUBLICATION_LIMIT} observations)" in result[
|
|
"error"
|
|
]
|
|
assert called is False
|
|
|
|
with sqlite3.connect(store_path) as conn:
|
|
pending_rows = conn.execute(
|
|
"SELECT sample_ts, payload FROM pending_publications ORDER BY id"
|
|
).fetchall()
|
|
assert len(pending_rows) == collector.PENDING_PUBLICATION_LIMIT
|
|
assert (pending_rows[0][0], pending_rows[-1][0]) == (
|
|
(old_time + timedelta(minutes=5)).isoformat(),
|
|
(
|
|
old_time
|
|
+ timedelta(minutes=5 * collector.PENDING_PUBLICATION_LIMIT)
|
|
).isoformat(),
|
|
)
|
|
assert [
|
|
json.loads(payload)["sample"]["data_units_written"]
|
|
for _, payload in pending_rows
|
|
] == list(range(1010, 1010 + collector.PENDING_PUBLICATION_LIMIT))
|
|
assert conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0] == 1
|
|
assert conn.execute("SELECT COUNT(*) FROM monitoring_periods").fetchone()[0] == 1
|