Jayway Liu
0aaa87e31a
- FormulaDefinition 增加 revision 字段（迁移 0005） - 保存接口在单事务内锁定用户所属行序列化并发：配额（50/数据集）检查与写入原子化， 不再可能产生第 51 条；唯一约束竞争返回 409 - 同名更新必须携带匹配的 revision，两个窗口不再相互静默覆盖； GET 列表返回当前 revision；保存增加 30/小时限流 - 审计与定义变更（保存/删除）同一事务提交，审计失败一并回滚 - 前端保存同名定义时携带已知 revision，并明示“同名覆盖”行为 - 测试：revision 边界（缺失/过期/正确/再次过期）、配额拒绝与释放、 审计失败回滚；PostgreSQL 并发配额用例独立门禁（本机 SQLite 跳过）
124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.db import connection
|
|
|
|
from finance.models import FormulaDefinition
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def column_definition(name="收入成本差"):
|
|
return {
|
|
"name": name,
|
|
"meaning": "实际收入减成本",
|
|
"scope": "所选分组与期间",
|
|
"missing": "结果不可用（#N/A）；零分母不可用",
|
|
"expression": {
|
|
"operator": "subtract",
|
|
"left": {"metric": "revenue"},
|
|
"right": {"metric": "cost"},
|
|
},
|
|
}
|
|
|
|
|
|
def _post(client, url, definition, revision=None):
|
|
return client.post(
|
|
url,
|
|
data={"definition": definition, **({"revision": revision} if revision is not None else {})},
|
|
content_type="application/json",
|
|
)
|
|
|
|
|
|
def test_same_name_update_requires_matching_revision(workspace):
|
|
ds, _, client, *_ = workspace
|
|
url = f"/api/datasets/{ds.pk}/formula-definitions"
|
|
|
|
saved = _post(client, url, column_definition())
|
|
assert saved.status_code == 201
|
|
item_id, revision = saved.json()["id"], saved.json()["revision"]
|
|
|
|
# 缺失或过期的 revision 都不能静默覆盖其他窗口的修改
|
|
assert _post(client, url, column_definition()).status_code == 409
|
|
assert _post(client, url, column_definition(), revision=999).status_code == 409
|
|
|
|
updated = _post(client, url, column_definition(), revision=revision)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["revision"] == revision + 1
|
|
|
|
# 基于旧版本的再次更新必须被拒绝，两个标签页不互相静默丢失
|
|
stale = _post(client, url, column_definition(), revision=revision)
|
|
assert stale.status_code == 409
|
|
current = FormulaDefinition.objects.get(pk=item_id)
|
|
assert current.revision == revision + 1
|
|
# 列表接口暴露当前 revision 供前端携带
|
|
listed = client.get(url).json()["items"]
|
|
assert listed[0]["revision"] == revision + 1
|
|
|
|
|
|
def test_definition_quota_is_enforced_and_release_allows_new_save(workspace):
|
|
ds, _, client, *_ = workspace
|
|
url = f"/api/datasets/{ds.pk}/formula-definitions"
|
|
for index in range(50):
|
|
FormulaDefinition.objects.create(
|
|
dataset=ds,
|
|
user=ds.created_by,
|
|
normalized_name=f"定义{index:02d}",
|
|
definition=column_definition(f"定义{index:02d}"),
|
|
)
|
|
|
|
rejected = _post(client, url, column_definition("第 51 个不同名定义"))
|
|
assert rejected.status_code == 400
|
|
assert "50" in rejected.json()["error"]
|
|
assert FormulaDefinition.objects.filter(dataset=ds).count() == 50
|
|
|
|
# 同名更新不受配额限制；删除后名额释放可再保存新名称
|
|
assert _post(client, url, column_definition("定义00"), revision=1).status_code == 200
|
|
FormulaDefinition.objects.filter(normalized_name="定义00").delete()
|
|
assert _post(client, url, column_definition("第 51 个不同名定义")).status_code == 201
|
|
|
|
|
|
def test_audit_failure_rolls_back_definition_change(workspace):
|
|
ds, _, client, *_ = workspace
|
|
url = f"/api/datasets/{ds.pk}/formula-definitions"
|
|
|
|
with patch("finance.formula_views.audit", side_effect=RuntimeError("audit unavailable")):
|
|
with pytest.raises(RuntimeError, match="audit unavailable"):
|
|
_post(client, url, column_definition())
|
|
|
|
assert FormulaDefinition.objects.filter(dataset=ds).count() == 0
|
|
|
|
|
|
def test_concurrent_distinct_saves_cannot_exceed_quota_on_postgres(workspace):
|
|
if connection.vendor != "postgresql":
|
|
pytest.skip("并发配额依赖 PostgreSQL 行锁语义，生产库单独验证")
|
|
import threading
|
|
|
|
from django.test import Client
|
|
|
|
ds, member, *_ = workspace
|
|
user = ds.created_by
|
|
for index in range(49):
|
|
FormulaDefinition.objects.create(
|
|
dataset=ds,
|
|
user=user,
|
|
normalized_name=f"定义{index:02d}",
|
|
definition=column_definition(f"定义{index:02d}"),
|
|
)
|
|
|
|
barrier = threading.Barrier(4)
|
|
|
|
def save(index):
|
|
client = Client()
|
|
assert client.login(username=user.username, password="Finance-TestPassword991!")
|
|
barrier.wait()
|
|
_post(client, f"/api/datasets/{ds.pk}/formula-definitions", column_definition(f"并发{index}"))
|
|
|
|
threads = [threading.Thread(target=save, args=(index,)) for index in range(4)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
assert FormulaDefinition.objects.filter(dataset=ds).count() == 50
|