mqtt-broker-stack/backend/tests/test_api.py

"""API tests: authentication, RBAC, client + ACL flows (spec section 64)."""

from __future__ import annotations

from tests.conftest import login


def test_health_live(client):
    response = client.get("/health/live")
    assert response.status_code == 200


def test_login_sets_cookie_and_session(client):
    response = login(client, "admin", "admin-password-123")
    assert response.json()["role"] == "super_admin"

    session = client.get("/api/v1/auth/session")
    assert session.status_code == 200
    assert session.json()["username"] == "admin"


def test_login_wrong_password_audited(client):
    response = client.post(
        "/api/v1/auth/login", json={"username": "admin", "password": "wrong"}
    )
    assert response.status_code == 401


def test_unauthenticated_api_rejected(client):
    response = client.get("/api/v1/mqtt/clients")
    assert response.status_code == 401


def test_auditor_cannot_create_clients(auditor_client):
    response = auditor_client.post(
        "/api/v1/mqtt/clients", json={"username": "ps042"}
    )
    assert response.status_code == 403


def test_create_client_returns_password_once(admin_client):
    response = admin_client.post("/api/v1/mqtt/clients", json={"username": "ps042"})
    assert response.status_code == 201
    body = response.json()
    assert body["username"] == "ps042"
    assert len(body["password"]) >= 20  # random password generated

    listing = admin_client.get("/api/v1/mqtt/clients")
    assert any(c["username"] == "ps042" for c in listing.json())


def test_disable_enable_delete_client(admin_client):
    admin_client.post("/api/v1/mqtt/clients", json={"username": "sensor-1"})

    response = admin_client.post("/api/v1/mqtt/clients/sensor-1/disable")
    assert response.status_code == 204
    clients = {c["username"]: c for c in admin_client.get("/api/v1/mqtt/clients").json()}
    assert clients["sensor-1"]["enabled"] is False

    assert admin_client.post("/api/v1/mqtt/clients/sensor-1/enable").status_code == 204

    response = admin_client.delete("/api/v1/mqtt/clients/sensor-1")
    assert response.status_code == 204


def test_role_acl_lifecycle(admin_client):
    admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})

    acls = [
        {
            "acltype": "publishClientSend",
            "topic": "site/%u/telemetry/#",
            "allow": True,
            "priority": 0,
        },
        {
            "acltype": "subscribePattern",
            "topic": "site/%u/config/#",
            "allow": True,
            "priority": 1,
        },
    ]
    response = admin_client.put("/api/v1/mqtt/roles/telemetry", json={"acls": acls})
    assert response.status_code == 200

    roles = {r["rolename"]: r for r in admin_client.get("/api/v1/mqtt/roles").json()}
    assert roles["telemetry"]["acls"] == acls


def test_group_lifecycle(admin_client):
    response = admin_client.post(
        "/api/v1/mqtt/groups", json={"groupname": "sensors"}
    )
    assert response.status_code == 201

    groups = {g["groupname"]: g for g in admin_client.get("/api/v1/mqtt/groups").json()}
    assert "sensors" in groups

    assert admin_client.delete("/api/v1/mqtt/groups/sensors").status_code == 204


def test_client_role_assignment(admin_client):
    admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})
    admin_client.post("/api/v1/mqtt/clients", json={"username": "assign-test"})

    response = admin_client.post(
        "/api/v1/mqtt/clients/assign-test/roles",
        json={"rolename": "telemetry", "priority": 5},
    )
    assert response.status_code == 204

    client = admin_client.get("/api/v1/mqtt/clients/assign-test").json()
    roles = {r["rolename"] for r in client["roles"]}
    assert "telemetry" in roles

    assert (
        admin_client.delete("/api/v1/mqtt/clients/assign-test/roles/telemetry").status_code
        == 204
    )
    client = admin_client.get("/api/v1/mqtt/clients/assign-test").json()
    assert all(r["rolename"] != "telemetry" for r in client["roles"])


def test_group_member_assignment(admin_client):
    admin_client.post("/api/v1/mqtt/groups", json={"groupname": "sensors"})
    admin_client.post("/api/v1/mqtt/clients", json={"username": "member-test"})

    response = admin_client.post(
        "/api/v1/mqtt/groups/sensors/clients",
        json={"username": "member-test", "priority": 1},
    )
    assert response.status_code == 204

    client = admin_client.get("/api/v1/mqtt/clients/member-test").json()
    groups = {g["groupname"] for g in client["groups"]}
    assert "sensors" in groups

    assert (
        admin_client.delete("/api/v1/mqtt/groups/sensors/clients/member-test").status_code
        == 204
    )
    client = admin_client.get("/api/v1/mqtt/clients/member-test").json()
    assert all(g["groupname"] != "sensors" for g in client["groups"])


def test_group_role_assignment(admin_client):
    admin_client.post("/api/v1/mqtt/groups", json={"groupname": "sensors"})
    admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})

    response = admin_client.post(
        "/api/v1/mqtt/groups/sensors/roles",
        json={"rolename": "telemetry", "priority": 2},
    )
    assert response.status_code == 204

    assert (
        admin_client.delete("/api/v1/mqtt/groups/sensors/roles/telemetry").status_code
        == 204
    )


def test_listener_management_and_preview(admin_client):
    response = admin_client.post(
        "/api/v1/broker/listeners",
        json={"name": "Production MQTT", "port": 1883, "protocol": "mqtt"},
    )
    assert response.status_code == 201
    listener_id = response.json()["id"]

    preview = admin_client.get("/api/v1/broker/config/preview")
    assert preview.status_code == 200
    config_text = preview.json()["config_text"]
    assert "listener 1883" in config_text
    # Generated config must not use options deprecated in 2.1 (spec 3.1).
    assert "password_file" not in config_text
    assert "acl_file" not in config_text
    assert "per_listener_settings" not in config_text

    delete = admin_client.delete(f"/api/v1/broker/listeners/{listener_id}")
    assert delete.status_code == 204


def test_audit_events_recorded(admin_client, db_session):
    from sqlalchemy import select

    from app.models import AuditEvent

    admin_client.post("/api/v1/mqtt/clients", json={"username": "audited-1"})
    actions = db_session.scalars(select(AuditEvent.action)).all()
    assert "CLIENT_CREATED" in actions