ad7fba6021
- core：配置（prod 拒绝默认密钥 / dev 短信 / 假模型）、租户事务 + 行级安全策略、 problem+json 错误码、信封加密与签名下载、对象存储、事务内入队、Valkey stream 事件流、权限点 - identity：手机号 HMAC 存储、注册、登录（挤下线需确认，API3）、服务端会话 - billing：额度流水只追加，预扣 / 结算 / 退回，一笔预扣只有一个结局 - document：分片上传、按文件头识别格式、Gotenberg 转换、文字层与稳定句子编号 - checklist：标注规则识别、模型抽取、三道校验、标注类条款规则兜底、重跑对齐 - ai：提示词注册表（extract.substantive v1 来自 S4 v2）、假模型、OpenAI 兼容供应商、ai_call 流水 - 迁移：data-model 基线、套餐、Procrastinate、会话状态函数 - 测试 30 个：租户隔离、额度、失败退回、对象加密、契约一致性 - Compose（含 migrate）、镜像、CI Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
"""用 A 企业的令牌访问 B 企业的资源，一律 404（架构 §3.3 第 3 层）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import httpx
|
||
|
||
from tests.conftest import project_with_tender, register, run_parse
|
||
|
||
|
||
async def test_other_tenant_resources_are_not_found(app_client: httpx.AsyncClient) -> None:
|
||
a = await register(app_client, "甲公司")
|
||
b = await register(app_client, "乙公司")
|
||
project_id, doc_id = await project_with_tender(a)
|
||
job = await run_parse(a, doc_id)
|
||
assert job["status"] == "succeeded"
|
||
|
||
for path in (
|
||
f"/projects/{project_id}",
|
||
f"/projects/{project_id}/checklist",
|
||
f"/projects/{project_id}/checklist/items",
|
||
f"/documents/{doc_id}",
|
||
f"/documents/{doc_id}/pages",
|
||
f"/documents/{doc_id}/file?variant=normalized",
|
||
f"/parse-jobs/{job['id']}",
|
||
):
|
||
r = await b.get(path)
|
||
assert r.status_code == 404, (path, r.status_code, r.text)
|
||
r = await b.post(f"/documents/{doc_id}/parse-jobs", json={"expected_charge": 1})
|
||
assert r.status_code == 404
|
||
r = await b.post(f"/parse-jobs/{job['id']}/cancel")
|
||
assert r.status_code == 404
|
||
|
||
assert (await b.get("/projects")).json()["items"] == []
|
||
assert [p["id"] for p in (await a.get("/projects")).json()["items"]] == [project_id]
|
||
|
||
|
||
async def test_database_blocks_queries_without_tenant() -> None:
|
||
"""代码漏了租户条件时，行级安全策略兜底：没设租户什么都查不到。"""
|
||
from sqlalchemy import text
|
||
|
||
from bid_doctor.core.db import anonymous_tx
|
||
|
||
async with anonymous_tx() as s:
|
||
for table in ("project", "source_document", "checklist_item", "quota_entry", "citation"):
|
||
assert (await s.execute(text(f"SELECT count(*) FROM {table}"))).scalar_one() == 0
|