466 lines
15 KiB
Python
466 lines
15 KiB
Python
"""Controller segmentation tests.
|
|
|
|
Covers acceptance criteria:
|
|
- ID-1: Identity key ladder, FR metadata only, independent axes
|
|
- ID-2: Frozen metadata snapshot at segment open
|
|
- ID-4: identity_degraded set exactly when key is blank
|
|
- AC-4: vid/ssvid from PCI node, stored null otherwise
|
|
- PR-16: Blank-key semantics (to/from blank quarantines, equal blanks continue)
|
|
"""
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
import pytest
|
|
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from fenris.store import init_store
|
|
from fenris.collector import normalize_identity, compute_identity_degraded, acquire_from_sysfs
|
|
from fenris.segment import find_current_segment, should_open_new_segment, open_segment
|
|
|
|
|
|
# Fixtures
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Path) -> sqlite3.Connection:
|
|
conn = init_store(tmp_path / "observations.db")
|
|
yield conn
|
|
conn.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def clock():
|
|
class FakeClock:
|
|
def __init__(self):
|
|
self.now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
def utcnow(self):
|
|
return self.now
|
|
return FakeClock()
|
|
|
|
|
|
@pytest.fixture
|
|
def identity_nqn() -> Dict[str, Any]:
|
|
return {
|
|
"subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678-1234-1234-1234-123456789abc",
|
|
"mn": "Samsung SSD 970 EVO Plus 1TB",
|
|
"sn": "S4EWNX0N123456",
|
|
"fr": "2B2QEXM7",
|
|
"vid": "0x144d",
|
|
"ssvid": "0x144d",
|
|
"transport": "pcie",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def identity_model_serial() -> Dict[str, Any]:
|
|
return {
|
|
"subnqn": "",
|
|
"mn": "Samsung SSD 970 EVO Plus 1TB",
|
|
"sn": "S4EWNX0N123456",
|
|
"fr": "2B2QEXM7",
|
|
"vid": "0x144d",
|
|
"ssvid": "0x144d",
|
|
"transport": "pcie",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def identity_degraded() -> Dict[str, Any]:
|
|
return {
|
|
"subnqn": "",
|
|
"mn": "",
|
|
"sn": "",
|
|
"fr": "2B2QEXM7",
|
|
"vid": "0x144d",
|
|
"ssvid": "0x144d",
|
|
"transport": "pcie",
|
|
}
|
|
|
|
|
|
# ID-1: Identity key ladder
|
|
|
|
class TestIdentityKeyLadder:
|
|
def test_subnqn_primary(self, identity_nqn):
|
|
key = normalize_identity(identity_nqn)
|
|
assert key == "nqn.2014-08.org.nvmexpress:uuid:12345678-1234-1234-1234-123456789abc"
|
|
|
|
def test_fallback_to_model_serial(self, identity_model_serial):
|
|
key = normalize_identity(identity_model_serial)
|
|
assert key == "Samsung SSD 970 EVO Plus 1TB|S4EWNX0N123456"
|
|
|
|
def test_all_blank_degraded(self, identity_degraded):
|
|
key = normalize_identity(identity_degraded)
|
|
assert key == ""
|
|
|
|
def test_firmware_is_metadata_only(self, identity_nqn):
|
|
key = normalize_identity(identity_nqn)
|
|
assert "2B2QEXM7" not in key
|
|
|
|
|
|
# ID-2: Frozen metadata snapshot
|
|
|
|
class TestFrozenMetadataSnapshot:
|
|
def test_segment_stores_all_metadata_fields(self, store, clock, identity_nqn):
|
|
key = normalize_identity(identity_nqn)
|
|
degraded = compute_identity_degraded(identity_nqn)
|
|
|
|
segment = open_segment(store, clock.now, identity_nqn, key, degraded)
|
|
|
|
assert segment["identity_key"] == key
|
|
assert segment["identity_degraded"] is False
|
|
assert segment["subnqn"] == identity_nqn["subnqn"]
|
|
assert segment["sn"] == identity_nqn["sn"]
|
|
assert segment["mn"] == identity_nqn["mn"]
|
|
assert segment["fr"] == identity_nqn["fr"]
|
|
assert segment["vid"] == identity_nqn["vid"]
|
|
assert segment["ssvid"] == identity_nqn["ssvid"]
|
|
assert segment["transport"] == identity_nqn["transport"]
|
|
|
|
def test_metadata_nullable(self, store, clock):
|
|
sparse_identity = {
|
|
"subnqn": "",
|
|
"mn": "Legacy Model",
|
|
"sn": "LEGACY123",
|
|
"fr": "1.0",
|
|
"vid": None,
|
|
"ssvid": None,
|
|
"transport": None,
|
|
}
|
|
|
|
key = normalize_identity(sparse_identity)
|
|
degraded = compute_identity_degraded(sparse_identity)
|
|
segment = open_segment(store, clock.now, sparse_identity, key, degraded)
|
|
|
|
assert segment["vid"] is None
|
|
assert segment["ssvid"] is None
|
|
assert segment["transport"] is None
|
|
|
|
def test_cntlid_excluded(self, store, clock, identity_nqn):
|
|
key = normalize_identity(identity_nqn)
|
|
degraded = compute_identity_degraded(identity_nqn)
|
|
|
|
segment = open_segment(store, clock.now, identity_nqn, key, degraded)
|
|
|
|
assert "cntlid" not in segment
|
|
|
|
def test_metadata_immutable_after_open(self, store, clock, identity_nqn):
|
|
key = normalize_identity(identity_nqn)
|
|
degraded = compute_identity_degraded(identity_nqn)
|
|
|
|
open_segment(store, clock.now, identity_nqn, key, degraded)
|
|
|
|
found = find_current_segment(store)
|
|
assert found["subnqn"] == identity_nqn["subnqn"]
|
|
assert found["vid"] == identity_nqn["vid"]
|
|
assert found["ssvid"] == identity_nqn["ssvid"]
|
|
|
|
|
|
# ID-4: identity_degraded
|
|
|
|
class TestIdentityDegraded:
|
|
def test_degraded_when_blank_key(self, store, clock, identity_degraded):
|
|
key = normalize_identity(identity_degraded)
|
|
degraded = compute_identity_degraded(identity_degraded)
|
|
|
|
assert key == ""
|
|
assert degraded is True
|
|
|
|
segment = open_segment(store, clock.now, identity_degraded, key, degraded)
|
|
assert segment["identity_degraded"] is True
|
|
|
|
def test_not_degraded_with_subnqn(self, store, clock, identity_nqn):
|
|
key = normalize_identity(identity_nqn)
|
|
degraded = compute_identity_degraded(identity_nqn)
|
|
|
|
assert key != ""
|
|
assert degraded is False
|
|
|
|
segment = open_segment(store, clock.now, identity_nqn, key, degraded)
|
|
assert segment["identity_degraded"] is False
|
|
|
|
def test_not_degraded_with_model_serial(self, store, clock, identity_model_serial):
|
|
key = normalize_identity(identity_model_serial)
|
|
degraded = compute_identity_degraded(identity_model_serial)
|
|
|
|
assert key != ""
|
|
assert degraded is False
|
|
|
|
segment = open_segment(store, clock.now, identity_model_serial, key, degraded)
|
|
assert segment["identity_degraded"] is False
|
|
|
|
|
|
# AC-4: vid/ssvid from PCI node
|
|
|
|
class TestVidSsvidAcquisition:
|
|
def test_vid_ssvid_from_pci_node(self, tmp_path: Path):
|
|
ctrl_dir = tmp_path / "nvme0"
|
|
ctrl_dir.mkdir()
|
|
|
|
(ctrl_dir / "subsysnqn").write_text("nqn.test\n")
|
|
(ctrl_dir / "model").write_text("Test Model\n")
|
|
(ctrl_dir / "serial").write_text("TEST123\n")
|
|
(ctrl_dir / "firmware_rev").write_text("1.0\n")
|
|
(ctrl_dir / "vendor").write_text("0x144d\n")
|
|
(ctrl_dir / "subsystem_vendor").write_text("0x144d\n")
|
|
|
|
identity = acquire_from_sysfs(ctrl_dir)
|
|
|
|
assert identity["vid"] == "0x144d"
|
|
assert identity["ssvid"] == "0x144d"
|
|
|
|
def test_vid_ssvid_null_when_absent(self, tmp_path: Path):
|
|
ctrl_dir = tmp_path / "nvme0"
|
|
ctrl_dir.mkdir()
|
|
|
|
(ctrl_dir / "subsysnqn").write_text("nqn.test\n")
|
|
(ctrl_dir / "model").write_text("Test Model\n")
|
|
(ctrl_dir / "serial").write_text("TEST123\n")
|
|
(ctrl_dir / "firmware_rev").write_text("1.0\n")
|
|
|
|
identity = acquire_from_sysfs(ctrl_dir)
|
|
|
|
assert identity["vid"] is None
|
|
assert identity["ssvid"] is None
|
|
|
|
def test_vid_ssvid_metadata_only(self, store, clock, identity_nqn):
|
|
identity_a = {**identity_nqn, "vid": "0x144d", "ssvid": "0x144d"}
|
|
identity_b = {**identity_nqn, "vid": "0xFFFF", "ssvid": "0xFFFF"}
|
|
|
|
key_a = normalize_identity(identity_a)
|
|
key_b = normalize_identity(identity_b)
|
|
|
|
assert key_a == key_b
|
|
|
|
|
|
# PR-16: Blank-key semantics
|
|
|
|
class TestBlankKeySemantics:
|
|
def test_to_blank_quarantines(self, store, clock):
|
|
healthy = {
|
|
"subnqn": "nqn.healthy",
|
|
"mn": "Model A", "sn": "SN1", "fr": "1.0",
|
|
"vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_healthy = normalize_identity(healthy)
|
|
degraded_healthy = compute_identity_degraded(healthy)
|
|
|
|
open_segment(store, clock.now, healthy, key_healthy, degraded_healthy)
|
|
current = find_current_segment(store)
|
|
|
|
blank = {
|
|
"subnqn": "", "mn": "", "sn": "",
|
|
"fr": "1.0", "vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_blank = normalize_identity(blank)
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key_blank, 1000, store
|
|
)
|
|
|
|
assert should_open is True
|
|
assert reason == "identity_change"
|
|
|
|
def test_from_blank_quarantines(self, store, clock):
|
|
blank = {
|
|
"subnqn": "", "mn": "", "sn": "",
|
|
"fr": "1.0", "vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_blank = normalize_identity(blank)
|
|
degraded_blank = compute_identity_degraded(blank)
|
|
|
|
open_segment(store, clock.now, blank, key_blank, degraded_blank)
|
|
current = find_current_segment(store)
|
|
|
|
healthy = {
|
|
"subnqn": "nqn.healthy",
|
|
"mn": "Model A", "sn": "SN1", "fr": "1.0",
|
|
"vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_healthy = normalize_identity(healthy)
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key_healthy, 1000, store
|
|
)
|
|
|
|
assert should_open is True
|
|
assert reason == "identity_change"
|
|
|
|
def test_equal_blanks_continue_by_duw(self, store, clock):
|
|
blank = {
|
|
"subnqn": "", "mn": "", "sn": "",
|
|
"fr": "1.0", "vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_blank = normalize_identity(blank)
|
|
degraded_blank = compute_identity_degraded(blank)
|
|
|
|
open_segment(store, clock.now, blank, key_blank, degraded_blank)
|
|
current = find_current_segment(store)
|
|
|
|
store.execute(
|
|
"INSERT INTO samples (ts, device, bytes_written) VALUES (?, ?, ?)",
|
|
("2026-09-01T12:00:00Z", "/dev/nvme0", 1000)
|
|
)
|
|
store.commit()
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key_blank, 2000, store
|
|
)
|
|
|
|
assert should_open is False
|
|
assert reason is None
|
|
|
|
def test_equal_blanks_duw_decrease_opens(self, store, clock):
|
|
blank = {
|
|
"subnqn": "", "mn": "", "sn": "",
|
|
"fr": "1.0", "vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_blank = normalize_identity(blank)
|
|
degraded_blank = compute_identity_degraded(blank)
|
|
|
|
open_segment(store, clock.now, blank, key_blank, degraded_blank)
|
|
current = find_current_segment(store)
|
|
|
|
store.execute(
|
|
"INSERT INTO samples (ts, device, bytes_written) VALUES (?, ?, ?)",
|
|
("2026-09-01T12:00:00Z", "/dev/nvme0", 2000)
|
|
)
|
|
store.commit()
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key_blank, 1000, store
|
|
)
|
|
|
|
assert should_open is True
|
|
assert reason == "duw_decrease"
|
|
|
|
|
|
# Independent segmentation axes
|
|
|
|
class TestIndependentAxes:
|
|
def test_identity_change_with_duw_increase(self, store, clock):
|
|
identity_a = {
|
|
"subnqn": "nqn.drive-a",
|
|
"mn": "Model A", "sn": "SN1", "fr": "1.0",
|
|
"vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_a = normalize_identity(identity_a)
|
|
degraded_a = compute_identity_degraded(identity_a)
|
|
|
|
open_segment(store, clock.now, identity_a, key_a, degraded_a)
|
|
current = find_current_segment(store)
|
|
|
|
store.execute(
|
|
"INSERT INTO samples (ts, device, bytes_written) VALUES (?, ?, ?)",
|
|
("2026-09-01T12:00:00Z", "/dev/nvme0", 2000)
|
|
)
|
|
store.commit()
|
|
|
|
identity_b = {
|
|
"subnqn": "nqn.drive-b",
|
|
"mn": "Model B", "sn": "SN2", "fr": "2.0",
|
|
"vid": "0x2", "ssvid": "0x2", "transport": "pcie",
|
|
}
|
|
key_b = normalize_identity(identity_b)
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key_b, 3000, store
|
|
)
|
|
|
|
assert should_open is True
|
|
assert reason == "identity_change"
|
|
|
|
def test_duw_decrease_same_identity(self, store, clock):
|
|
identity = {
|
|
"subnqn": "nqn.drive-a",
|
|
"mn": "Model A", "sn": "SN1", "fr": "1.0",
|
|
"vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key = normalize_identity(identity)
|
|
degraded = compute_identity_degraded(identity)
|
|
|
|
open_segment(store, clock.now, identity, key, degraded)
|
|
current = find_current_segment(store)
|
|
|
|
store.execute(
|
|
"INSERT INTO samples (ts, device, bytes_written) VALUES (?, ?, ?)",
|
|
("2026-09-01T12:00:00Z", "/dev/nvme0", 5000)
|
|
)
|
|
store.commit()
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key, 3000, store
|
|
)
|
|
|
|
assert should_open is True
|
|
assert reason == "duw_decrease"
|
|
|
|
|
|
# Segment lifecycle
|
|
|
|
class TestSegmentLifecycle:
|
|
def test_first_segment_always_opens(self, store):
|
|
key = "nqn.test"
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
None, key, 1000, store
|
|
)
|
|
|
|
assert should_open is True
|
|
assert reason == "first_segment"
|
|
|
|
def test_same_identity_duw_non_decreasing_continues(self, store, clock):
|
|
identity = {
|
|
"subnqn": "nqn.drive",
|
|
"mn": "Model", "sn": "SN1", "fr": "1.0",
|
|
"vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key = normalize_identity(identity)
|
|
degraded = compute_identity_degraded(identity)
|
|
|
|
open_segment(store, clock.now, identity, key, degraded)
|
|
current = find_current_segment(store)
|
|
|
|
store.execute(
|
|
"INSERT INTO samples (ts, device, bytes_written) VALUES (?, ?, ?)",
|
|
("2026-09-01T12:00:00Z", "/dev/nvme0", 1000)
|
|
)
|
|
store.commit()
|
|
|
|
should_open, reason = should_open_new_segment(
|
|
current, key, 1500, store
|
|
)
|
|
|
|
assert should_open is False
|
|
assert reason is None
|
|
|
|
def test_multiple_segments(self, store, clock):
|
|
identity_a = {
|
|
"subnqn": "nqn.drive-a",
|
|
"mn": "Model A", "sn": "SN1", "fr": "1.0",
|
|
"vid": "0x1", "ssvid": "0x1", "transport": "pcie",
|
|
}
|
|
key_a = normalize_identity(identity_a)
|
|
degraded_a = compute_identity_degraded(identity_a)
|
|
|
|
open_segment(store, clock.now, identity_a, key_a, degraded_a)
|
|
|
|
identity_b = {
|
|
"subnqn": "nqn.drive-b",
|
|
"mn": "Model B", "sn": "SN2", "fr": "2.0",
|
|
"vid": "0x2", "ssvid": "0x2", "transport": "pcie",
|
|
}
|
|
key_b = normalize_identity(identity_b)
|
|
degraded_b = compute_identity_degraded(identity_b)
|
|
|
|
open_segment(store, clock.now, identity_b, key_b, degraded_b)
|
|
|
|
cursor = store.execute("SELECT COUNT(*) FROM controller_segments")
|
|
count = cursor.fetchone()[0]
|
|
assert count == 2
|
|
|
|
current = find_current_segment(store)
|
|
assert current["identity_key"] == key_b
|