import json from unittest.mock import patch import pytest from django.test import Client from finance.assistant import model_intent from finance.models import Audit, FormulaDefinition, FormulaExport, Membership pytestmark = pytest.mark.django_db def test_model_intent_accepts_only_valid_dimensions_and_sends_no_rows(workspace, settings): ds, _, _, _, _ = workspace settings.OPENAI_MODEL = "test-model" settings.OPENAI_API_KEY = "test-key" class Response: def raise_for_status(self): return None def json(self): return { "status": "completed", "output": [ { "content": [ { "type": "output_text", "text": json.dumps( { "action": "overview", "metric": "revenue", "group": "department", "filters": [{"key": "department", "value": "儿科"}], } ), } ] } ], } with patch("finance.assistant.httpx.Client") as http: http.return_value.__enter__.return_value.post.return_value = Response() intent = model_intent("看儿科收入", ds, {"metric": "revenue", "group": "department", "filters": {}}) assert intent == { "action": "overview", "metric": "revenue", "group": "department", "filters": {"department": "儿科"}, } payload = http.return_value.__enter__.return_value.post.call_args.kwargs["json"] assert payload["store"] is False assert "C1" not in json.dumps(payload) def test_model_intent_rejects_unknown_filter_and_incomplete_provider_response(workspace, settings): ds, *_ = workspace settings.OPENAI_MODEL = "test-model" settings.OPENAI_API_KEY = "test-key" class Response: def raise_for_status(self): return None def json(self): return { "status": "completed", "output": [ { "content": [ { "type": "output_text", "text": '{"action":"overview","metric":"revenue","group":"department","filters":[{"key":"__class__","value":"x"}]}', } ] } ], } response = Response() with patch("finance.assistant.httpx.Client") as http: http.return_value.__enter__.return_value.post.return_value = response with pytest.raises(Exception, match="无效意图"): model_intent("分析收入", ds, {"metric": "revenue", "group": "department", "filters": {}}) with patch("finance.assistant.httpx.Client") as http: http.return_value.__enter__.return_value.post.return_value = type( "Response", (), {"raise_for_status": lambda self: None, "json": lambda self: {"status": "in_progress"}}, )() with pytest.raises(Exception, match="无效意图"): model_intent("分析收入", ds, {"metric": "revenue", "group": "department", "filters": {}}) def test_assistant_local_response_and_external_consent_guard(workspace, settings): ds, _, client, *_ = workspace url = f"/api/datasets/{ds.pk}/assistant" payload = { "question": "查看实际收入", "context": {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"}, } response = client.post(url, data=payload, content_type="application/json") assert response.status_code == 200 assert response.json()["provider"] == "local" assert Audit.objects.filter(action="assistant.query").exists() settings.AI_ENABLED = True settings.OPENAI_API_KEY = "test-key" settings.OPENAI_MODEL = "test-model" denied = client.post( url, data={**payload, "use_model": True, "consent": False}, content_type="application/json", ) assert denied.status_code == 400 assert "同意" in denied.json()["error"] def test_formula_design_and_export_api_validate_consent_and_scope(workspace, settings): ds, _, client, *_ = workspace url = f"/api/datasets/{ds.pk}" payload = { "context": {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"}, "messages": [{"role": "user", "content": "收入成本差"}], "consent": False, } denied = client.post(f"{url}/formula-design", data=payload, content_type="application/json") assert denied.status_code == 400 export = client.post( f"{url}/formula-exports", data={**payload["context"], "custom_columns": [{"expression": {}}]}, content_type="application/json", ) assert export.status_code == 400 assert FormulaExport.objects.count() == 0 scoped = Client() user = ds.created_by user.membership.departments = ["儿科"] user.membership.save(update_fields=["departments"]) assert scoped.login(username=user.username, password="Finance-TestPassword991!") forbidden = scoped.post( f"{url}/formula-exports", data=payload["context"], content_type="application/json" ) assert forbidden.status_code == 403 assert scoped.get(f"{url}/formula-definitions").status_code == 403 def test_formula_definition_library_is_private_validated_and_audited(workspace): ds, _, client, _, _ = workspace url = f"/api/datasets/{ds.pk}/formula-definitions" definition = { "name": "收入成本差", "meaning": "实际收入减成本", "scope": "所选分组与期间", "missing": "结果不可用（#N/A）；零分母不可用", "expression": { "operator": "subtract", "left": {"metric": "revenue"}, "right": {"metric": "cost"}, }, } saved = client.post(url, data={"definition": definition}, content_type="application/json") assert saved.status_code == 201 definition_id = saved.json()["id"] assert saved.json()["revision"] == 1 assert client.get(url).json()["items"][0]["definition"] == definition updated = client.post( url, data={ "definition": {**definition, "name": " 收入成本差 ", "meaning": "更新说明"}, "revision": 1, }, content_type="application/json", ) assert updated.status_code == 200 assert FormulaDefinition.objects.filter(dataset=ds, user=ds.created_by).count() == 1 assert updated.json()["definition"]["meaning"] == "更新说明" assert updated.json()["revision"] == 2 other = Client() other_user = ds.created_by.__class__.objects.create_user( "formula-library-other", password="Formula-Library-TestPassword991!" ) Membership.objects.create(user=other_user, organization=ds.organization, role="analyst") assert other.login(username=other_user.username, password="Formula-Library-TestPassword991!") assert other.get(url).json()["items"] == [] assert other.delete(f"{url}/{definition_id}").status_code == 404 deleted = client.delete(f"{url}/{definition_id}") assert deleted.status_code == 200 assert not FormulaDefinition.objects.filter(pk=definition_id).exists() assert Audit.objects.filter(action="formula_definition.save").count() == 2 assert Audit.objects.filter(action="formula_definition.delete").count() == 1 def test_formula_definition_library_rejects_unavailable_metric(workspace): ds, _, client, *_ = workspace mapping = {**ds.mapping, "columns": [c for c in ds.mapping["columns"] if c["role"] != "cost"]} ds.mapping = mapping ds.save(update_fields=["mapping"]) response = client.post( f"/api/datasets/{ds.pk}/formula-definitions", data={ "definition": { "name": "收入成本差", "meaning": "实际收入减成本", "scope": "所选分组与期间", "missing": "结果不可用（#N/A）；零分母不可用", "expression": { "operator": "subtract", "left": {"metric": "revenue"}, "right": {"metric": "cost"}, }, } }, content_type="application/json", ) assert response.status_code == 400 assert "成本" in response.json()["error"]