checkbot/tests/test_web_routing.py

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)