from tests.test_web import (ADMIN, VIEWER, admin, app, client, csrf_from, login_ok, post, # noqa: F401 users, viewer) TOKEN = "syt_SECRET-token.abc" def route_form(**overrides): data = {"name": "Web an Telegram", "match_host": "web*", "match_service": "", "min_state": "2", "kinds": ["problem", "recovery"], "targets": ["telegram|-100123"], "extra_targets": "", "enabled": "1"} data.update(overrides) return data def enable_telegram(app): store = app.state.store store.set("telegram.enabled", "1", "test") store.set("telegram.chat_id", "-100123", "test") def routes(app): return app.state.runtime.routes.list_all() def test_routing_requires_login(client): assert client.get("/routing").status_code == 303 def test_viewer_sees_list_but_cannot_change(client, app, viewer): login_ok(client, VIEWER) page = client.get("/routing") assert page.status_code == 200 and "Regel anlegen" not in page.text token = csrf_from(client, "/routing") assert client.post("/routing", data={"csrf": token, **route_form()}).status_code == 403 assert client.get("/routing/1").status_code == 403 assert routes(app) == [] def test_admin_creates_edits_toggles_deletes(client, app, admin): enable_telegram(app) login_ok(client) assert "telegram: -100123" in client.get("/routing").text assert post(client, "/routing", "/routing", **route_form()).status_code == 303 created = routes(app)[0] assert created.name == "Web an Telegram" and created.min_state == 2 assert [(t.backend, t.target) for t in created.targets] == [("telegram", "-100123")] page = client.get(f"/routing/{created.id}") assert page.status_code == 200 and "web*" in page.text response = post(client, f"/routing/{created.id}", f"/routing/{created.id}", **route_form(name="Neu", extra_targets="matrix:!room:example.de")) assert response.status_code == 303 updated = routes(app)[0] assert updated.name == "Neu" and len(updated.targets) == 2 post(client, f"/routing/{created.id}/toggle", "/routing", enabled="0") assert routes(app)[0].enabled is False post(client, f"/routing/{created.id}/delete", "/routing") assert routes(app) == [] actions = [r["action"] for r in app.state.db.query("SELECT action FROM audit")] assert {"route.create", "route.update", "route.toggle", "route.delete"} <= set(actions) audit_new = app.state.db.query("SELECT new_value FROM audit WHERE action='route.update'")[0]["new_value"] assert "Neu" in audit_new def test_move_changes_order(client, app, admin): login_ok(client) post(client, "/routing", "/routing", **route_form(name="A")) post(client, "/routing", "/routing", **route_form(name="B")) second = routes(app)[1] post(client, f"/routing/{second.id}/move", "/routing", direction="up") assert [r.name for r in routes(app)] == ["B", "A"] post(client, f"/routing/{second.id}/move", "/routing", direction="up") assert [r.name for r in routes(app)] == ["B", "A"] def test_invalid_input_rejected(client, app, admin): login_ok(client) for override in ({"name": ""}, {"min_state": "9"}, {"min_state": "x"}, {"kinds": []}, {"targets": ["telegram|abc"]}, {"extra_targets": "keinbackend"}, {"extra_targets": "signal:123"}, {"match_host": "a" * 300}): response = post(client, "/routing", "/routing", **route_form(**override)) assert response.status_code == 422, override assert routes(app) == [] def test_unknown_route_returns_404(client, admin): login_ok(client) assert client.get("/routing/99").status_code == 404 def test_matrix_settings_encrypt_token(client, app, admin): login_ok(client) form = {"matrix_enabled": "1", "matrix_homeserver": "https://matrix.example.de", "matrix_user_id": "@bot:example.de", "matrix_token": TOKEN, "matrix_room_id": "!abc:example.de", "matrix_allowlist": "@a:example.de, @b:example.de"} assert post(client, "/settings/matrix", "/settings/matrix", **form).status_code == 303 store = app.state.store assert store.get("matrix.token") == TOKEN raw = app.state.db.query("SELECT value, is_secret FROM settings WHERE key='matrix.token'")[0] assert raw["is_secret"] == 1 and TOKEN not in raw["value"] assert TOKEN not in client.get("/settings/matrix").text assert TOKEN not in str(app.state.db.query("SELECT old_value, new_value FROM audit")) def test_matrix_invalid_values_rejected(client, app, admin): login_ok(client) base = {"matrix_enabled": "1", "matrix_homeserver": "https://matrix.example.de", "matrix_user_id": "@bot:example.de", "matrix_token": TOKEN, "matrix_room_id": "!abc:example.de"} for override in ({"matrix_homeserver": "ftp://x"}, {"matrix_user_id": "bot"}, {"matrix_room_id": "#alias:x"}, {"matrix_allowlist": "nope"}, {"matrix_token": "a b"}): response = post(client, "/settings/matrix", "/settings/matrix", **{**base, **override}) assert response.status_code == 422, override assert app.state.store.get("matrix.homeserver") is None def test_matrix_enabled_requires_all_fields(client, admin): login_ok(client) response = post(client, "/settings/matrix", "/settings/matrix", matrix_enabled="1") assert response.status_code == 422 and "Access-Token fehlt" in response.text def test_matrix_optional_in_setup_and_test_message_admin_only(client, viewer, admin): login_ok(client) page = client.get("/setup") assert "Matrix" in page.text and "optional" in page.text def test_matrix_test_message_requires_admin(client, viewer): login_ok(client, VIEWER) token = csrf_from(client, "/") assert client.post("/settings/matrix/message", data={"csrf": token}).status_code == 403 def test_matrix_test_message_without_config_reports_error(client, admin): login_ok(client) response = post(client, "/settings/matrix/message", "/settings/matrix", matrix_room_id="!abc:example.de") assert response.status_code == 200 and response.json()["ok"] is False def test_alerting_section_saves_and_validates(client, app, admin): login_ok(client) form = {"poll_interval": "45", "alerts_default_delay_minutes": "10", "alerts_reminder_hours": "6", "alerts_recovery_enabled": "1", "alerts_bundle_seconds": "90", "alerts_ai_per_minute": "5"} assert post(client, "/settings/alerting", "/settings/alerting", **form).status_code == 303 assert app.state.store.get_int("poll.interval", 30) == 45 assert app.state.store.get_int("alerts.ai_per_minute", 10) == 5 assert app.state.store.get_int("alerts.default_delay_minutes", 15) == 10 assert app.state.store.get_int("alerts.reminder_hours", 0) == 6 for override in ({"poll_interval": "1"}, {"poll_interval": "abc"}, {"alerts_default_delay_minutes": "-3"}, {"alerts_default_delay_minutes": "1441"}, {"alerts_reminder_hours": "169"}, {"alerts_bundle_seconds": "99999"}, {"alerts_ai_per_minute": "1.5"}): response = post(client, "/settings/alerting", "/settings/alerting", **{**form, **override}) assert response.status_code == 422, override assert app.state.store.get_int("poll.interval", 30) == 45 def test_alerting_has_no_connection_test(client, admin): login_ok(client) assert "data-test-url=\"/settings/alerting/test\"" not in client.get("/settings/alerting").text assert post(client, "/settings/alerting/test", "/settings/alerting").status_code == 404 def test_health_includes_matrix_only_when_enabled(app): from app.health import build_checks assert "matrix" not in build_checks(app.state.store) app.state.store.set("matrix.enabled", "1", "test") assert "matrix" in build_checks(app.state.store)