pigway-frame-flow/tests/test_accounts.py

import hashlib
import time

from pigway_frame_flow.core.accounts import password_hash


def account(app):
    app.state.service.db.execute(
        "INSERT INTO users VALUES(?,?,1)",
        ("test-user", password_hash("fixture-password")),
    )


def login(client, app, remember=True):
    account(app)
    client.headers.pop("Authorization", None)
    return client.post(
        "/api/v1/auth/login",
        json={
            "username": "test-user",
            "password": "fixture-password",
            "remember": remember,
        },
    )


def test_session_login_and_csrf(client, app):
    response = login(client, app)
    assert response.status_code == 200
    assert "HttpOnly" in response.headers["set-cookie"]
    assert "SameSite=strict" in response.headers["set-cookie"]
    assert client.get("/api/v1/devices").status_code == 200
    assert client.post("/api/v1/jobs", json={}).status_code == 403
    assert client.get("/api/v1/auth/me").json()["username"] == "test-user"
    client.headers["X-CSRF-Token"] = response.json()["csrf_token"]
    assert client.post("/api/v1/auth/logout").status_code == 204
    assert client.get("/api/v1/devices").status_code == 401


def test_password_change_revokes_sessions(client, app):
    response = login(client, app)
    client.headers["X-CSRF-Token"] = response.json()["csrf_token"]
    assert (
        client.put(
            "/api/v1/auth/password",
            json={"current_password": "wrong", "new_password": "replacement-password"},
        ).status_code
        == 401
    )
    assert (
        client.put(
            "/api/v1/auth/password",
            json={
                "current_password": "fixture-password",
                "new_password": "replacement-password",
            },
        ).status_code
        == 204
    )
    assert client.get("/api/v1/auth/me").status_code == 401
    assert (
        client.post(
            "/api/v1/auth/login",
            json={"username": "test-user", "password": "replacement-password"},
        ).status_code
        == 200
    )


def test_login_rate_limit(client, app):
    account(app)
    for _ in range(5):
        assert (
            client.post(
                "/api/v1/auth/login",
                json={"username": "test-user", "password": "wrong"},
            ).status_code
            == 401
        )
    assert (
        client.post(
            "/api/v1/auth/login",
            json={"username": "test-user", "password": "fixture-password"},
        ).status_code
        == 429
    )


def test_session_expiry_and_no_plaintext(client, app):
    response = login(client, app, False)
    assert "Max-Age" not in response.headers["set-cookie"]
    raw = client.cookies.get("frameflow_session")
    row = app.state.service.db.one("SELECT * FROM sessions")
    assert row["token_hash"] == hashlib.sha256(raw.encode()).hexdigest()
    app.state.service.db.execute("UPDATE sessions SET expires_at=?", (time.time() - 1,))
    assert client.get("/api/v1/auth/me").status_code == 401
    assert "fixture-password" not in str(
        app.state.service.db.one("SELECT * FROM users")
    )


def test_reset_template(client):
    assert (
        client.put(
            "/api/v1/templates/quote/config", json={"data": {"title": "Saved title"}}
        ).status_code
        == 200
    )
    assert (
        client.get("/api/v1/templates/quote").json()["defaults"]["title"]
        == "Saved title"
    )
    assert client.delete("/api/v1/templates/quote/config").status_code == 204
    assert (
        client.get("/api/v1/templates/quote").json()["defaults"]["title"]
        != "Saved title"
    )


def test_openapi_describes_cookie_auth_and_docs_csrf(client):
    schema = client.get("/openapi.json").json()
    cookie = schema["components"]["securitySchemes"]["AccountSession"]
    assert cookie["in"] == "cookie" and cookie["name"] == "frameflow_session"
    assert {"AccountSession": []} in schema["paths"]["/api/v1/devices"]["get"][
        "security"
    ]
    assert "X-CSRF-Token" in client.get("/docs").text


def test_browser_admin_can_read_other_clients_operation(client, app, service):
    operation_id = service.submit(
        "render",
        {"template": "quote"},
        "living-room",
        "scheduler",
        "fixture-request",
        None,
    )
    assert login(client, app).status_code == 200
    assert client.get("/api/v1/operations/" + operation_id).status_code == 200
    assert operation_id in [
        op["operation_id"] for op in client.get("/api/v1/operations").json()
    ]