87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
"""Tests for app/context/wizard.py state machine."""
|
||
import sqlite3
|
||
import pytest
|
||
from pathlib import Path
|
||
from app.context.wizard import get_schema, advance_step, is_complete, apply_session, TOTAL_STEPS
|
||
|
||
|
||
@pytest.fixture
|
||
def db(tmp_path):
|
||
db_path = tmp_path / "t.db"
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.executescript("""
|
||
CREATE TABLE context_facts (
|
||
id TEXT PRIMARY KEY, category TEXT NOT NULL, key TEXT NOT NULL,
|
||
value TEXT NOT NULL, source TEXT, created_at TEXT NOT NULL
|
||
);
|
||
""")
|
||
conn.commit()
|
||
conn.close()
|
||
return db_path
|
||
|
||
|
||
def test_get_schema_returns_steps():
|
||
schema = get_schema()
|
||
assert len(schema) == TOTAL_STEPS
|
||
assert all("step" in s and "id" in s and "title" in s for s in schema)
|
||
|
||
|
||
def test_advance_step_records_answer():
|
||
session: dict = {"current_step": 1, "answers": {}}
|
||
updated = advance_step(session, "os", "Linux (systemd/journald)")
|
||
assert updated["answers"]["os"] == "Linux (systemd/journald)"
|
||
assert updated["current_step"] == 2
|
||
|
||
|
||
def test_advance_step_is_immutable():
|
||
session: dict = {"current_step": 1, "answers": {}}
|
||
updated = advance_step(session, "os", "Linux (systemd/journald)")
|
||
assert session["current_step"] == 1 # original unchanged
|
||
|
||
|
||
def test_is_complete_false_mid_wizard():
|
||
session = {"current_step": 3, "answers": {}}
|
||
assert is_complete(session) is False
|
||
|
||
|
||
def test_is_complete_true_after_all_steps():
|
||
session = {"current_step": TOTAL_STEPS + 1, "answers": {}}
|
||
assert is_complete(session) is True
|
||
|
||
|
||
def test_apply_session_writes_hostname_fact(db):
|
||
session = {
|
||
"current_step": TOTAL_STEPS + 1,
|
||
"answers": {
|
||
"os": "Linux (systemd/journald)",
|
||
"hostname": "heimdall.local",
|
||
"services": "plex.service, sonarr.service",
|
||
"docker": "Yes — Docker",
|
||
"syslog": "No",
|
||
},
|
||
}
|
||
result = apply_session(db, session)
|
||
assert result["facts_written"] >= 2
|
||
assert result["source_count"] >= 3 # journald×2 + docker
|
||
|
||
conn = sqlite3.connect(str(db))
|
||
facts = conn.execute("SELECT key, value FROM context_facts").fetchall()
|
||
conn.close()
|
||
keys = [f[0] for f in facts]
|
||
assert "hostname" in keys
|
||
|
||
|
||
def test_apply_session_no_services(db):
|
||
session = {
|
||
"current_step": TOTAL_STEPS + 1,
|
||
"answers": {
|
||
"hostname": "strahl",
|
||
"os": "Linux (systemd/journald)",
|
||
"services": "",
|
||
"docker": "No",
|
||
"syslog": "No",
|
||
},
|
||
}
|
||
result = apply_session(db, session)
|
||
# At least one journald source (catch-all), no docker, no syslog
|
||
assert result["source_count"] == 1
|