修复手动搜索接口测试隔离
This commit is contained in:
@@ -46,12 +46,30 @@ def test_manual_agent_run_endpoint_starts_background_job(monkeypatch) -> None:
|
|||||||
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
||||||
get_settings.cache_clear()
|
get_settings.cache_clear()
|
||||||
started = []
|
started = []
|
||||||
|
engine = create_engine(
|
||||||
|
"sqlite:///:memory:",
|
||||||
|
connect_args={"check_same_thread": False},
|
||||||
|
poolclass=StaticPool,
|
||||||
|
future=True,
|
||||||
|
)
|
||||||
|
TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
||||||
|
Base.metadata.create_all(engine)
|
||||||
|
|
||||||
def fake_run_scheduled_job(settings):
|
def fake_run_scheduled_job(settings):
|
||||||
started.append(settings.env)
|
started.append(settings.env)
|
||||||
|
|
||||||
|
def override_get_db():
|
||||||
|
# 手动搜索测试只验证接口调度行为,运行中状态查询使用内存库。
|
||||||
|
db = TestingSessionLocal()
|
||||||
|
try:
|
||||||
|
yield db
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
monkeypatch.setattr("app.api.routes.run_scheduled_job", fake_run_scheduled_job)
|
monkeypatch.setattr("app.api.routes.run_scheduled_job", fake_run_scheduled_job)
|
||||||
with TestClient(create_app()) as client:
|
app = create_app()
|
||||||
|
app.dependency_overrides[get_db] = override_get_db
|
||||||
|
with TestClient(app) as client:
|
||||||
response = client.post("/agent/runs/daily")
|
response = client.post("/agent/runs/daily")
|
||||||
get_settings.cache_clear()
|
get_settings.cache_clear()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user