pigway-frame-flow/tests/test_history_management.py

import hashlib
import json

from pigway_frame_flow.core.accounts import password_hash


def make_op(service, n, status="COMPLETED", client_id="admin", action="render"):
    key = service.submit(action, {}, "living-room", client_id, f"request-{n}")
    service.update(key, status)
    service.db.execute(
        "UPDATE operations SET created_at=?,result=? WHERE id=?",
        (
            f"2026-10-{n:02d}T10:00:00+00:00",
            json.dumps({"image_name": f"photo-{n}.jpg"}),
            key,
        ),
    )
    return key


def test_pagination_filters_and_page_clamp(client, service):
    ids = [make_op(service, n, "FAILED" if n % 2 else "COMPLETED") for n in range(1, 7)]
    page = client.get("/api/v1/operation-history?page_size=2").json()
    assert page["total"] == 6
    assert [x["operation_id"] for x in page["items"]] == ids[-1:-3:-1]
    assert (
        client.get("/api/v1/operation-history?page=2&page_size=2").json()["items"][0][
            "operation_id"
        ]
        == ids[3]
    )
    assert (
        client.get("/api/v1/operation-history?page=99&page_size=2").json()["page"] == 3
    )
    assert client.get("/api/v1/operation-history?status=FAILED").json()["total"] == 3
    assert (
        client.get("/api/v1/operation-history?q=photo-3").json()["items"][0][
            "operation_id"
        ]
        == ids[2]
    )
    assert (
        client.get("/api/v1/operation-history?start=2026-10-03&end=2026-10-04").json()[
            "total"
        ]
        == 2
    )
    assert (
        client.get(
            "/api/v1/operation-history?start=2026-10-05&end=2026-10-01"
        ).status_code
        == 422
    )
    assert client.get("/api/v1/operation-history?device_id=other").json()["total"] == 0
    assert client.get("/api/v1/operation-history?action=display").json()["total"] == 0


def test_delete_preserves_receipts_and_protects_running(client, service):
    done = make_op(service, 1)
    running = make_op(service, 2, "QUEUED")
    assert (
        client.request(
            "DELETE",
            "/api/v1/operation-history",
            json={"operation_ids": [done, running]},
        ).status_code
        == 409
    )
    assert client.get("/api/v1/operation-history").json()["total"] == 2
    assert (
        client.request(
            "DELETE",
            "/api/v1/operation-history",
            json={"operation_ids": [done, "missing"]},
        ).status_code
        == 404
    )
    assert (
        client.request(
            "DELETE", "/api/v1/operation-history", json={"operation_ids": [done, done]}
        ).json()["deleted"]
        == 1
    )
    assert client.get("/api/v1/operation-history").json()["total"] == 1
    assert done not in [
        x["operation_id"] for x in client.get("/api/v1/operations").json()
    ]
    assert client.get("/api/v1/operations/" + done).status_code == 200
    assert service.db.one("SELECT id FROM operations WHERE id=?", (done,))
    assert (
        client.request(
            "DELETE", "/api/v1/operation-history", json={"operation_ids": [done]}
        ).json()["deleted"]
        == 0
    )


def test_history_permissions(client, service):
    mine = make_op(service, 1, client_id="reader")
    make_op(service, 2)
    service.db.execute(
        "INSERT INTO clients VALUES(?,?,?,?,?,?,?)",
        (
            "reader",
            hashlib.sha256(b"reader-token").hexdigest(),
            "Reader",
            json.dumps(["history:read"]),
            "",
            None,
            1,
        ),
    )
    client.headers["Authorization"] = "Bearer reader-token"
    result = client.get("/api/v1/operation-history").json()
    assert result["total"] == 1 and result["items"][0]["operation_id"] == mine
    assert (
        client.request(
            "DELETE", "/api/v1/operation-history", json={"operation_ids": [mine]}
        ).status_code
        == 403
    )
    client.headers.pop("Authorization")
    assert client.get("/api/v1/operation-history").status_code == 401


def test_history_delete_requires_browser_csrf(client, service):
    op = make_op(service, 1)
    service.db.execute(
        "INSERT INTO users VALUES(?,?,1)",
        ("fixture", password_hash("fixture-password")),
    )
    client.headers.pop("Authorization")
    login = client.post(
        "/api/v1/auth/login",
        json={"username": "fixture", "password": "fixture-password"},
    ).json()
    assert (
        client.request(
            "DELETE", "/api/v1/operation-history", json={"operation_ids": [op]}
        ).status_code
        == 403
    )
    assert (
        client.request(
            "DELETE",
            "/api/v1/operation-history",
            json={"operation_ids": [op]},
            headers={"X-CSRF-Token": login["csrf_token"]},
        ).status_code
        == 200
    )