fix(T+): 补齐指导成本与读取服务保护
补充业务详情指导成本入口、来源版本和历史封存保护,接入月份锁并完善权限复核。 同步接入读取服务运行状态到页面,补充前端控制、回归测试和服务端只读连接授权校验。 定向隔离测试:89 项通过,1 条既有 Starlette 弃用警告。
This commit is contained in:
1 parent
68e70fd6e2
commit
b8920c0d28
17 files changed
+375
-47
No files matched your search
@@ -21,7 +21,10 @@ def script(monkeypatch):
|
||||
spec = importlib.util.spec_from_file_location("compatibility_script", path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
settings = Settings(_env_file=None, app_secret_key="x" * 32, t_plus_encrypt=True)
|
||||
settings = Settings(
|
||||
_env_file=None, app_secret_key="x" * 32, t_plus_encrypt=True,
|
||||
t_plus_allow_dev_sa=False, t_plus_allow_admin_account=False,
|
||||
)
|
||||
monkeypatch.setattr(module, "get_settings", lambda: settings)
|
||||
return module
|
||||
|
||||
@@ -44,7 +47,7 @@ def test_脚本复用连接文件校验与加密设置(script, tmp_path):
|
||||
assert script.get_settings().t_plus_enabled is False
|
||||
payload["user"] = "sa"
|
||||
path.write_text(json.dumps(payload), encoding="utf-8")
|
||||
with pytest.raises(RuntimeError, match="专用只读"):
|
||||
with pytest.raises(RuntimeError, match="管理员账号未获授权"):
|
||||
script.compatibility_settings(path)
|
||||
|
||||
|
||||
|
||||
@@ -72,11 +72,37 @@ def test_连接文件不接受sa或应用配置覆盖(tmp_path):
|
||||
resolve_tplus_settings(settings)
|
||||
|
||||
|
||||
def test_生产连接文件允许显式授权的管理员账号(tmp_path):
|
||||
path = tmp_path / "reader.json"
|
||||
path.write_text(
|
||||
json.dumps({
|
||||
"host": "127.0.0.1", "port": 1433, "database": "test_source",
|
||||
"user": "sa", "password": "test-only-value", "driver": "test_driver",
|
||||
}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
settings = Settings(
|
||||
_env_file=None,
|
||||
app_secret_key="x" * 32,
|
||||
app_env="prod",
|
||||
t_plus_enabled=True,
|
||||
t_plus_allow_admin_account=True,
|
||||
t_plus_connection_file=str(path),
|
||||
)
|
||||
assert resolve_tplus_settings(settings).t_plus_user == "sa"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("environment", "allowed", "accepted"),
|
||||
[("dev", False, False), ("prod", True, False), ("dev", True, True)],
|
||||
("environment", "allowed", "admin_allowed", "accepted"),
|
||||
[
|
||||
("dev", False, False, False),
|
||||
("prod", True, False, False),
|
||||
("dev", True, False, True),
|
||||
("prod", False, True, True),
|
||||
("prod", False, False, False),
|
||||
],
|
||||
)
|
||||
def test_sa仅允许显式启用的开发环境(environment, allowed, accepted):
|
||||
def test_sa仅允许显式授权(environment, allowed, admin_allowed, accepted):
|
||||
settings = Settings(
|
||||
_env_file=None,
|
||||
app_secret_key="x" * 32,
|
||||
@@ -86,9 +112,10 @@ def test_sa仅允许显式启用的开发环境(environment, allowed, accepted):
|
||||
t_plus_database="test_source",
|
||||
t_plus_odbc_driver="test_driver",
|
||||
t_plus_allow_dev_sa=allowed,
|
||||
t_plus_allow_admin_account=admin_allowed,
|
||||
)
|
||||
if accepted:
|
||||
assert resolve_tplus_settings(settings).t_plus_user == "sa"
|
||||
else:
|
||||
with pytest.raises(RuntimeError, match="不能使用 sa"):
|
||||
with pytest.raises(RuntimeError, match="管理员账号未获授权"):
|
||||
resolve_tplus_settings(settings)
|
||||
@@ -433,6 +433,21 @@ def test_资料版本冲突不覆盖_历史保留(work_session):
|
||||
assert s.scalar(select(func.count()).select_from(TPlusFact)) == 2
|
||||
|
||||
|
||||
def test_旧工作台不能绕过指导成本封存检查(work_session):
|
||||
from app.core.exceptions import BadRequestError
|
||||
|
||||
data = {"product_id": "9", "month": "2026-03", "basis": "UNIT", "amount": "25", "unit": "件"}
|
||||
with pytest.raises(BadRequestError, match="业务详情"):
|
||||
fact(work_session, "GUIDANCE", "legacy", {**data, "is_finalized": True})
|
||||
work_session.add(TPlusFact(
|
||||
source_database="test_source", kind="GUIDANCE", fact_key="sealed", version=1,
|
||||
data_json={**data, "is_finalized": True}, evidence="已核对", requested_by=1,
|
||||
))
|
||||
work_session.commit()
|
||||
with pytest.raises(BadRequestError, match="已有封存"):
|
||||
fact(work_session, "GUIDANCE", "another-key", data)
|
||||
|
||||
|
||||
def test_更正事实要求排除或原业务归属():
|
||||
with pytest.raises(ValidationError):
|
||||
FactIn(
|
||||
|
||||
@@ -57,6 +57,13 @@ def test_忙锁时不领取任务也不恢复其他运行记录(tplus_session, m
|
||||
assert batch.status == "QUEUED"
|
||||
|
||||
|
||||
def test_停止请求阻止领取排队任务(tplus_session, monkeypatch):
|
||||
source = _worker_setup(monkeypatch)
|
||||
batch = _queued(tplus_session)
|
||||
assert worker.run_once(tplus_session, source_factory=source, may_claim=lambda _session: False) is None
|
||||
assert batch.status == "QUEUED"
|
||||
|
||||
|
||||
def test_源连接失败写入固定错误不泄漏凭据(tplus_session, monkeypatch):
|
||||
_worker_setup(monkeypatch)
|
||||
batch = _queued(tplus_session)
|
||||
|
||||
Reference in new issue
Block a user