fa-agent/tests/test_views_workflows.py

import json

import pytest
from django.contrib.auth.models import User
from django.test import Client

from finance.models import Audit, Investigation, Membership


def test_api_auth_method_and_request_shape_guards(workspace):
    ds, _, client, *_ = workspace
    anonymous = Client()
    assert anonymous.get(f"/api/datasets/{ds.pk}").status_code == 401
    assert client.get(f"/api/datasets/{ds.pk}/query").status_code == 405
    assert (
        client.post(f"/api/datasets/{ds.pk}/query", data="[]", content_type="application/json").status_code
        == 400
    )
    assert (
        client.post(f"/api/datasets/{ds.pk}/query", data="{}", content_type="text/plain").status_code == 400
    )
    bad_login = anonymous.post(
        "/api/login", data=json.dumps({"username": "x", "password": "wrong"}), content_type="application/json"
    )
    assert bad_login.status_code == 401
    assert anonymous.post("/api/logout").status_code == 401


pytestmark = pytest.mark.django_db


def test_session_login_logout_and_health_contract(workspace):
    ds, _, _, _, _ = workspace
    client = Client()
    assert client.get("/api/health").json() == {"status": "ok"}
    session = client.get("/api/session").json()
    assert session["user"] is None
    login = client.post(
        "/api/login",
        data=json.dumps({"username": ds.created_by.username, "password": "Finance-TestPassword991!"}),
        content_type="application/json",
    )
    assert login.status_code == 200
    assert client.get("/api/session").json()["user"]["role"] == "admin"
    assert client.post("/api/logout").json() == {"ok": True}
    assert client.get("/api/session").json()["user"] is None
    assert Audit.objects.filter(action__in=["auth.login", "auth.logout"]).count() == 2


def test_user_management_enforces_org_scope_and_self_protection(workspace):
    ds, _, client, _, _ = workspace
    base = {"role": "analyst", "departments": ["儿科"]}
    created = client.post(
        "/api/users",
        data=json.dumps({**base, "username": "managed-user", "password": "Managed-TestPassword991!"}),
        content_type="application/json",
    )
    assert created.status_code == 200
    managed = User.objects.get(username="managed-user")
    membership = Membership.objects.get(user=managed)
    assert membership.organization == ds.organization
    assert membership.departments == ["儿科"]

    denied = client.post(
        "/api/users",
        data=json.dumps({**base, "username": "weak-user", "password": "123"}),
        content_type="application/json",
    )
    assert denied.status_code == 400
    assert (
        client.patch(
            "/api/users",
            data=json.dumps({"id": ds.created_by_id, "active": False, **base}),
            content_type="application/json",
        ).status_code
        == 400
    )
    assert (
        client.patch(
            "/api/users",
            data=json.dumps({"id": managed.pk, "active": False, "role": "viewer", "departments": []}),
            content_type="application/json",
        ).status_code
        == 200
    )
    managed.refresh_from_db()
    membership.refresh_from_db()
    assert not managed.is_active
    assert membership.role == "viewer"

    assert client.get("/api/users").status_code == 200
    assert client.get("/api/audit").status_code == 200
    assert Audit.objects.filter(action="user.update").count() == 2


def test_investigation_lifecycle_and_viewer_write_denial(workspace):
    ds, _, client, _, _ = workspace
    context = {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"}
    url = f"/api/datasets/{ds.pk}/investigations"
    saved = client.post(
        url,
        data=json.dumps({**context, "status": "reviewing", "note": "复核收入成本"}),
        content_type="application/json",
    )
    assert saved.status_code == 200
    assert Investigation.objects.filter(dataset=ds, status="reviewing", note="复核收入成本").exists()
    assert client.get(url).json()["items"][0]["status"] == "reviewing"
    invalid = client.post(
        url,
        data=json.dumps({**context, "status": "closed", "note": ""}),
        content_type="application/json",
    )
    assert invalid.status_code == 400

    viewer = User.objects.create_user("finance-viewer", password="Finance-TestPassword991!")
    Membership.objects.create(user=viewer, organization=ds.organization, role="viewer")
    readonly = Client()
    assert readonly.login(username=viewer.username, password="Finance-TestPassword991!")
    denied = readonly.post(
        url,
        data=json.dumps({**context, "status": "resolved", "note": ""}),
        content_type="application/json",
    )
    assert denied.status_code == 403


def test_evidence_rejects_bad_pagination_and_exports_exact_xlsx(workspace):
    ds, _, client, *_ = workspace
    url = f"/api/datasets/{ds.pk}/evidence"
    invalid = client.post(
        url,
        data=json.dumps({"start": "2026-08", "end": "2026-08", "limit": 201}),
        content_type="application/json",
    )
    assert invalid.status_code == 400
    exported = client.post(
        f"/api/datasets/{ds.pk}/export",
        data=json.dumps({"start": "2026-08", "end": "2026-08", "format": "xlsx", "group": "department"}),
        content_type="application/json",
    )
    assert exported.status_code == 200
    assert exported.streaming
    assert exported["Content-Disposition"].endswith('"traceiq-evidence.xlsx"')
    assert b"".join(exported.streaming_content).startswith(b"PK")