feat(tui): enter rated TBW from the dashboard
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).
This commit is contained in:
xavierk
@@ -0,0 +1,214 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user