chenshoulin
58850314cf
用「WCF + 通用存储过程网关」接口替代 Selenium（D://Learning//Python//工单自动化）导入工单： 单 Sprint 12 个项目 175s 跑完（旧方案 25~35 分钟），失败可断点续跑且不重复建单。 - 分层：transport（HTTP/协议）/ api（存储过程语义）/ domain（纯规则）/ workflow（阶段编排）/ store（SQLite 台账）/ report / CLI - 幂等：台账驱动断点续跑；dry-run 与跳过行不污染进度；--rollback-on-fail 清理半成品 - 命令：plan / apply / report / status，真实写入需 --yes - tools：只读探针、存储过程源码导出（sp_helptext）、导入诊断 - docs：调研与设计文档（含联调踩到的 5 个坑）+ 6 个关键存储过程源码 实测：Sprint 2026 Sprint 10 的 12 个项目全部完成建单→导入→确认→推送 TFS， 工单 225134–225145，35 个功能模块，重跑 plan 全部 skip-pushed。69 个单测通过。
77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""台账测试：dry-run 隔离、跨 run 续跑、字段合并。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from qtimport.store import Store
|
|
|
|
|
|
class StoreTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.folder = tempfile.TemporaryDirectory()
|
|
self.store = Store(os.path.join(self.folder.name, "t.db"))
|
|
|
|
def tearDown(self) -> None:
|
|
self.store.close()
|
|
self.folder.cleanup()
|
|
|
|
def _job(self, run_id: str, stage: str, worksheet_id: int | None = 1) -> None:
|
|
self.store.upsert_job(
|
|
run_id,
|
|
project_id="p1",
|
|
project_name="P",
|
|
system_name="2026 Sprint 10",
|
|
source_file="f.xlsx",
|
|
action="create",
|
|
stage=stage,
|
|
worksheet_id=worksheet_id,
|
|
)
|
|
|
|
def test_latest_stage_across_runs(self) -> None:
|
|
self.store.start_run("r1", "s", "d", dry_run=False)
|
|
self._job("r1", "WS_CREATED")
|
|
self.store.start_run("r2", "s", "d", dry_run=False)
|
|
self._job("r2", "WS_FILLED")
|
|
self.assertEqual(self.store.latest_stage("p1", "2026 Sprint 10"), "WS_FILLED")
|
|
|
|
def test_dry_run_does_not_pollute_ledger(self) -> None:
|
|
self.store.start_run("real", "s", "d", dry_run=False)
|
|
self._job("real", "WS_FILLED")
|
|
self.store.start_run("dry", "s", "d", dry_run=True)
|
|
self._job("dry", "TFS_PUSHED")
|
|
self.assertEqual(
|
|
self.store.latest_stage("p1", "2026 Sprint 10"),
|
|
"WS_FILLED",
|
|
"dry-run 不能把项目标成已推送",
|
|
)
|
|
|
|
def test_worksheet_id_survives_partial_update(self) -> None:
|
|
self.store.start_run("r1", "s", "d", dry_run=False)
|
|
self._job("r1", "WS_CREATED", worksheet_id=777)
|
|
self._job("r1", "WS_FILLED", worksheet_id=None)
|
|
job = self.store.latest_job("p1", "2026 Sprint 10")
|
|
self.assertEqual(job["worksheet_id"], 777)
|
|
self.assertEqual(job["stage"], "WS_FILLED")
|
|
|
|
def test_skipped_row_does_not_mask_progress(self) -> None:
|
|
"""跳过行不带进度：不能把「已推送」覆盖成「SKIPPED」，否则下次会重推 TFS。"""
|
|
self.store.start_run("r1", "s", "d", dry_run=False)
|
|
self._job("r1", "TFS_PUSHED")
|
|
self.store.start_run("r2", "s", "d", dry_run=False)
|
|
self._job("r2", "SKIPPED")
|
|
job = self.store.latest_job("p1", "2026 Sprint 10")
|
|
self.assertEqual(job["stage"], "TFS_PUSHED")
|
|
|
|
def test_events_ordered(self) -> None:
|
|
self.store.start_run("r1", "s", "d", dry_run=False)
|
|
self.store.add_event("r1", stage="A", action="ok", detail="1")
|
|
self.store.add_event("r1", stage="B", action="ok", detail="2")
|
|
self.assertEqual([e["stage"] for e in self.store.events("r1")], ["A", "B"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|