108 lines
4.4 KiB
Python
108 lines
4.4 KiB
Python
import json
|
|
|
|
from app.constants import (RETENTION_AGENT_RUNS_DAYS, RETENTION_ALERTS_DAYS, RETENTION_AUDIT_DAYS,
|
|
RETENTION_MUTES_DAYS, RETENTION_STATE_LOG_DAYS)
|
|
from app.retention import RetentionBusy
|
|
from tests.test_web import (VIEWER, admin, app, client, login_ok, post, # noqa: F401
|
|
users, viewer)
|
|
|
|
PAGE = "/settings/retention"
|
|
|
|
|
|
def form(**overrides):
|
|
data = {"retention_alerts_days": "90", "retention_agent_runs_days": "90", "retention_state_log_days": "30",
|
|
"retention_audit_days": "365", "retention_mutes_days": "7"}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def seed_old_rows(app):
|
|
db = app.state.db
|
|
db.insert("INSERT INTO audit(ts, user, action, target) VALUES (?,?,?,?)", (1, "u", "alt", "t"))
|
|
db.insert("INSERT INTO state_log(host, service, state, ts) VALUES (?,?,?,?)", ("h", "s", 1, 1))
|
|
db.insert("INSERT INTO agent_runs(ts, backend, sender, command, question, answer) VALUES (?,?,?,?,?,?)",
|
|
(1, "telegram", "1", "ask", "q", "a"))
|
|
|
|
|
|
def test_section_is_admin_only(client, viewer):
|
|
login_ok(client, VIEWER)
|
|
assert client.get(PAGE).status_code == 403
|
|
assert post(client, "/retention/run", "/prompts/alert").status_code == 403
|
|
|
|
|
|
def test_section_shows_defaults_hints_and_no_run_yet(client, admin):
|
|
login_ok(client)
|
|
page = client.get(PAGE)
|
|
assert page.status_code == 200 and "Aufbewahrung" in page.text
|
|
for expected in ('value="90"', 'value="30"', 'value="365"', 'value="7"', "Noch kein Lauf", "Jetzt aufräumen",
|
|
"03:00", "Service-Zustände"):
|
|
assert expected in page.text, expected
|
|
assert 'href="/settings/retention"' in page.text
|
|
|
|
|
|
def test_setup_lists_retention_as_optional_step(client, admin):
|
|
login_ok(client)
|
|
page = client.get("/setup")
|
|
assert "Aufbewahrung" in page.text and "/settings/retention" in page.text
|
|
|
|
|
|
def test_save_valid_values(client, app, admin):
|
|
login_ok(client)
|
|
response = post(client, PAGE, PAGE, **form(retention_alerts_days="30", retention_mutes_days="1"))
|
|
assert response.status_code == 303
|
|
store = app.state.store
|
|
assert store.get(RETENTION_ALERTS_DAYS) == "30" and store.get(RETENTION_MUTES_DAYS) == "1"
|
|
assert 'value="30"' in client.get(PAGE).text
|
|
|
|
|
|
def test_invalid_values_are_rejected_and_not_saved(client, app, admin):
|
|
login_ok(client)
|
|
for key, bad in (("retention_alerts_days", "0"), ("retention_agent_runs_days", "-3"),
|
|
("retention_state_log_days", "abc"), ("retention_audit_days", "99999"),
|
|
("retention_mutes_days", "400")):
|
|
response = post(client, PAGE, PAGE, **form(**{key: bad}))
|
|
assert response.status_code == 422 and "nichts versehentlich ausgeschaltet" in response.text, key
|
|
store = app.state.store
|
|
for key in (RETENTION_ALERTS_DAYS, RETENTION_AGENT_RUNS_DAYS, RETENTION_STATE_LOG_DAYS,
|
|
RETENTION_AUDIT_DAYS, RETENTION_MUTES_DAYS):
|
|
assert store.get(key) is None
|
|
|
|
|
|
def test_run_now_deletes_reports_and_audits(client, app, admin):
|
|
seed_old_rows(app)
|
|
login_ok(client)
|
|
response = post(client, "/retention/run", PAGE)
|
|
assert response.status_code == 303 and response.headers["location"] == PAGE
|
|
page = client.get(PAGE)
|
|
assert "Aufgeräumt. Gelöscht:" in page.text and "Letzter Lauf" in page.text
|
|
assert "Audit-Einträge: 1" in page.text and "Statusverlauf: 1" in page.text and "manuell" in page.text
|
|
db = app.state.db
|
|
assert db.query("SELECT COUNT(*) FROM agent_runs")[0][0] == 0
|
|
rows = db.query("SELECT user, new_value FROM audit WHERE action='retention.run'")
|
|
assert len(rows) == 1 and rows[0]["user"] == "admin1" and json.loads(rows[0]["new_value"])["agent_runs"] == 1
|
|
|
|
|
|
def test_run_now_requires_csrf(client, admin):
|
|
login_ok(client)
|
|
assert client.post("/retention/run").status_code == 403
|
|
|
|
|
|
def test_run_now_busy_and_failure_messages(client, app, admin, monkeypatch):
|
|
login_ok(client)
|
|
service = app.state.runtime.retention
|
|
|
|
def busy(*args, **kwargs):
|
|
raise RetentionBusy
|
|
|
|
monkeypatch.setattr(service, "run", busy)
|
|
post(client, "/retention/run", PAGE)
|
|
assert "läuft bereits" in client.get(PAGE).text
|
|
|
|
def broken(*args, **kwargs):
|
|
raise RuntimeError("interner-fehlertext")
|
|
|
|
monkeypatch.setattr(service, "run", broken)
|
|
post(client, "/retention/run", PAGE)
|
|
page = client.get(PAGE)
|
|
assert "Aufräumen fehlgeschlagen" in page.text and "interner-fehlertext" not in page.text
|