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 个单测通过。
141 lines
5.7 KiB
Python
141 lines
5.7 KiB
Python
"""domain 层规则测试：项目匹配、动作判定、版本推进、Excel 预检。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
|
|
from qtimport.domain import Action, decide, match_project, next_transition, precheck_xlsx
|
|
from qtimport.models import Project, WorkSheet, normalize_iteration, normalize_name
|
|
|
|
|
|
def ws(worksheet_id: int = 1, status: str = "00", code: str = "W_1") -> WorkSheet:
|
|
return WorkSheet(worksheet_id=worksheet_id, code=code, status=status)
|
|
|
|
|
|
class NormalizeTest(unittest.TestCase):
|
|
def test_space_and_dash(self) -> None:
|
|
self.assertEqual(normalize_name("资管 - 华泰资管"), normalize_name("资管－华泰资管"))
|
|
self.assertEqual(normalize_name("信托 - 上海信托"), normalize_name("信托-上海信托"))
|
|
|
|
def test_iteration_leading_zero(self) -> None:
|
|
self.assertEqual(normalize_iteration("2026 Sprint 06"), normalize_iteration("2026 Sprint 6"))
|
|
self.assertNotEqual(normalize_iteration("2026 Sprint 10"), normalize_iteration("2026 Sprint 1"))
|
|
|
|
|
|
class MatchProjectTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.projects = [
|
|
Project("p1", "资管 - 华泰资管"),
|
|
Project("p2", "资管 - 华泰保险资管"),
|
|
Project("p3", "信托 - 上海信托"),
|
|
]
|
|
|
|
def test_exact_near_names_do_not_collide(self) -> None:
|
|
"""Selenium 用 contains，会把「华泰资管」误命中「华泰保险资管」；这里必须精确。"""
|
|
project, reason = match_project("资管 - 华泰资管", self.projects)
|
|
self.assertEqual(project.project_id, "p1")
|
|
self.assertEqual(reason, "")
|
|
|
|
def test_missing_gives_hint(self) -> None:
|
|
project, reason = match_project("资管 - 华泰", self.projects)
|
|
self.assertIsNone(project)
|
|
self.assertIn("相近项目", reason)
|
|
|
|
def test_ambiguous(self) -> None:
|
|
projects = [Project("a", "X - Y"), Project("b", "X-Y")]
|
|
project, reason = match_project("X - Y", projects)
|
|
self.assertIsNone(project)
|
|
self.assertIn("歧义", reason)
|
|
|
|
|
|
class DecideTest(unittest.TestCase):
|
|
def test_create_when_absent(self) -> None:
|
|
action, _, _ = decide([], ledger_stage=None, adopt=False)
|
|
self.assertEqual(action, Action.CREATE)
|
|
|
|
def test_skip_when_done(self) -> None:
|
|
action, _, _ = decide([ws(status="03")], ledger_stage=None, adopt=False)
|
|
self.assertEqual(action, Action.SKIP_DONE)
|
|
|
|
def test_skip_pushed(self) -> None:
|
|
action, _, _ = decide([ws(status="01")], ledger_stage="TFS_PUSHED", adopt=False)
|
|
self.assertEqual(action, Action.SKIP_PUSHED)
|
|
|
|
def test_resume_from_ledger(self) -> None:
|
|
action, _, sheet = decide([ws()], ledger_stage="WS_FILLED", adopt=False)
|
|
self.assertEqual(action, Action.RESUME)
|
|
self.assertEqual(sheet.worksheet_id, 1)
|
|
|
|
def test_existing_without_ledger_skips(self) -> None:
|
|
action, reason, _ = decide([ws()], ledger_stage=None, adopt=False)
|
|
self.assertEqual(action, Action.SKIP_EXISTS)
|
|
self.assertIn("未接管", reason)
|
|
|
|
def test_adopt_takes_over(self) -> None:
|
|
action, _, _ = decide([ws()], ledger_stage=None, adopt=True)
|
|
self.assertEqual(action, Action.RESUME)
|
|
|
|
def test_writes_flag(self) -> None:
|
|
self.assertTrue(Action.CREATE.writes)
|
|
self.assertTrue(Action.RESUME.writes)
|
|
for action in (Action.SKIP_DONE, Action.SKIP_EXISTS, Action.SKIP_PUSHED, Action.ERROR):
|
|
self.assertFalse(action.writes)
|
|
|
|
|
|
class VersionTest(unittest.TestCase):
|
|
def test_first_step(self) -> None:
|
|
self.assertEqual(next_transition("技术责任人确认版"), ("CustomerLeader", "客户责任人确认版", 20))
|
|
|
|
def test_draft_like_first_step(self) -> None:
|
|
self.assertEqual(next_transition(""), ("CustomerLeader", "客户责任人确认版", 20))
|
|
|
|
def test_second_step(self) -> None:
|
|
self.assertEqual(next_transition("客户责任人确认版")[1], "甲方确认版")
|
|
|
|
def test_terminal_raises(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
next_transition("甲方确认版")
|
|
|
|
def test_unknown_raises(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
next_transition("甲方确认版-v2")
|
|
|
|
|
|
class PrecheckTest(unittest.TestCase):
|
|
def _make_xlsx(self, path: str, strings: list[str]) -> None:
|
|
shared = "<sst>" + "".join(f"<si><t>{text}</t></si>" for text in strings) + "</sst>"
|
|
with zipfile.ZipFile(path, "w") as archive:
|
|
archive.writestr("xl/workbook.xml", "<workbook/>")
|
|
archive.writestr("xl/sharedStrings.xml", shared)
|
|
|
|
def test_valid_template(self) -> None:
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = os.path.join(folder, "a.xlsx")
|
|
self._make_xlsx(path, ["工作量评估表", "服务模块", "服务内容"])
|
|
ok, why = precheck_xlsx(path)
|
|
self.assertTrue(ok, why)
|
|
|
|
def test_missing_title(self) -> None:
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = os.path.join(folder, "b.xlsx")
|
|
self._make_xlsx(path, ["随便一个表", "服务模块"])
|
|
ok, why = precheck_xlsx(path)
|
|
self.assertFalse(ok)
|
|
self.assertIn("工作量评估表", why)
|
|
|
|
def test_not_zip(self) -> None:
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = os.path.join(folder, "c.xlsx")
|
|
with open(path, "w", encoding="utf-8") as handle:
|
|
handle.write("not a zip")
|
|
ok, why = precheck_xlsx(path)
|
|
self.assertFalse(ok)
|
|
self.assertIn("xlsx", why)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|