mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 03:11:59 +00:00
ThearchyHelios
ed72ee68d6
- 剖析核对关联关系时，覆盖率从够有确证的线（95%）掉到线以下的记一条 coverage_drop（取值变化，warning）：剖析本身照旧 按规则降回推断（人工确认的只补上新的覆盖率），这里只把「覆盖率下降」留进变化记录。人点的剖析也记（变化记录全站一份）。 关系的发现里多带 previous_coverage。 - 处理动作「清理目录」逐表清：这张表上说到被删的列的项，加上别的表指向它的关联关系（表删了时尤其如此）；记这张表的 目录版本，别的表也动了的另记 catalog_versions。 - 剖析任务的逐表进度多一类 observe（取值观察的四类合成一组），前端「取值观察 N/M」用它。 - 定时维护补一条测试：AGENTLAB_SCHEDULER 关着时调度循环不起，到点的维护不触发、不建任务。 - scripts/dialect-it.sh / dialect_it.py 扩到第 2 波的对象索引字典查询（走字典查询不退回 Inspector、类型、表注释、行数 估算、Oracle 小写名的 quote）和这一波的取值观察（经剖析引擎只观察：行数、小表码值整表、大表 PG / Oracle 块采样、MySQL 前 N 行、最新业务日期排除无效日期，max_date_sql 直接跑）。夹具的订单表加了表注释。本机跑了一次四种库的容器 （postgres:16、mysql:8.4、mariadb:11、gvenzl/oracle-free:23-slim）：121 项全过、0 项不通过，跑完容器已删、镜像留着。 验证：后端全量 14046 通过（11 跳过、8 预期失败）。
236 lines
12 KiB
Python
236 lines
12 KiB
Python
"""数据源的定时维护（app/data/source_maintenance.py，next-round 03 B4、总方案 X4）。
|
||
|
||
定时只有一套：维护就是一条 target_kind=source_maintenance 的 Schedule，认领、补跑、积压只记一条都是 scheduler 那一套。
|
||
这里守：
|
||
|
||
- 到点按顺序建探查任务 → 剖析任务（范围按选的步骤），任务上记 trigger=schedule 和 fire_at；剖析排在同一次触发的探查后面
|
||
（同源队串行：探查做完剖析才开始）；触发记录写明建了哪几个任务；
|
||
- 照实记没执行的：剖析没开时剖析、观察两步不执行（探查照做）；数据源停用、不存在、是上传表格时整次不执行；上一次的探查
|
||
还没做完时这次跳过；
|
||
- 调度循环的认领：错过的只补最近一次、积压只记一条；
|
||
- 维护步骤的接口：不合规 422，合规存进 options、数据源表单不显示它；核对目标（check_target）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
import uuid
|
||
from datetime import timedelta
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy import delete, select
|
||
|
||
from app.data import catalog, catalog_profile_jobs, introspect_jobs, source_maintenance as sm
|
||
from app.db.base import SessionLocal, utcnow
|
||
from app.db.models import CatalogProfileJob, DataSource, Schedule, ScheduleFire
|
||
from app.db.models_sources import IntrospectJob
|
||
from app.main import app
|
||
from app.scheduler import loop, service
|
||
from tests.fixtures.sources import drop_source, introspect_now
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
async def _source(client, path, *, profile: bool = True) -> str:
|
||
db = sqlite3.connect(path)
|
||
db.executescript("""
|
||
CREATE TABLE orders (id INTEGER PRIMARY KEY, status VARCHAR(10), paid_on DATE);
|
||
INSERT INTO orders VALUES (1, 'paid', '2026-10-01'), (2, 'void', '2026-10-02');
|
||
""")
|
||
db.commit()
|
||
db.close()
|
||
r = await client.post("/api/datasources", json={"name": f"mnt_{uuid.uuid4().hex[:8]}", "kind": "sqlite",
|
||
"database": str(path), "readonly": True})
|
||
sid = r.json()["id"]
|
||
await introspect_now(client, sid)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
if profile:
|
||
row.options = {**(row.options or {}), "catalog_profile": {"enabled": True}}
|
||
await session.commit()
|
||
await catalog.write_entry(session, sid, "orders", {
|
||
"columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human")}},
|
||
}, if_version=0, actor="测试员")
|
||
return sid
|
||
|
||
|
||
async def _schedule(sid: str, **over) -> str:
|
||
async with SessionLocal() as session:
|
||
fields = {"name": f"维护-{uuid.uuid4().hex[:6]}", "cron": "0 3 * * *", "timezone": "Asia/Shanghai",
|
||
**service.target_columns("source_maintenance", sid), **over}
|
||
row = Schedule(**fields)
|
||
session.add(row)
|
||
await session.commit()
|
||
return row.id
|
||
|
||
|
||
@pytest.fixture
|
||
async def made():
|
||
ids: list[str] = []
|
||
yield ids
|
||
async with SessionLocal() as session:
|
||
await session.execute(delete(Schedule).where(Schedule.id.in_(ids)))
|
||
await session.commit()
|
||
|
||
|
||
async def _wait_jobs(fire: ScheduleFire) -> None:
|
||
for job in (fire.inputs or {}).get("jobs") or []:
|
||
manager = introspect_jobs.manager if job["kind"] == "introspect" else catalog_profile_jobs.manager
|
||
await manager.wait(job["id"], timeout=60)
|
||
|
||
|
||
async def test_fire_builds_introspect_then_profile_in_order(client, tmp_path, made):
|
||
sid = await _source(client, tmp_path / "a.db")
|
||
try:
|
||
schedule_id = await _schedule(sid)
|
||
made.append(schedule_id)
|
||
at = utcnow().replace(microsecond=0)
|
||
fire = await service.fire_once(schedule_id, at, "manual", actor="测试员")
|
||
assert fire.status == "started" and fire.reason is None, fire.reason
|
||
jobs = fire.inputs["jobs"]
|
||
assert [j["kind"] for j in jobs] == ["introspect", "profile"] and jobs[1]["scope"] == "maintenance"
|
||
assert fire.inputs["steps"] == list(sm.STEPS)
|
||
await _wait_jobs(fire)
|
||
async with SessionLocal() as session:
|
||
intro = await session.get(IntrospectJob, jobs[0]["id"])
|
||
prof = await session.get(CatalogProfileJob, jobs[1]["id"])
|
||
assert intro.trigger == "schedule" and intro.scope["fire_at"] == at.isoformat()
|
||
assert intro.scope["schedule_id"] == schedule_id and intro.created_by == "测试员"
|
||
assert catalog_profile_jobs.trigger_of(prof) == {"trigger": "schedule", "fire_at": at.isoformat(),
|
||
"schedule_id": schedule_id}
|
||
assert prof.started_at >= intro.finished_at, "同源串行：探查做完，剖析才开始"
|
||
assert prof.status in ("succeeded", "partial"), prof.error
|
||
# 同一次触发再来一遍：交回已有的那一条，不再建任务
|
||
again = await service.fire_once(schedule_id, at, "manual", actor="测试员")
|
||
assert again.id == fire.id
|
||
finally:
|
||
await drop_source(sid)
|
||
|
||
|
||
async def test_profile_off_records_steps_not_run_and_still_introspects(client, tmp_path, made):
|
||
sid = await _source(client, tmp_path / "b.db", profile=False)
|
||
try:
|
||
schedule_id = await _schedule(sid)
|
||
made.append(schedule_id)
|
||
fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual")
|
||
assert fire.status == "started"
|
||
assert [j["kind"] for j in fire.inputs["jobs"]] == ["introspect"]
|
||
assert {s["step"] for s in fire.inputs["skipped"]} == {"profile_needed", "observe"}
|
||
assert "剖析需要的表没有执行" in fire.reason and "取值观察没有执行" in fire.reason and "数据剖析" in fire.reason
|
||
await _wait_jobs(fire)
|
||
# 只选观察一步：剖析没开就整次没执行
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
row.options = sm.with_steps(row.options, ["observe"], actor="测试员")
|
||
await session.commit()
|
||
fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=1), "manual")
|
||
assert fire.status == "failed_to_start" and "取值观察没有执行" in fire.reason
|
||
finally:
|
||
await drop_source(sid)
|
||
|
||
|
||
async def test_disabled_missing_and_busy_sources(client, tmp_path, made):
|
||
sid = await _source(client, tmp_path / "c.db")
|
||
try:
|
||
schedule_id = await _schedule(sid)
|
||
made.append(schedule_id)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
row.enabled = False
|
||
await session.commit()
|
||
fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual")
|
||
assert fire.status == "failed_to_start" and "已停用" in fire.reason
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
row.enabled = True
|
||
session.add(IntrospectJob(source_id=sid, status="queued", scope={}))
|
||
await session.commit()
|
||
fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=2), "manual")
|
||
assert fire.status == "skipped_overlap" and "正在探查结构" in fire.reason
|
||
async with SessionLocal() as session:
|
||
await session.execute(delete(IntrospectJob).where(IntrospectJob.source_id == sid,
|
||
IntrospectJob.status == "queued"))
|
||
await session.commit()
|
||
finally:
|
||
await drop_source(sid)
|
||
fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=3), "manual")
|
||
assert fire.status == "failed_to_start" and fire.reason == sm.SOURCE_GONE
|
||
|
||
|
||
async def test_loop_claims_and_only_the_latest_missed_fire_runs(client, tmp_path, made):
|
||
sid = await _source(client, tmp_path / "d.db")
|
||
try:
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
row.options = sm.with_steps(row.options, ["introspect"], actor="测试员")
|
||
await session.commit()
|
||
now = utcnow().replace(second=0, microsecond=0)
|
||
# 每 10 分钟一次，停机错过了好几次：只补最近一次，更早的积压只记一条
|
||
schedule_id = await _schedule(sid, cron="*/10 * * * *", timezone="UTC",
|
||
next_run_at=now - timedelta(minutes=40), misfire_grace_s=3600)
|
||
made.append(schedule_id)
|
||
fires = await loop.fire_due(now + timedelta(minutes=1))
|
||
assert sorted(f.status for f in fires) == ["skipped_misfire", "started"], [(f.kind, f.status) for f in fires]
|
||
started = next(f for f in fires if f.status == "started")
|
||
assert [j["kind"] for j in started.inputs["jobs"]] == ["introspect"]
|
||
await _wait_jobs(started)
|
||
async with SessionLocal() as session:
|
||
sched = await session.get(Schedule, schedule_id)
|
||
count = len(list((await session.execute(select(ScheduleFire).where(
|
||
ScheduleFire.schedule_id == schedule_id))).scalars()))
|
||
assert sched.next_run_at > now and count == 2
|
||
finally:
|
||
await drop_source(sid)
|
||
|
||
|
||
async def test_steps_api_and_check_target(client, tmp_path):
|
||
sid = await _source(client, tmp_path / "e.db")
|
||
try:
|
||
url = f"/api/datasources/{sid}/maintenance"
|
||
body = (await client.get(url)).json()
|
||
assert body["steps"] == list(sm.STEPS) and body["profile_enabled"] is True and body["schedules"] == []
|
||
assert (await client.put(url, json={"steps": ["nope"]})).status_code == 422
|
||
assert (await client.put(url, json={"steps": []})).status_code == 422
|
||
r = await client.put(url, json={"steps": ["observe", "introspect"]})
|
||
assert r.status_code == 200 and r.json()["steps"] == ["introspect", "observe"], "按固定顺序存"
|
||
[shown] = [s["options"] for s in (await client.get("/api/datasources")).json() if s["id"] == sid]
|
||
assert sm.MAINTENANCE_OPTION not in shown, "数据源表单不显示、不提交它"
|
||
async with SessionLocal() as session:
|
||
assert await sm.check_target(session, sid) is None
|
||
assert await sm.check_target(session, "missing") == sm.SOURCE_GONE
|
||
row = await session.get(DataSource, sid)
|
||
row.origin = "upload"
|
||
await session.commit()
|
||
assert await sm.check_target(session, sid) == sm.NOT_MAINTAINED
|
||
finally:
|
||
await drop_source(sid)
|
||
|
||
|
||
async def test_scheduler_switched_off_never_fires_maintenance(client, tmp_path, made):
|
||
"""AGENTLAB_SCHEDULER 关着（测试环境就是关着的，见 conftest）：调度循环不起，到点的维护不会触发——拷来的数据目录不替原库
|
||
查业务库。照实说：这时没有触发记录（定时只有一套，没起循环就没有「这一拍」）。"""
|
||
import asyncio
|
||
|
||
from app import scheduler
|
||
|
||
sid = await _source(client, tmp_path / "f.db")
|
||
try:
|
||
schedule_id = await _schedule(sid, next_run_at=utcnow() - timedelta(minutes=1))
|
||
made.append(schedule_id)
|
||
assert scheduler.scheduler_enabled() is False
|
||
await scheduler.start()
|
||
await asyncio.sleep(0.3)
|
||
await scheduler.stop()
|
||
async with SessionLocal() as session:
|
||
fires = list((await session.execute(select(ScheduleFire).where(
|
||
ScheduleFire.schedule_id == schedule_id))).scalars())
|
||
jobs = list((await session.execute(select(IntrospectJob).where(
|
||
IntrospectJob.source_id == sid, IntrospectJob.trigger == "schedule"))).scalars())
|
||
assert fires == [] and jobs == []
|
||
finally:
|
||
await drop_source(sid)
|