测试里建的数据源用完就删：留在共用测试库里的源会混进助手的数据源清单、触发挑表，别的用例的结果随执行顺序变

backend/tests/fixtures/sources.py

@@ -0,0 +1,26 @@
+"""测试里建的数据源，用完就删。
+
+数据源表在每个测试进程的测试库里是共用的：留下来的源会出现在助手的数据源清单里（50 张表的合成库还会
+触发挑表、多一次模型调用），别的用例的结果就跟着执行顺序变。阶段 1B 就是这样被绊过一次。
+删数据源时它的数据目录随外键级联删除（测试库开着 PRAGMA foreign_keys=ON）。
+"""
+from __future__ import annotations
+
+from sqlalchemy import delete, select
+
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource
+
+
+async def drop_source(source_id: str | None = None, *, name: str | None = None) -> None:
+    """按 id 或名字删一个数据源，并作废它的连接池。不存在就什么都不做。"""
+    async with SessionLocal() as session:
+        if source_id is None:
+            source_id = (await session.execute(
+                select(DataSource.id).where(DataSource.name == name))).scalar_one_or_none()
+            if source_id is None:
+                return
+        await session.execute(delete(DataSource).where(DataSource.id == source_id))
+        await session.commit()
+    await engines.invalidate(source_id)

backend/tests/test_catalog_api.py

@@ -11,12 +11,12 @@ import pytest
 from httpx import ASGITransport, AsyncClient
 
 from app.data import catalog
-from app.data.engine import engines
 from app.db.base import SessionLocal
 from app.db.models import DataSource
 from app.main import app as _fastapi_app
 from tests.fixtures.catalog import scenic
 from tests.fixtures.catalog.fake_model import FakeModel
+from tests.fixtures.sources import drop_source
 
 ACTOR = {"X-Actor": quote("王敏")}
 
@@ -38,7 +38,7 @@ async def source_id(client, scenic_db):
     r = await client.post(f"/api/datasources/{sid}/introspect")
     assert r.status_code == 200, r.text
     yield sid
-    await engines.invalidate(sid)
+    await drop_source(sid)
 
 
 def _base(sid: str) -> str:
@@ -184,9 +184,12 @@ async def test_draft_without_schema_is_refused(client):
         session.add(row)
         await session.commit()
         sid = row.id
-    r = await client.post(f"{_base(sid)}/draft", json={})
-    assert r.status_code == 409 and "探查结构" in r.json()["detail"]
-    assert (await client.get(_base(sid))).json()["tables"] == []
+    try:
+        r = await client.post(f"{_base(sid)}/draft", json={})
+        assert r.status_code == 409 and "探查结构" in r.json()["detail"]
+        assert (await client.get(_base(sid))).json()["tables"] == []
+    finally:
+        await drop_source(sid)
 
 
 async def test_schema_endpoint_detail_carries_catalog(client, source_id):

backend/tests/test_catalog_draft.py

@@ -15,6 +15,7 @@ from app.data.catalog import make_item
 from app.db.base import SessionLocal
 from app.db.models import DataSource
 from tests.fixtures.catalog.fake_model import FakeModel
+from tests.fixtures.sources import drop_source
 
 
 def _col(name: str, type_: str = "INTEGER", comment: str | None = None, nullable: bool = True) -> dict:
@@ -255,6 +256,16 @@ async def test_model_draft_respects_system_notes_of_imported_tables():
 # ---------------------------------------------------------------- 起草并写入
 
 
+_CREATED: list[str] = []
+
+
+@pytest.fixture(autouse=True)
+async def _drop_created_sources():
+    yield
+    while _CREATED:
+        await drop_source(_CREATED.pop())
+
+
 async def _db_source(tables: dict, **kw) -> DataSource:
     for name, meta in tables.items():
         meta["qualified"] = name
@@ -263,6 +274,7 @@ async def _db_source(tables: dict, **kw) -> DataSource:
                          schema_cache={"tables": tables}, **kw)
         session.add(row)
         await session.commit()
+        _CREATED.append(row.id)
         return row
 
 

backend/tests/test_catalog_render.py

@@ -7,6 +7,7 @@ from app.data import catalog
 from app.data.catalog import make_item
 from app.db.base import SessionLocal
 from app.db.models import Artifact, DataSource, Run
+from tests.fixtures.sources import drop_source
 
 MARK = "（推断，未确认）"
 
@@ -184,4 +185,5 @@ async def test_table_usage_counts_run_queries_per_table(monkeypatch):
         await session.commit()
         source = await session.get(DataSource, sid)
         usage = await catalog.table_usage(session, source)
+    await drop_source(sid)
     assert usage == {"visits": 4, "明细": 2, "parks": 1}

backend/tests/test_catalog_runtime.py

@@ -14,12 +14,12 @@ from types import SimpleNamespace
 from app.core.artifact_store import canonical_json, content_hash, load
 from app.data import catalog
 from app.data.catalog import make_item
-from app.data.engine import engines
 from app.data.introspect import describe_table, introspect
 from app.db.base import SessionLocal
 from app.db.models import Artifact, DataSource
 from app.tools.datasource import _store_schema, build_datasource_tools
 from app.tools.registry import ToolContext
+from tests.fixtures.sources import drop_source
 
 MARK = "（推断，未确认）"
 
@@ -182,4 +182,4 @@ async def test_tools_show_and_freeze_catalog_and_record_query_tables(scenic_db):
             ref = await session.get(Artifact, (result["artifact"], run_id))
         assert ref.meta == {"source_id": row.id, "source": row.name, "tables": ["visits", "parks"]}
     finally:
-        await engines.invalidate(row.id)
+        await drop_source(row.id)

backend/tests/test_copilot_sqlcheck.py

@@ -20,6 +20,7 @@ from app.data.engine import engines
 from app.data.introspect import introspect
 from app.data.sqlcheck import SqlChecker
 from tests.fixtures.catalog import scenic_notes
+from tests.fixtures.sources import drop_source
 
 FANOUT_SQL = ("SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id "
               "WHERE o.status = 1")
@@ -177,10 +178,7 @@ async def scenic_source(scenic_db):
             await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
         source_id = row.id
     yield name
-    async with SessionLocal() as session:
-        row = await session.get(DataSource, source_id)
-        row.enabled = False
-        await session.commit()
+    await drop_source(source_id)
 
 
 async def _stream(monkeypatch, model, source):

backend/tests/test_publish_sqlcheck.py

@@ -23,6 +23,7 @@ from app.engine.governance import lint_for_publish, publish_issues
 from app.engine.schema import GraphSpec
 from app.main import app
 from tests.fixtures.catalog import scenic_notes
+from tests.fixtures.sources import drop_source
 
 FANOUT_SQL = ("SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id "
               "WHERE o.status = 1")
@@ -133,10 +134,7 @@ async def scenic_source(scenic_db):
             await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
         source_id = row.id
     yield name
-    async with SessionLocal() as session:
-        row = await session.get(DataSource, source_id)
-        row.enabled = False
-        await session.commit()
+    await drop_source(source_id)
 
 
 @pytest.fixture

backend/tests/test_runtime_sqlcheck.py

@@ -32,6 +32,7 @@ from app.main import app
 from app.tools.datasource import build_datasource_tools
 from app.tools.registry import ToolContext
 from tests.fixtures.catalog import scenic_notes
+from tests.fixtures.sources import drop_source
 
 FANOUT_SQL = ("SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id "
               "WHERE o.status = 1")
@@ -54,6 +55,16 @@ async def _schema(path: str) -> dict:
     return _CACHE[path]
 
 
+_CREATED: list[str] = []
+
+
+@pytest.fixture(autouse=True)
+async def _drop_created_sources():
+    yield
+    while _CREATED:
+        await drop_source(_CREATED.pop())
+
+
 async def _source(path: str, *, with_catalog: bool = True) -> tuple[str, str]:
     name = f"scenic_{uuid.uuid4().hex[:8]}"
     async with SessionLocal() as session:
@@ -61,6 +72,7 @@ async def _source(path: str, *, with_catalog: bool = True) -> tuple[str, str]:
         row.schema_cache = await _schema(path)
         session.add(row)
         await session.commit()
+        _CREATED.append(row.id)
         if with_catalog:
             for table, notes in scenic_notes.notes().items():
                 await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
@@ -69,9 +81,8 @@ async def _source(path: str, *, with_catalog: bool = True) -> tuple[str, str]:
 
 @pytest.fixture
 async def scenic_source(scenic_db):
-    source_id, name = await _source(scenic_db)
+    _, name = await _source(scenic_db)
     yield name
-    await engines.invalidate(source_id)
 
 
 async def _query_tool(name: str):

backend/tests/test_truncated_results.py

@@ -34,6 +34,7 @@ from app.engine.expressions import TruncatedUse, eval_expression
 from app.engine.issuance import decide_tier, incomplete_gaps
 from app.engine.runner import run_manager
 from app.main import app
+from tests.fixtures.sources import drop_source
 
 # --------------------------------------------------------------------------
 # 求值器：整组用到截断结果时记一笔，单格不记
@@ -345,7 +346,7 @@ async def park(ticket_db, engine_up):
     # 连接池按数据源 id 缓存：换了库文件不作废，查的还是上一个测试的库
     await engines.invalidate(source_id)
     yield
-    await engines.invalidate(source_id)
+    await drop_source(source_id)
 
 
 def node(nid, ntype, **config):