From 3dd53eaeb1258ebd52c3c920827d568b3842ec5f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 16:03:10 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=86=E6=9E=90=E7=9B=B8=E5=85=B3=E7=94=A8?= =?UTF-8?q?=E4=BE=8B=E7=9A=84=E6=95=B0=E6=8D=AE=E6=BA=90=E5=A4=B9=E5=85=B7?= =?UTF-8?q?=E5=89=8D=E5=90=8E=E9=83=BD=E5=88=A0=E5=B9=B2=E5=87=80=EF=BC=9A?= =?UTF-8?q?tests/fixtures/sources=20=E5=8A=A0=20sqlite=5Fsource=EF=BC=88?= =?UTF-8?q?=E5=BB=BA=E4=B9=8B=E5=89=8D=E5=85=88=E6=8A=8A=E5=90=8C=E5=90=8D?= =?UTF-8?q?=E7=9A=84=E6=97=A7=E6=BA=90=E8=BF=9E=E5=90=8C=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E3=80=81=E5=89=96=E6=9E=90=E5=92=8C=E7=8E=B0?= =?UTF-8?q?=E6=9F=A5=E7=BC=93=E5=AD=98=E5=88=A0=E6=8E=89=EF=BC=8C=E4=BA=A4?= =?UTF-8?q?=E5=9B=9E=E6=96=B0=20id=EF=BC=8C=E5=8F=AF=E9=A1=BA=E6=89=8B?= =?UTF-8?q?=E6=8E=A2=E5=A5=BD=E8=A1=A8=E7=BB=93=E6=9E=84=EF=BC=89=E5=92=8C?= =?UTF-8?q?=20forget=5Fsource=EF=BC=88=E5=88=A0=E6=BA=90=E3=80=81=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=9A=8F=E5=A4=96=E9=94=AE=E7=BA=A7?= =?UTF-8?q?=E8=81=94=E5=88=A0=E9=99=A4=EF=BC=8C=E5=86=8D=E6=B8=85=E6=8E=89?= =?UTF-8?q?=20coverage=20=E7=8E=B0=E6=9F=A5=E7=BC=93=E5=AD=98=20=5FLIVE=20?= =?UTF-8?q?=E9=87=8C=E6=8C=89=E8=BF=99=E4=B8=AA=E6=BA=90=20id=20=E8=AE=B0?= =?UTF-8?q?=E7=9A=84=E8=A1=A8=E8=8C=83=E5=9B=B4=EF=BC=89=EF=BC=8Ctests/con?= =?UTF-8?q?ftest.py=20=E5=8A=A0=E5=85=B1=E7=94=A8=E5=A4=B9=E5=85=B7=20scra?= =?UTF-8?q?tch=5Fsource=EF=BC=88=E6=8C=89=E5=AE=83=E4=BB=AC=E5=BB=BA?= =?UTF-8?q?=E3=80=81=E6=B5=8B=E8=AF=95=E7=BB=93=E6=9D=9F=E6=97=B6=E5=88=A0?= =?UTF-8?q?=EF=BC=89=E3=80=82=E4=BB=A5=E5=89=8D=E5=90=84=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E7=9A=84=20shop=20=E5=A4=B9=E5=85=B7=E5=A4=8D=E7=94=A8?= =?UTF-8?q?=E5=90=8D=E5=8F=AB=20shop=20=E7=9A=84=E6=BA=90=E5=8D=B4?= =?UTF-8?q?=E4=B8=8D=E6=B8=85=E5=AE=83=E7=9A=84=E6=95=B0=E6=8D=AE=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=EF=BC=88=E5=A4=8D=E6=A0=B8=E6=97=B6=E4=B8=80=E6=9D=A1?= =?UTF-8?q?=E5=86=99=E4=BA=86=E7=9B=AE=E5=BD=95=E7=9A=84=E7=AB=AF=E5=88=B0?= =?UTF-8?q?=E7=AB=AF=E4=B8=B2=E6=89=B0=E4=BA=86=E5=90=8C=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E5=90=8E=E8=B7=91=E7=9A=84=E7=94=A8=E4=BE=8B=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E7=8E=B0=E6=9F=A5=E7=BC=93=E5=AD=98=E6=8C=89=E6=BA=90=20id=20?= =?UTF-8?q?=E8=AE=B0=E3=80=815=20=E5=88=86=E9=92=9F=E5=86=85=E5=90=8C?= =?UTF-8?q?=E4=B8=80=E4=B8=AA=E8=BF=9B=E7=A8=8B=E9=87=8C=E5=8F=AF=E8=83=BD?= =?UTF-8?q?=E6=8B=BF=E5=88=B0=E5=88=AB=E7=9A=84=E7=94=A8=E4=BE=8B=E7=9A=84?= =?UTF-8?q?=E6=97=A7=E8=8C=83=E5=9B=B4=E3=80=82test=5Fanalysis=5Fe2e?= =?UTF-8?q?=E3=80=81test=5Fanalysis=5Fe2e=5Fclaims=E3=80=81test=5Fanalysis?= =?UTF-8?q?=5Ftools=E3=80=81test=5Fmetric=5Fcomparability=E3=80=81test=5Fc?= =?UTF-8?q?ontract=5Fanalysis=5Fclaims=20=E7=9A=84=20shop=20=E5=A4=B9?= =?UTF-8?q?=E5=85=B7=E6=94=B9=E7=94=A8=E5=AE=83=EF=BC=88=E7=9B=B4=E6=8E=A5?= =?UTF-8?q?=E5=BB=BA=E9=81=AE=E7=BD=A9=E6=BA=90=E7=9A=84=E4=B8=A4=E6=9D=A1?= =?UTF-8?q?=E7=94=A8=E4=BE=8B=E4=B8=80=E5=B9=B6=E6=94=B9=EF=BC=8C=E5=8E=BB?= =?UTF-8?q?=E6=8E=89=20try/finally=EF=BC=89=EF=BC=9Btest=5Fanalysis=5Ftool?= =?UTF-8?q?s=20=E5=BC=80=E5=A4=B4=E3=80=8C=E7=AE=97=E5=AD=90=E8=BF=98?= =?UTF-8?q?=E6=B2=A1=E8=90=BD=E5=9C=B0=E3=80=8D=E7=9A=84=E8=AF=B4=E6=98=8E?= =?UTF-8?q?=E6=94=B9=E6=88=90=E7=85=A7=E5=AE=9E=E5=86=99=E6=9B=BF=E8=BA=AB?= =?UTF-8?q?=E5=92=8C=E7=9C=9F=E7=AE=97=E5=AD=90=E7=9A=84=E7=AB=AF=E5=88=B0?= =?UTF-8?q?=E7=AB=AF=E5=90=84=E5=9C=A8=E5=93=AA=E3=80=82=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=EF=BC=9A=E6=97=A7=E6=BA=90=E7=95=99=E4=B8=8B=E7=9A=84=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E7=9B=AE=E5=BD=95=E5=92=8C=E7=8E=B0=E6=9F=A5=E8=8C=83?= =?UTF-8?q?=E5=9B=B4=E5=9C=A8=E5=BB=BA=E6=96=B0=E6=BA=90=E6=97=B6=E6=B8=85?= =?UTF-8?q?=E6=8E=89=EF=BC=8C=E5=88=A0=E6=BA=90=E5=90=8E=E7=8E=B0=E6=9F=A5?= =?UTF-8?q?=E7=BC=93=E5=AD=98=E9=87=8C=E6=B2=A1=E6=9C=89=E5=AE=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/conftest.py | 18 +++++ backend/tests/fixtures/sources.py | 42 ++++++++++++ backend/tests/test_analysis_e2e.py | 64 +++++++++++++---- backend/tests/test_analysis_e2e_claims.py | 30 +++----- backend/tests/test_analysis_tools.py | 68 +++++++------------ .../tests/test_contract_analysis_claims.py | 19 ++---- backend/tests/test_metric_comparability.py | 17 +---- 7 files changed, 151 insertions(+), 107 deletions(-) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index d4be71bf..88c1d8a8 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -111,6 +111,24 @@ def pytest_collection_modifyitems(config, items) -> None: item.add_marker(skip) +@pytest.fixture +async def scratch_source(): + """测试里建的 SQLite 数据源：source_id = await scratch_source("shop", 库文件路径, options={...})。同名的旧源先删掉（连同 + 它的数据目录、剖析和现查缓存，tests/fixtures/sources.sqlite_source），测试结束时再删一遍、清掉现查缓存（forget_source）。 + 分析相关的用例都建名叫 shop 的源：别的用例留下的数据目录、5 分钟内现查过的表范围会改掉这次的判定。""" + from tests.fixtures.sources import forget_source, sqlite_source + + made: list[str] = [] + + async def make(name: str, path, **kw) -> str: + made.append(await sqlite_source(name, path, **kw)) + return made[-1] + + yield make + for source_id in made: + await forget_source(source_id) + + @pytest.fixture(scope="session") def scenic_db(tmp_path_factory) -> str: """合成的景区业务库（tests/fixtures/catalog/scenic.py）：50 张表、真外键与只能靠命名推断的关系并存。 diff --git a/backend/tests/fixtures/sources.py b/backend/tests/fixtures/sources.py index 81d9b6da..54eb501d 100644 --- a/backend/tests/fixtures/sources.py +++ b/backend/tests/fixtures/sources.py @@ -3,9 +3,15 @@ 数据源表在每个测试进程的测试库里是共用的：留下来的源会出现在助手的数据源清单里（50 张表的合成库还会 触发挑表、多一次模型调用），别的用例的结果就跟着执行顺序变。阶段 1B 就是这样被绊过一次。 删数据源时它的数据目录随外键级联删除（测试库开着 PRAGMA foreign_keys=ON）。 + +分析相关的用例都建名叫 shop 的源（sqlite_source / forget_source，tests/conftest.py 的 scratch_source 夹具按它们建、按它们删）： +别的用例留下的数据目录（金额标成了流量，窗口长度不同就从有条件可比变成不可比）、coverage 的现查缓存里 5 分钟内按同一个 +源 id 记下的表范围（另一份库的日期范围），都会改掉这次的判定。所以建之前先把同名的旧源连同这两样删干净，用完再删一遍。 """ from __future__ import annotations +from typing import Any + from sqlalchemy import delete, select from app.data.engine import engines @@ -26,6 +32,42 @@ async def drop_source(source_id: str | None = None, *, name: str | None = None) await engines.invalidate(source_id) +async def sqlite_source(name: str, path: Any, *, options: dict[str, Any] | None = None, schema: bool = False) -> str: + """建一个指向这份 SQLite 库文件的只读数据源（启用），交回 id。同名的旧源先删掉（forget_source：连同它的数据目录、剖析 + 和现查缓存），这次的源是新的 id。schema 为真时顺手探好表结构（要写数据目录的用例用）。用完调 forget_source。""" + await forget_source(name=name) + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, + options=dict(options or {})) + session.add(row) + await session.flush() + if schema: + from app.data.introspect import introspect + + row.schema_cache = await introspect(row) + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + return source_id + + +async def forget_source(source_id: str | None = None, *, name: str | None = None) -> None: + """按 id 或名字删一个数据源（drop_source：数据目录、剖析随外键级联删除，连接池作废），再清掉 coverage 现查缓存里它的 + 表范围（data/coverage 的 _LIVE，按数据源 id 记、5 分钟有效：同一个进程里后面的用例会拿到这份旧范围）。不存在就什么都 + 不做。""" + from app.data import coverage + + if source_id is None: + async with SessionLocal() as session: + source_id = (await session.execute( + select(DataSource.id).where(DataSource.name == name))).scalar_one_or_none() + if source_id is None: + return + await drop_source(source_id) + for key in [k for k in coverage._LIVE if k and k[0] == source_id]: + coverage._LIVE.pop(key, None) + + async def introspect_now(client, source_id: str, *, expect: int = 202) -> dict: """探查结构并等它做完（探查是后台任务：POST /introspect 回 202 和任务，next-round 03 A1），返回数据源此刻的样子 （DataSourceOut）。expect 不是 202 时只发请求、断言状态码，返回响应体。""" diff --git a/backend/tests/test_analysis_e2e.py b/backend/tests/test_analysis_e2e.py index b66d8dcd..5dd0afbd 100644 --- a/backend/tests/test_analysis_e2e.py +++ b/backend/tests/test_analysis_e2e.py @@ -19,9 +19,8 @@ import app.analysis as analysis_ops from app.analysis import contract from app.core import artifact_store from app.core.artifact_store import canonical_json, content_hash -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine.runner import run_manager TOOL = "db_query__shop" @@ -64,24 +63,15 @@ async def engine_up(): @pytest.fixture(autouse=True) -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): + """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，用完再删）。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);") db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders()) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - yield - await engines.invalidate(source_id) + return await scratch_source("shop", path) def node(nid, ntype, **config): @@ -349,3 +339,49 @@ async def test_a_ratio_between_two_regions_is_not_a_period_comparison(south_week assert issuance["tier"] == "degraded" assert issuance["gaps"] == ["指标「华东对华南的倍数」两期不可比：两边除了日期，筛选的还不一样"] recomputes(artifact_store.load(found["artifact"])) + + +# -------------------------------------------------------------------------- +# 测试夹具：复用名叫 shop 的源，前后都删干净 +# -------------------------------------------------------------------------- + + +async def test_a_scratch_source_starts_clean_and_forgets_its_probes(tmp_path): + """分析相关的用例都建名叫 shop 的源（tests/conftest.py 的 scratch_source）：别的用例留下的数据目录、coverage 的现查缓存里 + 按同一个源 id 记下的表范围（5 分钟有效）会改掉这次的判定。sqlite_source 建的源是新的 id，同名的旧源连同数据目录、现查 + 缓存先删掉；forget_source 删源（数据目录随外键级联删除）、清掉现查缓存里它的表范围。""" + from app.analysis.probe import make_prober + from app.data import catalog as data_catalog + from app.data import coverage + from app.data.catalog import make_item + from app.db.models import CatalogNote, DataSource + from tests.fixtures.sources import forget_source, sqlite_source + + path = tmp_path / "other.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (day TEXT, amount INTEGER); INSERT INTO orders VALUES ('2026-01-05', 1);") + db.close() + + async def leave_traces(source_id): + async with SessionLocal() as session: + await data_catalog.write_entry(session, source_id, "orders", {"columns": {"amount": { + "measure": make_item("flow", "human")}}}, if_version=0, actor=None) + row = await session.get(DataSource, source_id) + probe = await make_prober({"scratch": row})("scratch", "orders", "day") + assert probe["max"] == "2026-01-05" and probe["basis"] == "live" + assert any(k[0] == source_id for k in coverage._LIVE) + + async def traces(source_id): + async with SessionLocal() as session: + notes = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id))).scalars().all() + row = await session.get(DataSource, source_id) + return {"row": row is not None, "catalog": len(notes), "probes": sum(k[0] == source_id for k in coverage._LIVE)} + + old = await sqlite_source("scratch", path) + await leave_traces(old) + new = await sqlite_source("scratch", path) + assert new != old and await traces(old) == {"row": False, "catalog": 0, "probes": 0} + assert await traces(new) == {"row": True, "catalog": 0, "probes": 0} + await leave_traces(new) + await forget_source(new) + assert await traces(new) == {"row": False, "catalog": 0, "probes": 0} diff --git a/backend/tests/test_analysis_e2e_claims.py b/backend/tests/test_analysis_e2e_claims.py index c07b74f3..9c3a98eb 100644 --- a/backend/tests/test_analysis_e2e_claims.py +++ b/backend/tests/test_analysis_e2e_claims.py @@ -26,15 +26,12 @@ from app.analysis import contract from app.core import artifact_store from app.data import catalog as data_catalog from app.data.catalog import make_item -from app.data.engine import engines -from app.data.introspect import introspect from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine import analysis_claims as ac from app.engine.evidence import iter_segments from app.engine.runner import run_manager from app.main import app -from tests.fixtures.sources import drop_source TOOL = "db_query__shop" DECOMPOSE = contract.TOOL_NAMES["decompose"] @@ -74,7 +71,7 @@ async def engine_up(): @pytest.fixture(autouse=True) -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, region TEXT, channel TEXT, amount INTEGER);") @@ -84,25 +81,14 @@ async def shop(tmp_path): db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [(i, w, r, c, a) for i, (r, w, c, a) in enumerate(rows, 1)]) db.commit() db.close() + # scratch_source 建的是新源（同名的旧源连同数据目录、现查缓存先删掉），测试结束时删掉、数据目录随外键级联删除：同一个 + # 测试进程里别的用例也用名叫 shop 的源，留着这份目录会改掉它们的判定（比如金额标成了流量，窗口长度不同就从有条件可比 + # 变成不可比） + source_id = await scratch_source("shop", path, schema=True) async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - row.schema_cache = await introspect(row) - await session.commit() for table, notes in NOTES.items(): - current = await data_catalog.read_entry(session, row.id, table) - await data_catalog.write_entry(session, row.id, table, notes, if_version=current.version if current else 0, - actor=None) - source_id = row.id - await engines.invalidate(source_id) - yield - # 删掉这个源（数据目录随外键级联删除）：同一个测试进程里别的用例也用名叫 shop 的源，留着这份目录会改掉它们的判定 - # （比如金额标成了流量，窗口长度不同就从有条件可比变成不可比） - await drop_source(source_id) + await data_catalog.write_entry(session, source_id, table, notes, if_version=0, actor=None) + return source_id def _node(nid, ntype, **config): diff --git a/backend/tests/test_analysis_tools.py b/backend/tests/test_analysis_tools.py index 9c4bf7e4..96513247 100644 --- a/backend/tests/test_analysis_tools.py +++ b/backend/tests/test_analysis_tools.py @@ -1,8 +1,9 @@ """agent 分析工具（docs/design/analysis-round/00-总方案.md §3.2）：参数、方法说明、查询容器、台账、编号、下钻、次数上限、 遮罩列、续跑。 -算子（app.analysis.run）还没落地：换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps），断言入口交给算子的 -东西、算子交回后入口做的事。模型换成脚本（MockChatModel._decide），不调用任何模型。库是虚构的零售库 shop。 +算子（app.analysis.run）换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps），断言入口交给算子的东西、算子 +交回后入口做的事；五个工具接真算子的端到端见 tests/test_analysis_e2e.py。模型换成脚本（MockChatModel._decide），不调用任何 +模型。库是虚构的零售库 shop。 """ from __future__ import annotations @@ -18,9 +19,8 @@ from app.analysis import contract from app.analysis.contract import CallError from app.core import artifact_store from app.core.artifact_store import canonical_json, content_hash -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import Artifact, DataSource, Run, RunEvent +from app.db.models import Artifact, Run, RunEvent from app.engine.runner import run_manager from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, METHOD_GUIDE, NO_SCOPE @@ -45,30 +45,20 @@ async def engine_up(): await run_manager.shutdown() -async def _shop(tmp_path, options=None): +async def _shop(tmp_path, scratch_source, options=None): + """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL, phone TEXT);") db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ROWS) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - return source_id + return await scratch_source("shop", path, options=options) @pytest.fixture -async def shop(tmp_path): - source_id = await _shop(tmp_path) - yield - await engines.invalidate(source_id) +async def shop(tmp_path, scratch_source): + return await _shop(tmp_path, scratch_source) class Script: @@ -407,31 +397,25 @@ async def test_the_number_of_analyses_per_node_is_capped(monkeypatch, shop): assert ops.calls[0]["settings"]["max_calls"] == 1 -async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path): - source_id = await _shop(tmp_path, {"mask_columns": ["phone"]}) - try: - ops = FakeOps(monkeypatch) - Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})]) - await run_graph(agent_graph()) - masked = ops.calls[0]["snapshots"]["Q1"]["masked"] - # 按 SQL 认出改了名字的遮罩列（phone AS tel）排在前面，数据源设的列名也在 - assert masked[0] == "tel" and "phone" in masked - finally: - await engines.invalidate(source_id) +async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path, scratch_source): + await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]}) + ops = FakeOps(monkeypatch) + Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})]) + await run_graph(agent_graph()) + masked = ops.calls[0]["snapshots"]["Q1"]["masked"] + # 按 SQL 认出改了名字的遮罩列（phone AS tel）排在前面，数据源设的列名也在 + assert masked[0] == "tel" and "phone" in masked -async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path): - source_id = await _shop(tmp_path, {"mask_columns": ["phone"]}) - try: - ops = FakeOps(monkeypatch) - Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})]) - await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True}) - snap = ops.calls[0]["snapshots"]["Q1"] - assert "tel" in snap["masked"] - tel = snap["columns"].index("tel") - assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照" - finally: - await engines.invalidate(source_id) +async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path, scratch_source): + await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]}) + ops = FakeOps(monkeypatch) + Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})]) + await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True}) + snap = ops.calls[0]["snapshots"]["Q1"] + assert "tel" in snap["masked"] + tel = snap["columns"].index("tel") + assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照" # -------------------------------------------------------------------------- diff --git a/backend/tests/test_contract_analysis_claims.py b/backend/tests/test_contract_analysis_claims.py index d61f6b47..1b5859df 100644 --- a/backend/tests/test_contract_analysis_claims.py +++ b/backend/tests/test_contract_analysis_claims.py @@ -17,9 +17,8 @@ from httpx import ASGITransport, AsyncClient from langchain_core.messages import AIMessage from sqlalchemy import select -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent, Workflow, WorkflowVersion +from app.db.models import Run, RunEvent, Workflow, WorkflowVersion from app.engine import analysis_claims as ac from app.engine import compiler from app.engine.nodes.io import _analysis_claims, _comparability @@ -271,25 +270,15 @@ async def test_a_governed_formal_run_counts_the_untraced_cited_metric(client, mo @pytest.fixture -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): + """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, amount REAL);") db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "2026-W37", 100.5), (2, "2026-W37", 200.0)]) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - else: - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - yield - await engines.invalidate(source_id) + return await scratch_source("shop", path) async def test_a_metric_read_from_a_query_cell_is_traced(client, monkeypatch, shop): diff --git a/backend/tests/test_metric_comparability.py b/backend/tests/test_metric_comparability.py index 87d02bec..477617f4 100644 --- a/backend/tests/test_metric_comparability.py +++ b/backend/tests/test_metric_comparability.py @@ -19,9 +19,8 @@ from sqlalchemy import select from app.analysis.contract import CallError from app.core import artifact_store -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine.nodes.metrics import COMPARABILITY_SKIPPED from app.engine.runner import run_manager from tests.fixtures.analysis_ops import FakeOps @@ -41,7 +40,7 @@ async def engine_up(): @pytest.fixture(autouse=True) -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, store INTEGER, amount REAL, " @@ -51,17 +50,7 @@ async def shop(tmp_path): (3, "2026-10-13", "华东", 1203, 450.0, 0.0), (4, "2026-10-15", "华北", 1204, 300.0, 15.0)]) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - yield - await engines.invalidate(source_id) + return await scratch_source("shop", path) def node(nid, ntype, **config):