511 lines
19 KiB
Python
511 lines
19 KiB
Python
import json
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
from django.test import Client
|
|
|
|
from finance.assistant import model_intent
|
|
from finance.models import AIConfig, Audit, FormulaDefinition, FormulaExport, Membership
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def completed_response(intent=None):
|
|
class Response:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {
|
|
"status": "completed",
|
|
"output": [
|
|
{
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": json.dumps(
|
|
intent
|
|
or {
|
|
"action": "overview",
|
|
"metric": "revenue",
|
|
"group": "department",
|
|
"filters": [{"key": "department", "value": "儿科"}],
|
|
}
|
|
),
|
|
}
|
|
]
|
|
}
|
|
],
|
|
}
|
|
|
|
return Response()
|
|
|
|
|
|
def test_model_intent_polls_in_progress_and_retries_invalid_response(workspace, settings):
|
|
ds, _, _, _, _ = workspace
|
|
settings.AI_ENABLED = True
|
|
settings.OPENAI_MODEL = "test-model"
|
|
settings.OPENAI_API_KEY = "test-key"
|
|
polled = {"n": 0, "posts": 0}
|
|
|
|
class InProgress:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"status": "in_progress", "id": "resp-1"}
|
|
|
|
class ClientMock:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def post(self, *args, **kwargs):
|
|
polled["posts"] += 1
|
|
# 第一次 POST 返回损坏文本（触发整体重试），第二次 POST 返回 in_progress
|
|
if polled["posts"] == 1:
|
|
|
|
class Bad:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {
|
|
"status": "completed",
|
|
"output": [{"content": [{"type": "output_text", "text": "not-json"}]}],
|
|
}
|
|
|
|
return Bad()
|
|
return InProgress()
|
|
|
|
def get(self, *args, **kwargs):
|
|
polled["n"] += 1
|
|
assert args[0].endswith("/responses/resp-1")
|
|
return completed_response(
|
|
{"action": "anomalies", "metric": "revenue", "group": "department", "filters": []}
|
|
)
|
|
|
|
with patch("finance.ai.httpx.Client", ClientMock):
|
|
intent = model_intent("看异常", ds, {"metric": "revenue", "group": "department", "filters": {}})
|
|
|
|
assert intent["action"] == "anomalies"
|
|
assert polled["n"] == 1
|
|
|
|
|
|
def test_ai_config_requires_full_admin_and_masks_key(workspace):
|
|
ds, member, client, _, _ = workspace
|
|
url = "/api/ai-config"
|
|
|
|
viewer = User.objects.create_user("ai-viewer", password="Finance-TestPassword991!")
|
|
Membership.objects.create(user=viewer, organization=ds.organization, role="viewer")
|
|
viewer_client = Client()
|
|
assert viewer_client.login(username="ai-viewer", password="Finance-TestPassword991!")
|
|
assert viewer_client.get(url).status_code == 403
|
|
|
|
member.departments = ["儿科"]
|
|
member.save(update_fields=["departments"])
|
|
scoped = Client()
|
|
assert scoped.login(username=member.user.username, password="Finance-TestPassword991!")
|
|
assert scoped.get(url).status_code == 403
|
|
member.departments = []
|
|
member.save(update_fields=["departments"])
|
|
|
|
assert client.get(url).json() == {
|
|
"enabled": False,
|
|
"provider": "local",
|
|
"base_url": "",
|
|
"model": "",
|
|
"api_key_set": False,
|
|
"updated_at": None,
|
|
}
|
|
|
|
invalid = [
|
|
{"enabled": True, "provider": "local", "base_url": "", "model": "qwen2.5:7b"},
|
|
{"enabled": True, "provider": "openai", "base_url": "", "model": "gpt-4o"},
|
|
{"enabled": True, "provider": "local", "base_url": "ftp://x", "model": "m"},
|
|
{"enabled": True, "provider": "local", "base_url": "http://x/v1", "model": ""},
|
|
]
|
|
for payload in invalid:
|
|
assert client.put(url, data=payload, content_type="application/json").status_code == 400
|
|
|
|
saved = client.put(
|
|
url,
|
|
data={
|
|
"enabled": True,
|
|
"provider": "local",
|
|
"base_url": "http://127.0.0.1:11434/v1",
|
|
"api_key": "ollama-secret",
|
|
"model": "qwen2.5:7b",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
assert saved.status_code == 200
|
|
shown = client.get(url).json()
|
|
assert shown["api_key_set"] is True
|
|
assert shown["enabled"] is True and shown["provider"] == "local"
|
|
assert "ollama-secret" not in json.dumps(shown)
|
|
assert client.get("/api/session").json()["ai_enabled"] is True
|
|
assert Audit.objects.filter(action="ai_config.update").exists()
|
|
|
|
|
|
def test_ai_config_save_returns_saved_config_and_preserves_masked_key(workspace):
|
|
ds, _, client, _, _ = workspace
|
|
payload = {
|
|
"enabled": True,
|
|
"provider": "local",
|
|
"base_url": "http://127.0.0.1:11434/v1/",
|
|
"api_key": "saved-config-test-secret",
|
|
"model": "qwen2.5:7b",
|
|
}
|
|
|
|
for api_key, model in [(payload["api_key"], "qwen2.5:7b"), ("", "llama3.1:8b")]:
|
|
payload.update(api_key=api_key, model=model)
|
|
response = client.put("/api/ai-config", data=payload, content_type="application/json")
|
|
assert response.status_code == 200
|
|
saved = response.json()
|
|
shown = client.get("/api/ai-config").json()
|
|
assert saved == {"ok": True, **shown}
|
|
assert saved["enabled"] is True
|
|
assert saved["provider"] == "local"
|
|
assert saved["base_url"] == "http://127.0.0.1:11434/v1"
|
|
assert saved["model"] == model
|
|
assert saved["api_key_set"] is True
|
|
assert saved["updated_at"]
|
|
assert "api_key" not in saved
|
|
assert "saved-config-test-secret" not in response.content.decode()
|
|
assert AIConfig.objects.get(organization=ds.organization).api_key == "saved-config-test-secret"
|
|
|
|
|
|
def test_ai_models_lists_local_models(workspace):
|
|
ds, _, client, _, _ = workspace
|
|
client.put(
|
|
"/api/ai-config",
|
|
data={
|
|
"enabled": False,
|
|
"provider": "local",
|
|
"base_url": "http://127.0.0.1:11434/v1",
|
|
"model": "",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
with patch("finance.ai.httpx.Client") as http:
|
|
http.return_value.__enter__.return_value.get.return_value = type(
|
|
"Response",
|
|
(),
|
|
{
|
|
"raise_for_status": lambda self: None,
|
|
"json": lambda self: {"data": [{"id": "qwen2.5:7b"}, {"id": "llama3.1:8b"}]},
|
|
},
|
|
)()
|
|
listed = client.get("/api/ai-models")
|
|
assert listed.status_code == 200
|
|
assert listed.json()["items"] == ["qwen2.5:7b", "llama3.1:8b"]
|
|
call = http.return_value.__enter__.return_value.get.call_args
|
|
assert call.args[0] == "http://127.0.0.1:11434/v1/models"
|
|
|
|
with patch("finance.ai.httpx.Client") as http:
|
|
http.return_value.__enter__.return_value.get.side_effect = httpx.ConnectError("boom")
|
|
failed = client.get("/api/ai-models")
|
|
assert failed.status_code == 502
|
|
|
|
|
|
def test_assistant_uses_org_db_config(workspace):
|
|
ds, _, client, *_ = workspace
|
|
client.put(
|
|
"/api/ai-config",
|
|
data={
|
|
"enabled": True,
|
|
"provider": "local",
|
|
"base_url": "http://127.0.0.1:11434/v1",
|
|
"model": "qwen2.5:7b",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
with patch("finance.ai.httpx.Client") as http:
|
|
http.return_value.__enter__.return_value.post.return_value = completed_response()
|
|
response = client.post(
|
|
f"/api/datasets/{ds.pk}/assistant",
|
|
data={
|
|
"question": "分析儿科收入",
|
|
"context": {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
|
|
"use_model": True,
|
|
"consent": True,
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["provider"] == "local"
|
|
call = http.return_value.__enter__.return_value.post.call_args
|
|
assert call.args[0] == "http://127.0.0.1:11434/v1/responses"
|
|
assert call.kwargs["json"]["model"] == "qwen2.5:7b"
|
|
assert "Authorization" not in call.kwargs["headers"]
|
|
|
|
disabled = AIConfig.objects.get(organization=ds.organization)
|
|
disabled.enabled = False
|
|
disabled.save(update_fields=["enabled"])
|
|
denied = client.post(
|
|
f"/api/datasets/{ds.pk}/assistant",
|
|
data={"question": "分析儿科收入", "use_model": True, "consent": True},
|
|
content_type="application/json",
|
|
)
|
|
assert denied.status_code == 503
|
|
|
|
|
|
def test_model_intent_accepts_only_valid_dimensions_and_sends_no_rows(workspace, settings):
|
|
ds, _, _, _, _ = workspace
|
|
settings.AI_ENABLED = True
|
|
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.ai.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.AI_ENABLED = True
|
|
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.ai.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.ai.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, "shared": True}, content_type="application/json")
|
|
assert saved.status_code == 201
|
|
assert saved.json()["shared"] is True
|
|
definition_id = saved.json()["id"]
|
|
assert client.get(url).json()["items"][0]["definition"] == definition
|
|
|
|
updated = client.post(
|
|
url,
|
|
data={
|
|
"definition": {**definition, "name": " 收入成本差 ", "meaning": "更新说明"},
|
|
"shared": True,
|
|
"revision": saved.json()["revision"],
|
|
},
|
|
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
|
|
assert len(updated.json()["history"]) == 1
|
|
|
|
stale = client.post(
|
|
url,
|
|
data={"definition": {**definition, "meaning": "旧标签页覆盖"}, "revision": 1},
|
|
content_type="application/json",
|
|
)
|
|
assert stale.status_code == 409
|
|
assert FormulaDefinition.objects.get(pk=definition_id).definition["meaning"] == "更新说明"
|
|
|
|
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!")
|
|
shared = other.get(url).json()["items"]
|
|
assert len(shared) == 1 and shared[0]["owned"] is False and shared[0]["shared"] is True
|
|
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"]
|
|
|
|
|
|
def test_formula_definition_audit_failure_rolls_back_definition(workspace, monkeypatch):
|
|
ds, _, client, *_ = workspace
|
|
definition = {
|
|
"name": "审计回滚",
|
|
"meaning": "实际收入",
|
|
"scope": "所选分组与期间",
|
|
"missing": "结果不可用（#N/A）；零分母不可用",
|
|
"expression": {"metric": "revenue"},
|
|
}
|
|
|
|
def fail(*_args, **_kwargs):
|
|
raise RuntimeError("audit unavailable")
|
|
|
|
monkeypatch.setattr("finance.formula_views.audit", fail)
|
|
with pytest.raises(RuntimeError, match="audit unavailable"):
|
|
client.post(
|
|
f"/api/datasets/{ds.pk}/formula-definitions",
|
|
data={"definition": definition},
|
|
content_type="application/json",
|
|
)
|
|
assert not FormulaDefinition.objects.filter(dataset=ds, user=ds.created_by).exists()
|