xavierk
ef61ececc1
Add a b key and a no-baseline banner that open a form for rated TBW and an optional source URL. The value is persisted as an unverified override through fenris-monitor baseline set under the existing polkit action, and the outcome is reported synchronously. Verified overrides stay CLI-only (ADR 0012 section 7).
215 lines
8.7 KiB
Python
215 lines
8.7 KiB
Python
"""Enter rated TBW from the TUI (issue #107, ADR 0012 §7).
|
|
|
|
Seam: the app's helper invocation (`_run_helper`), faked here the way the
|
|
other TUI tests fake it. The form builds an unverified override and hands it
|
|
to `fenris-monitor baseline set`; it never offers verified provenance.
|
|
"""
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from fenris.monitor import cmd_baseline_set
|
|
from fenris.store import init_store
|
|
from fenris.tui import FenrisTuiApp
|
|
from test_staged_projection import seed_evidence
|
|
|
|
SERVICES = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
return tmp_path / "observations.db"
|
|
|
|
|
|
def _app(db):
|
|
return FenrisTuiApp(store_path=db)
|
|
|
|
|
|
def _seed_without_baseline(db):
|
|
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
|
|
conn = init_store(db)
|
|
seed_evidence(conn, 3, baseline=None, start=now - timedelta(hours=3))
|
|
conn.close()
|
|
|
|
|
|
def _text(app, selector):
|
|
return str(app.query_one(selector).render())
|
|
|
|
|
|
async def _open_form(pilot):
|
|
await pilot.press("b")
|
|
await pilot.pause()
|
|
|
|
|
|
async def _fill(pilot, app, tbw, url=""):
|
|
app.screen.query_one("#tbw-input").value = tbw
|
|
app.screen.query_one("#source-input").value = url
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
|
|
|
|
class TestForm:
|
|
@pytest.mark.asyncio
|
|
async def test_b_opens_form_with_required_tbw_and_optional_source(self, db):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
await _open_form(pilot)
|
|
assert len(app.screen_stack) == 2
|
|
assert app.screen.query_one("#tbw-input")
|
|
assert app.screen.query_one("#source-input")
|
|
text = str(app.screen.query_one("#baseline-header").render())
|
|
assert "unverified" in text.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("value", ["", "abc", "0", "-5", "nan", "inf"])
|
|
async def test_invalid_tbw_is_rejected_without_calling_helper(self, db, value):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
with patch.object(app, "_run_helper") as run_helper:
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
await _open_form(pilot)
|
|
await _fill(pilot, app, value)
|
|
assert len(app.screen_stack) == 2
|
|
assert str(app.screen.query_one("#baseline-error").render()).strip()
|
|
run_helper.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_source_url_is_rejected(self, db):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
with patch.object(app, "_run_helper") as run_helper:
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
await _open_form(pilot)
|
|
await _fill(pilot, app, "600", "not a url")
|
|
assert len(app.screen_stack) == 2
|
|
assert "url" in str(app.screen.query_one("#baseline-error").render()).lower()
|
|
run_helper.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_escape_cancels_without_calling_helper(self, db):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
with patch.object(app, "_run_helper") as run_helper:
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
await _open_form(pilot)
|
|
await pilot.press("escape")
|
|
await pilot.pause()
|
|
assert len(app.screen_stack) == 1
|
|
run_helper.assert_not_called()
|
|
|
|
|
|
class TestSubmit:
|
|
@pytest.mark.asyncio
|
|
async def test_submit_routes_unverified_override_to_baseline_set(self, db):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
with patch.object(app, "_run_helper", return_value=True) as run_helper:
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
await _open_form(pilot)
|
|
await _fill(pilot, app, "600", "https://example.com/spec")
|
|
assert len(app.screen_stack) == 1
|
|
run_helper.assert_called_once()
|
|
operation, args = run_helper.call_args.args
|
|
assert operation == "baseline"
|
|
assert args[0] == "set"
|
|
payload = json.loads(args[1])
|
|
assert payload["tbw_terabytes"] == 600
|
|
assert payload["source_url"] == "https://example.com/spec"
|
|
assert payload["verified"] is False
|
|
assert payload["validated_by"] == "user"
|
|
# Incomplete provenance keeps the override unverified.
|
|
assert payload["document_revision"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_blank_source_is_stored_as_absent(self, db):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
with patch.object(app, "_run_helper", return_value=True) as run_helper:
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
await _open_form(pilot)
|
|
await _fill(pilot, app, "1.5")
|
|
payload = json.loads(run_helper.call_args.args[1][1])
|
|
assert payload["tbw_terabytes"] == 1.5
|
|
assert payload["source_url"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_headline_changes_from_write_rate_to_lifespan(self, db):
|
|
_seed_without_baseline(db)
|
|
|
|
def fake_helper(operation, extra_args=None):
|
|
cmd_baseline_set(SimpleNamespace(store_path=db, baseline_json=extra_args[1]))
|
|
return True
|
|
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
with patch.object(app, "_run_helper", side_effect=fake_helper):
|
|
async with app.run_test(size=(120, 50)) as pilot:
|
|
before = _text(app, "#headline-band")
|
|
assert "Write rate" in before
|
|
assert "remaining" not in before
|
|
await _open_form(pilot)
|
|
await _fill(pilot, app, "600")
|
|
app.on_refresh_tick()
|
|
await pilot.pause()
|
|
after = _text(app, "#headline-band")
|
|
assert "remaining" in after
|
|
assert "Write rate" not in after
|
|
|
|
|
|
class TestBanner:
|
|
@pytest.mark.asyncio
|
|
async def test_banner_shown_when_no_baseline_applies(self, db):
|
|
_seed_without_baseline(db)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
async with app.run_test(size=(120, 50)):
|
|
banner = app.query_one("#baseline-banner")
|
|
assert banner.display
|
|
assert "b" in str(banner.render()).lower()
|
|
assert "rated tbw" in str(banner.render()).lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_banner_shown_on_first_run(self, tmp_path):
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(tmp_path / "missing.db")
|
|
async with app.run_test(size=(120, 50)):
|
|
assert app.query_one("#baseline-banner").display
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_banner_hidden_once_baseline_exists(self, db):
|
|
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
|
|
conn = init_store(db)
|
|
seed_evidence(conn, 3, start=now - timedelta(hours=3))
|
|
conn.close()
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
async with app.run_test(size=(120, 50)):
|
|
assert not app.query_one("#baseline-banner").display
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_banner_hidden_for_unreadable_store(self, db):
|
|
db.write_bytes(b"not a sqlite database" * 100)
|
|
with patch("fenris.status.query_service_state", return_value=SERVICES):
|
|
app = _app(db)
|
|
async with app.run_test(size=(120, 50)):
|
|
assert not app.query_one("#baseline-banner").display
|