测试：conftest 把 AGENTLAB_SECRET_KEY、AGENTLAB_SCHEDULER 挪到任何 app.* 导入之前（PostgreSQL 分支先导入 app.db.base 装连接池，settings 那一刻就定下来了，以前单独跑 test_schedule_events_api.py 在 PostgreSQL 上 scheduler_on 是 True）；test_trigger_options 按名字（q）查要核对的那几个对象，不再指望一页 100 个装得下同一个库里别的测试建的工作流和数据源

backend/tests/conftest.py

@@ -19,11 +19,18 @@ from pathlib import Path
 
 _TMP = Path(tempfile.mkdtemp(prefix="agentlab-tests-"))
 os.environ["AGENTLAB_DATA_DIR"] = str(_TMP)
+# 下面这几项也都要在任何 app.* 导入之前设（两种库同一个顺序）：PostgreSQL 分支要先导入 app.db.base 装连接池，settings 在那一刻
+# 就按当时的环境定下来了——以前这两项写在那之后，PostgreSQL 上 AGENTLAB_SCHEDULER 没生效（scheduler_on 成了 True）
+# 别把真实的 key 带进测试环境
+os.environ.setdefault("AGENTLAB_SECRET_KEY", "test-only-not-a-real-key")
+# 定时任务的调度循环不在测试里起：走 lifespan 的测试不该到点替人发起运行。要测调度的直接调它
+os.environ.setdefault("AGENTLAB_SCHEDULER", "0")
 
 
 def _postgres_database(admin_url: str) -> str:
     """AGENTLAB_TEST_DATABASE_URL 给的是管理用的库：每个 xdist 进程在同一台服务器上建一个自己的空库（库名加进程号后缀，
-    已有的同名库先删掉），进程退出时删掉。返回这个库的地址。scripts/test-postgres.sh 起临时容器、设好这个变量。"""
+    已有的同名库先删掉），进程退出时删掉。返回这个库的地址。scripts/test-postgres.sh 起临时容器、设好这个变量。
+    这里导入的 app.db.urls 只用 SQLAlchemy 的地址解析，不碰 settings（settings 要等环境变量都设好）。"""
     import psycopg
     from sqlalchemy.engine import make_url
 
@@ -64,10 +71,6 @@ if _TEST_PG:
     from tests.pgpool import install as _loop_local_pool  # noqa: E402
 
     _loop_local_pool(_app_engine)
-# 别把真实的 key 带进测试环境
-os.environ.setdefault("AGENTLAB_SECRET_KEY", "test-only-not-a-real-key")
-# 定时任务的调度循环不在测试里起：走 lifespan 的测试不该到点替人发起运行。要测调度的直接调它
-os.environ.setdefault("AGENTLAB_SCHEDULER", "0")
 
 import pytest  # noqa: E402
 

backend/tests/test_schedule_events_api.py

@@ -356,15 +356,25 @@ async def test_trigger_options(client):
     table = await upload_table("门店销售")
     flat = await upload_table("门店名单", mode="replace")
     wf, wf_name = await workflow("日报")
-    draft, _ = await workflow("草稿", published=False)
+    draft, draft_name = await workflow("草稿", published=False)
     r = await client.get("/api/schedules/trigger-options")
     assert r.status_code == 200
     out = r.json()
     events = {e["event"]: e for e in out["events"]}
     assert list(events) == ["table.version", "workflow.finished", "source.changed"]
-    tables = {o["id"]: o for o in events["table.version"]["objects"]}
+
+    # 对象按名字排、一页 100 个：同一个库里别的测试建的工作流、数据源多了，这里建的可能排到第一页外面。按名字（q）查要的那几个
+    async def objects(event: str, name: str) -> dict[str, dict]:
+        got = await client.get("/api/schedules/trigger-options", params={"event": event, "q": name})
+        assert got.status_code == 200
+        [entry] = got.json()["events"]
+        return {o["id"]: o for o in entry["objects"]}
+
+    async with SessionLocal() as session:
+        table_name, flat_name = [(await session.get(DataSource, i)).name for i in (table, flat)]
+    tables = {**await objects("table.version", table_name), **await objects("table.version", flat_name)}
     assert tables[table]["ok"] is True and tables[flat]["ok"] is False and "按期累积" in tables[flat]["reason"]
-    flows = {o["id"]: o for o in events["workflow.finished"]["objects"]}
+    flows = {**await objects("workflow.finished", wf_name), **await objects("workflow.finished", draft_name)}
     assert flows[wf] == {"id": wf, "name": wf_name, "ok": True, "reason": None}
     assert flows[draft]["ok"] is False and "发布" in flows[draft]["reason"]
     assert "hook" in [o["kind"] for o in out["origins"]] and out["max_depth"] == 5