分析测试的「今天」按题集的时区取,不看本机时区:题集里的期次(上周、本周)按 suite.timezone 算,生成虚构库、判今天是不是周一也照它,本机在别的时区(或跨午夜那几个钟头)时两边不是同一天

- test_eval_analysis_runs.py:TZ 从题集读,新增 now()(datetime.now(ZoneInfo(TZ))),生成库用 now().date(),原来写死的 Asia/Shanghai 去掉
- test_eval_analysis_suite.py:从题集内容取时区算今天,判周一也按它
- test_evals_api.py:零售样例题集的「今天」同样从题集取时区
This commit is contained in:
ThearchyHelios
2026-10-21 14:57:06 +00:00
parent 1c117d78aa
commit 454e82e211
3 changed files with 13 additions and 8 deletions
+3 -2
View File
@@ -32,8 +32,9 @@ from app.eval.gold import Context
from tests import eval_samples as es from tests import eval_samples as es
from tests.fixtures.acceptance import retail_trends from tests.fixtures.acceptance import retail_trends
TZ = "Asia/Shanghai"
SUITE = json.loads((Path(__file__).parent / "fixtures" / "eval" / "analysis_suite.json").read_text(encoding="utf-8")) SUITE = json.loads((Path(__file__).parent / "fixtures" / "eval" / "analysis_suite.json").read_text(encoding="utf-8"))
#: 题集的时区:题目里的期次({{ schedule.* }})按它算,生成库的「今天」也按它取,不看本机时区
TZ = SUITE["suite"]["timezone"]
#: 题集里的第 1 题(上周销售额下降主要来自华东)和第 3 题(订单数的常规波动) #: 题集里的第 1 题(上周销售额下降主要来自华东)和第 3 题(订单数的常规波动)
REGION = next(c for c in SUITE["cases"] if c["seq"] == 1) REGION = next(c for c in SUITE["cases"] if c["seq"] == 1)
ORDERS = next(c for c in SUITE["cases"] if c["seq"] == 3) ORDERS = next(c for c in SUITE["cases"] if c["seq"] == 3)
@@ -53,7 +54,7 @@ async def engine_up(monkeypatch):
@pytest.fixture @pytest.fixture
async def source(tmp_path, scratch_source): async def source(tmp_path, scratch_source):
path = retail_trends.build(tmp_path / "trends.db", date.today()) path = retail_trends.build(tmp_path / "trends.db", now().date())
sid = await scratch_source(retail_trends.SOURCE_NAME, path, schema=True) sid = await scratch_source(retail_trends.SOURCE_NAME, path, schema=True)
return {"id": sid, "name": retail_trends.SOURCE_NAME, "path": path} return {"id": sid, "name": retail_trends.SOURCE_NAME, "path": path}
+7 -4
View File
@@ -4,9 +4,10 @@
from __future__ import annotations from __future__ import annotations
import json import json
from datetime import date from datetime import datetime
from pathlib import Path from pathlib import Path
from urllib.parse import quote from urllib.parse import quote
from zoneinfo import ZoneInfo
import pytest import pytest
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
@@ -33,9 +34,11 @@ async def engine_up():
async def test_the_analysis_suite_imports_and_every_expectation_computes(tmp_path, scratch_source): async def test_the_analysis_suite_imports_and_every_expectation_computes(tmp_path, scratch_source):
path = retail_trends.build(tmp_path / "trends.db", date.today())
sid = await scratch_source(retail_trends.SOURCE_NAME, path, schema=True)
content = json.loads(SUITE_FILE.read_text(encoding="utf-8")) content = json.loads(SUITE_FILE.read_text(encoding="utf-8"))
# 「今天」按题集的时区取(题目里的期次按它算),不看本机时区:本机在别的时区时,跨日那几个钟头两边的今天不是同一天
today = datetime.now(ZoneInfo(content["suite"]["timezone"])).date()
path = retail_trends.build(tmp_path / "trends.db", today)
sid = await scratch_source(retail_trends.SOURCE_NAME, path, schema=True)
assert {c["seq"] for c in content["cases"]} == set(WANTED) assert {c["seq"] for c in content["cases"]} == set(WANTED)
assert all(c["gold"]["datasource"] == retail_trends.SOURCE_NAME for c in content["cases"]) assert all(c["gold"]["datasource"] == retail_trends.SOURCE_NAME for c in content["cases"])
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
@@ -44,7 +47,7 @@ async def test_the_analysis_suite_imports_and_every_expectation_computes(tmp_pat
assert res.json()["imported"] == len(WANTED) and res.json()["skipped"] == [] assert res.json()["imported"] == len(WANTED) and res.json()["skipped"] == []
suite = res.json()["suite"] suite = res.json()["suite"]
assert all(c["gold"]["datasource_id"] == sid for c in suite["cases"]) assert all(c["gold"]["datasource_id"] == sid for c in suite["cases"])
monday = date.today().weekday() == 0 monday = today.weekday() == 0
for case in suite["cases"]: for case in suite["cases"]:
assert case["trial"]["status"] == "ok", (case["seq"], case["trial"]) assert case["trial"]["status"] == "ok", (case["seq"], case["trial"])
trial = (await client.post(f"/api/evals/suites/{suite['id']}/cases/{case['id']}/trial")).json()["trial"] trial = (await client.post(f"/api/evals/suites/{suite['id']}/cases/{case['id']}/trial")).json()["trial"]
+3 -2
View File
@@ -386,7 +386,9 @@ async def test_export_then_import_matches_sources_by_name(client, source):
async def test_the_retail_sample_suite_trials_on_the_retail_database(client, tmp_path): async def test_the_retail_sample_suite_trials_on_the_retail_database(client, tmp_path):
"""仓库里附的零售样例题集:在虚构零售库上导入,每道题都试算通过,期望和直接查库一样,陷阱按口径算。""" """仓库里附的零售样例题集:在虚构零售库上导入,每道题都试算通过,期望和直接查库一样,陷阱按口径算。"""
today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # 「今天」按题集的时区取(题目里的期次按它算),不看本机时区:本机在别的时区时,跨日那几个钟头两边的今天不是同一天
content = json.loads(SUITE_FILE.read_text(encoding="utf-8"))
today = datetime.now(ZoneInfo(content["suite"]["timezone"])).date()
path = retail.build(tmp_path / "retail.db", today) path = retail.build(tmp_path / "retail.db", today)
await drop_source(name=retail.SOURCE_NAME) await drop_source(name=retail.SOURCE_NAME)
async with SessionLocal() as session: async with SessionLocal() as session:
@@ -398,7 +400,6 @@ async def test_the_retail_sample_suite_trials_on_the_retail_database(client, tmp
await session.commit() await session.commit()
sid = row.id sid = row.id
try: try:
content = json.loads(SUITE_FILE.read_text(encoding="utf-8"))
res = await client.post("/api/evals/suites/import", json={"content": content}, headers=ACTOR) res = await client.post("/api/evals/suites/import", json={"content": content}, headers=ACTOR)
assert res.status_code == 201, res.text assert res.status_code == 201, res.text
assert res.json()["imported"] == 6 and res.json()["skipped"] == [] assert res.json()["imported"] == 6 and res.json()["skipped"] == []