311 lines
10 KiB
Python
311 lines
10 KiB
Python
"""Tests for fenris-monitor helper.
|
|
|
|
Spec: §8.4, §8.5, §8.6, §8.7
|
|
"""
|
|
import json
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from fenris.monitor import (
|
|
cmd_enable,
|
|
cmd_disable,
|
|
cmd_collect,
|
|
cmd_baseline_set,
|
|
cmd_baseline_clear,
|
|
is_root,
|
|
)
|
|
from fenris.store import init_store
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_store(tmp_path):
|
|
"""Create a temporary observation store."""
|
|
store_path = tmp_path / "observations.db"
|
|
conn = init_store(store_path)
|
|
yield conn
|
|
conn.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def store_path(tmp_path):
|
|
"""Return path to a temporary observation store."""
|
|
return tmp_path / "observations.db"
|
|
|
|
|
|
class TestIsRoot:
|
|
def test_root_returns_true(self):
|
|
with patch("os.geteuid", return_value=0):
|
|
assert is_root() is True
|
|
|
|
def test_non_root_returns_false(self):
|
|
with patch("os.geteuid", return_value=1000):
|
|
assert is_root() is False
|
|
|
|
|
|
class TestEnableIdempotentMatrix:
|
|
"""§8.6: Period-row idempotent matrix."""
|
|
|
|
def test_first_opens_period(self, store_path):
|
|
"""First-ever enable opens a period at the enable moment."""
|
|
# Initialize store
|
|
init_store(store_path)
|
|
|
|
args = MagicMock(now=False, store_path=store_path)
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(returncode=0)
|
|
cmd_enable(args)
|
|
|
|
# Period should be open
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute(
|
|
"SELECT ended_at FROM monitoring_periods WHERE ended_at IS NULL"
|
|
)
|
|
assert cursor.fetchone() is not None
|
|
conn.close()
|
|
|
|
def test_resume_with_open_period_noop(self, store_path):
|
|
"""Resume with open period: no-op (gap stays inside as unknown)."""
|
|
# Initialize store and open a period
|
|
conn = init_store(store_path)
|
|
now = datetime.now(timezone.utc)
|
|
conn.execute(
|
|
"INSERT INTO monitoring_periods (started_at) VALUES (?)",
|
|
(now.isoformat(),),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
args = MagicMock(now=True, store_path=store_path)
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(returncode=0)
|
|
cmd_enable(args)
|
|
|
|
# Should still have exactly one open period
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute(
|
|
"SELECT COUNT(*) FROM monitoring_periods WHERE ended_at IS NULL"
|
|
)
|
|
assert cursor.fetchone()[0] == 1
|
|
conn.close()
|
|
|
|
def test_resume_with_no_period_opens_new(self, store_path):
|
|
"""Resume with no open period opens a new row."""
|
|
# Initialize store and close any existing period
|
|
conn = init_store(store_path)
|
|
conn.execute(
|
|
"UPDATE monitoring_periods SET ended_at = ?, end_cause = ?",
|
|
(datetime.now(timezone.utc).isoformat(), "user_disabled"),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
args = MagicMock(now=True, store_path=store_path)
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(returncode=0)
|
|
cmd_enable(args)
|
|
|
|
# Should have a new open period
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute(
|
|
"SELECT COUNT(*) FROM monitoring_periods WHERE ended_at IS NULL"
|
|
)
|
|
assert cursor.fetchone()[0] == 1
|
|
conn.close()
|
|
|
|
|
|
class TestDisableIdempotentMatrix:
|
|
"""§8.6: Period-row idempotent matrix."""
|
|
|
|
def test_pause_with_open_period_closes_user_disabled(self, store_path):
|
|
"""Pause with open period closes it user_disabled."""
|
|
# Initialize store and open a period
|
|
conn = init_store(store_path)
|
|
now = datetime.now(timezone.utc)
|
|
conn.execute(
|
|
"INSERT INTO monitoring_periods (started_at) VALUES (?)",
|
|
(now.isoformat(),),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
args = MagicMock(now=True, store_path=store_path)
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(returncode=0)
|
|
cmd_disable(args)
|
|
|
|
# Period should be closed with user_disabled
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute(
|
|
"SELECT end_cause FROM monitoring_periods WHERE ended_at IS NOT NULL"
|
|
)
|
|
assert cursor.fetchone()[0] == "user_disabled"
|
|
conn.close()
|
|
|
|
def test_pause_without_open_period_noop(self, store_path):
|
|
"""Pause otherwise no-ops."""
|
|
# Initialize store
|
|
init_store(store_path)
|
|
|
|
args = MagicMock(now=True, store_path=store_path)
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(returncode=0)
|
|
cmd_disable(args)
|
|
|
|
# No periods should exist
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute("SELECT COUNT(*) FROM monitoring_periods")
|
|
assert cursor.fetchone()[0] == 0
|
|
conn.close()
|
|
|
|
def test_raw_systemctl_stop_never_records_user_disabled(self, store_path):
|
|
"""Raw systemctl stop outside helper never records user_disabled."""
|
|
# Initialize store and open a period
|
|
conn = init_store(store_path)
|
|
now = datetime.now(timezone.utc)
|
|
conn.execute(
|
|
"INSERT INTO monitoring_periods (started_at) VALUES (?)",
|
|
(now.isoformat(),),
|
|
)
|
|
conn.commit()
|
|
|
|
# Simulate raw systemctl stop (no monitor involved)
|
|
# The period stays open - only the sanctioned path closes it
|
|
cursor = conn.execute(
|
|
"SELECT end_cause FROM monitoring_periods WHERE ended_at IS NULL"
|
|
)
|
|
assert cursor.fetchone() is not None # Still open
|
|
conn.close()
|
|
|
|
|
|
class TestCollectTrigger:
|
|
"""§8.7: On-demand collection via helper path."""
|
|
|
|
def test_collect_triggers_systemctl_start(self):
|
|
"""Collect starts fenris-collect.service synchronously."""
|
|
args = MagicMock()
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(returncode=0)
|
|
cmd_collect(args)
|
|
|
|
mock_sub.run.assert_called_once_with(
|
|
["systemctl", "start", "fenris-collect.service"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
def test_collect_failure_exits_nonzero(self):
|
|
"""Collect failure exits with nonzero status."""
|
|
args = MagicMock()
|
|
|
|
with patch("fenris.monitor.subprocess") as mock_sub:
|
|
mock_sub.run.return_value = MagicMock(
|
|
returncode=1, stderr="Unit not found"
|
|
)
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
cmd_collect(args)
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
class TestBaselinePersistence:
|
|
"""PR-14: Baseline persistence behind polkit-guarded helper."""
|
|
|
|
def test_baseline_set_persists(self, store_path):
|
|
"""baseline set persists the baseline row."""
|
|
# Initialize store
|
|
init_store(store_path)
|
|
|
|
args = MagicMock(
|
|
store_path=store_path,
|
|
baseline_json=json.dumps(
|
|
{
|
|
"tbw_terabytes": 600,
|
|
"source_url": "https://example.com/spec",
|
|
"document_revision": "rev1",
|
|
"entry_date": "2024-01-01",
|
|
"model_string": "Samsung 990 Pro",
|
|
"nominal_capacity_bytes": 2000000000000,
|
|
}
|
|
)
|
|
)
|
|
|
|
cmd_baseline_set(args)
|
|
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute("SELECT * FROM endurance_baseline")
|
|
row = cursor.fetchone()
|
|
assert row is not None
|
|
assert row[1] == 600.0 # tbw_terabytes
|
|
conn.close()
|
|
|
|
def test_baseline_set_replaces_existing(self, store_path):
|
|
"""baseline set replaces any existing baseline."""
|
|
# Initialize store and insert initial baseline
|
|
conn = init_store(store_path)
|
|
now = datetime.now(timezone.utc)
|
|
conn.execute(
|
|
"INSERT INTO endurance_baseline (tbw_terabytes, source_url, "
|
|
"document_revision, entry_date, model_string, nominal_capacity_bytes, "
|
|
"created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(400, "old", "v1", "2023-01-01", "Old Model", 1000000000000,
|
|
now.isoformat(), now.isoformat()),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
args = MagicMock(
|
|
store_path=store_path,
|
|
baseline_json=json.dumps(
|
|
{
|
|
"tbw_terabytes": 600,
|
|
"source_url": "new",
|
|
"document_revision": "v2",
|
|
"entry_date": "2024-01-01",
|
|
"model_string": "New Model",
|
|
"nominal_capacity_bytes": 2000000000000,
|
|
}
|
|
)
|
|
)
|
|
|
|
cmd_baseline_set(args)
|
|
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute("SELECT COUNT(*) FROM endurance_baseline")
|
|
assert cursor.fetchone()[0] == 1 # Only one row
|
|
conn.close()
|
|
|
|
def test_baseline_clear_removes(self, store_path):
|
|
"""baseline clear removes the baseline."""
|
|
# Initialize store and insert baseline
|
|
conn = init_store(store_path)
|
|
now = datetime.now(timezone.utc)
|
|
conn.execute(
|
|
"INSERT INTO endurance_baseline (tbw_terabytes, source_url, "
|
|
"document_revision, entry_date, model_string, nominal_capacity_bytes, "
|
|
"created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(400, "src", "v1", "2024-01-01", "Model", 1000000000000,
|
|
now.isoformat(), now.isoformat()),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
args = MagicMock(store_path=store_path)
|
|
cmd_baseline_clear(args)
|
|
|
|
conn = init_store(store_path)
|
|
cursor = conn.execute("SELECT COUNT(*) FROM endurance_baseline")
|
|
assert cursor.fetchone()[0] == 0
|
|
conn.close()
|