"""api 层：参数构造与返回判定的单元测试（不联网）。""" from __future__ import annotations import hashlib import json import unittest import urllib.parse from qtimport.api import QuickTeamApi from qtimport.errors import Category, QtError from qtimport.models import Project, TfsTeam class FakeTransport: def __init__(self) -> None: self.calls: list[tuple] = [] self.table: dict = {} self.result: object = 0 self.import_result: object = [{"Column1": 0}] self.tfs_result: object = [] self.upload_result = "E:\\tmp\\x.xlsx" def exec_table(self, connection, procedure, params, *, op="", write=False): self.calls.append(("exec_table", connection, procedure, params, write)) return self.table.get(procedure, []) def exec_result(self, connection, procedure, params, *, op=""): self.calls.append(("exec_result", connection, procedure, params)) return self.result def batch_import(self, procedure, params, file_path): self.calls.append(("batch_import", procedure, params, file_path)) return self.import_result def upload(self, local_path, folder, *, op="FileUpload"): self.calls.append(("upload", local_path, folder)) return self.upload_result def tfs(self, operation, payload=None, *, op="", write=False): self.calls.append(("tfs", operation, payload)) return self.tfs_result PROJECT = Project("0f52", "信托 - 上海信托") class LoginTest(unittest.TestCase): def test_password_is_md5(self) -> None: fake = FakeTransport() fake.table["dbo.usp_Authentication"] = [ {"GuidId": "g-1", "RealName": "陈首林", "UserName": "chen@goldenstand.cn"} ] api = QuickTeamApi(fake) user = api.login("chen@goldenstand.cn", "dummy-pass") self.assertEqual(user.user_id, "g-1") self.assertEqual(user.real_name, "陈首林") _, connection, procedure, params, _ = fake.calls[0] self.assertEqual(connection, "UserCenter") self.assertEqual(procedure, "dbo.usp_Authentication") sent = dict((name, value) for name, value, _ in params) self.assertEqual(sent["Password"], hashlib.md5(b"dummy-pass").hexdigest()) def test_empty_result_is_auth_error(self) -> None: api = QuickTeamApi(FakeTransport()) with self.assertRaises(QtError) as ctx: api.login("a", "b") self.assertEqual(ctx.exception.category, Category.AUTH) class FillWorkSheetTest(unittest.TestCase): def _api(self) -> tuple[QuickTeamApi, FakeTransport]: fake = FakeTransport() fake.table["QuickTeam.usp_UpdateWorkSheetById"] = [{"Result": 1}] return QuickTeamApi(fake), fake def test_payload(self) -> None: api, fake = self._api() api.fill_worksheet( 123, PROJECT, system_name="2026 Sprint 10", start_time="2026-10-01", end_time="2026-10-31", evaluator="陈首林", technical_leader="陈首林", customer_leader="杜衡", belong_team="资管投资人事业部", content="需求开发测试", operator="陈首林", ) _, connection, procedure, params, write = fake.calls[0] self.assertEqual(connection, "QuickTeam") self.assertEqual(procedure, "QuickTeam.usp_UpdateWorkSheetById") self.assertTrue(write, "写操作必须标记 write，避免被误重试") sent = {name: (value, dbtype) for name, value, dbtype in params} self.assertEqual(len(sent), 17) self.assertEqual(sent["Operation"][0], "new") self.assertEqual(sent["DeleteFlag"][0], "0") self.assertEqual(sent["SystemName"][0], "2026 Sprint 10") self.assertEqual(sent["ProjectId"][0], "0f52") self.assertEqual(sent["CustomerLeader"][0], "杜衡") self.assertEqual(sent["WorkSheetId"], (123, "string")) # 类型转换在 transport 层做 self.assertEqual(sent["Status"][0], "00", "新建工单从「等待技术责任人确认」开始") def test_zero_result_but_state_applied_is_success(self) -> None: """实测：UpdateWorkSheetById 的 Result 恒为 0，但数据已落库，不能因此判失败。""" fake = FakeTransport() fake.table["QuickTeam.usp_UpdateWorkSheetById"] = [{"Result": 0}] fake.table["QuickTeam.usp_GetWorkSheetById"] = [ {"WorkSheetId": 1, "SystemName": "2026 Sprint 10", "Status": "00", "CurrentVersionId": 5} ] api = QuickTeamApi(fake) self.assertEqual( api.fill_worksheet( 1, PROJECT, system_name="2026 Sprint 10", start_time="a", end_time="b", evaluator="e", technical_leader="t", customer_leader="c", belong_team="bt", content="ct", operator="op", ), 0, ) def test_zero_result_and_state_missing_raises(self) -> None: fake = FakeTransport() fake.table["QuickTeam.usp_UpdateWorkSheetById"] = [{"Result": 0}] fake.table["QuickTeam.usp_GetWorkSheetById"] = [{"WorkSheetId": 1, "SystemName": "", "Status": ""}] api = QuickTeamApi(fake) with self.assertRaises(QtError): api.fill_worksheet( 1, PROJECT, system_name="2026 Sprint 10", start_time="a", end_time="b", evaluator="e", technical_leader="t", customer_leader="c", belong_team="bt", content="ct", operator="op", ) class ImportExcelTest(unittest.TestCase): def test_success(self) -> None: fake = FakeTransport() api = QuickTeamApi(fake) api.import_excel(9, 260113, "E:\\tmp\\x.xlsx") _, procedure, params, path = fake.calls[0] self.assertEqual(procedure, "QuickTeam.usp_BatchImportFeatureFromExcel") self.assertEqual(params, [("WorkSheetId", 9, "int"), ("VersionId", 260113, "int")]) self.assertEqual(path, "E:\\tmp\\x.xlsx") def test_column1_nonzero_raises_excel(self) -> None: fake = FakeTransport() fake.import_result = [{"Column1": 3}] api = QuickTeamApi(fake) with self.assertRaises(QtError) as ctx: api.import_excel(9, 1, "p") self.assertEqual(ctx.exception.category, Category.EXCEL) def test_unexpected_payload(self) -> None: fake = FakeTransport() fake.import_result = None api = QuickTeamApi(fake) with self.assertRaises(QtError): api.import_excel(9, 1, "p") class WorkSheetListTest(unittest.TestCase): """项目工单列表：一次查全（该 SP 一行一个版本，工单本体行 VersionId 为空）。""" ROWS = [ {"WorkSheetId": 501, "WorkSheetCode": "W_501", "SystemName": "2026 Sprint 10", "Status": "01", "CurrentVersionId": 260220, "VersionId": "", "VersionCode": "", "StartTime": "/Date(1791388800000)/"}, {"WorkSheetId": 501, "WorkSheetCode": "技术责任人确认版", "SystemName": "2026 Sprint 10", "Status": "01", "CurrentVersionId": 260220, "VersionId": 260220, "VersionCode": "技术责任人确认版"}, {"WorkSheetId": 502, "WorkSheetCode": "客户责任人确认版", "SystemName": "2026 Sprint 9", "Status": "01", "CurrentVersionId": 260180, "VersionId": 260180, "VersionCode": "客户责任人确认版"}, ] def test_dedupes_by_worksheet_and_prefers_base_row(self) -> None: fake = FakeTransport() fake.table["QuickTeam.usp_GetWorkSheetsById"] = self.ROWS api = QuickTeamApi(fake) sheets = api.work_sheets("0f52") self.assertEqual([s.worksheet_id for s in sheets], [501, 502]) first = sheets[0] self.assertEqual(first.code, "W_501", "版本行把 WorkSheetCode 覆盖成了 VersionCode，必须取本体行") self.assertEqual(first.system_name, "2026 Sprint 10") self.assertEqual(first.status, "01") self.assertEqual(first.version_id, 260220) self.assertEqual(first.start_time, "2026-10-08") self.assertEqual(sheets[1].system_name, "2026 Sprint 9") class WorksheetVersionTest(unittest.TestCase): """版本行来自 usp_GetWorkSheetsById（列表页那个 SP，一行一个版本）。""" ROWS = [ # union 出来的「无版本」占位行，VersionId 是空串 {"WorkSheetId": 501, "VersionId": "", "VersionCode": "", "VersionStatus": "", "NextOperator": ""}, {"WorkSheetId": 501, "VersionId": 260299, "VersionCode": "客户责任人确认版", "VersionStatus": "Y", "NextOperator": "CustomerLeader"}, {"WorkSheetId": 501, "VersionId": 260300, "VersionCode": "技术责任人确认版", "VersionStatus": "N", "NextOperator": "TechnicalLeader"}, {"WorkSheetId": 502, "VersionId": 260400, "VersionCode": "技术责任人确认版", "VersionStatus": "N", "NextOperator": "TechnicalLeader"}, ] def _api(self) -> tuple[QuickTeamApi, FakeTransport]: fake = FakeTransport() fake.table["QuickTeam.usp_GetWorkSheetsById"] = self.ROWS return QuickTeamApi(fake), fake def test_filters_placeholder_and_other_worksheets(self) -> None: api, fake = self._api() versions = api.worksheet_versions("0f52", 501) self.assertEqual([v.version_id for v in versions], [260299, 260300]) _, connection, procedure, params, write = fake.calls[0] self.assertEqual(procedure, "QuickTeam.usp_GetWorkSheetsById") self.assertFalse(write) def test_pending_version_prefers_n(self) -> None: """CurrentVersionId 是 NULL，导入必须落到『待确认』那一版，否则会撞 FK。""" api, _ = self._api() pending = api.pending_version("0f52", 501) self.assertIsNotNone(pending) self.assertEqual(pending.version_id, 260300) self.assertEqual(pending.version_code, "技术责任人确认版") def test_pending_falls_back_to_latest_when_all_confirmed(self) -> None: fake = FakeTransport() fake.table["QuickTeam.usp_GetWorkSheetsById"] = [ {"WorkSheetId": 501, "VersionId": 1, "VersionCode": "技术责任人确认版", "VersionStatus": "Y"}, {"WorkSheetId": 501, "VersionId": 2, "VersionCode": "甲方确认版", "VersionStatus": "Y"}, ] api = QuickTeamApi(fake) self.assertEqual(api.pending_version("0f52", 501).version_id, 2) def test_no_version_returns_none(self) -> None: api = QuickTeamApi(FakeTransport()) self.assertIsNone(api.pending_version("0f52", 501)) class VerifyTest(unittest.TestCase): def test_success_zero(self) -> None: fake = FakeTransport() fake.result = 0 api = QuickTeamApi(fake) api.verify_worksheet(1, 2, next_operator="CustomerLeader", next_version_code="客户责任人确认版", operator="陈首林", percent=20) _, connection, procedure, params = fake.calls[0] self.assertEqual(procedure, "QuickTeam.usp_VerifyWorkSheetById") sent = {name: value for name, value, _ in params} self.assertEqual(sent["NextOperator"], "CustomerLeader") self.assertEqual(sent["Percent"], 20) def test_nonzero_raises(self) -> None: fake = FakeTransport() fake.result = 1 api = QuickTeamApi(fake) with self.assertRaises(QtError): api.verify_worksheet(1, 2, next_operator="x", next_version_code="y", operator="o", percent=20) class TfsTest(unittest.TestCase): TEAMS = [ TfsTeam("DataTechnologies", "Tools_Dev\\DataTechnologies", [{"Name": "2026 Sprint 10", "Path": "Tools_Dev\\2026 Sprint\\2026 Sprint 10"}]), TfsTeam("FundHub", "Tools_Dev\\FundHub", [{"Name": "2026 Sprint 06", "Path": "Tools_Dev\\FundHub\\2026 Sprint 06"}]), ] def test_resolve_exact(self) -> None: area, iteration = QuickTeamApi.resolve_tfs_context(self.TEAMS, "DataTechnologies", "2026 Sprint 10") self.assertEqual(area, "Tools_Dev\\DataTechnologies") self.assertEqual(iteration, "Tools_Dev\\2026 Sprint\\2026 Sprint 10") def test_resolve_leading_zero(self) -> None: _, iteration = QuickTeamApi.resolve_tfs_context(self.TEAMS, "FundHub", "2026 Sprint 6") self.assertTrue(iteration.endswith("2026 Sprint 06")) def test_missing_team(self) -> None: with self.assertRaises(QtError): QuickTeamApi.resolve_tfs_context(self.TEAMS, "Nope", "2026 Sprint 10") def test_missing_iteration(self) -> None: with self.assertRaises(QtError) as ctx: QuickTeamApi.resolve_tfs_context(self.TEAMS, "DataTechnologies", "2026 Sprint 99") self.assertEqual(ctx.exception.category, Category.TFS) def test_push_payload(self) -> None: fake = FakeTransport() api = QuickTeamApi(fake) api.push_tfs([11, 22], "Tools_Dev", "Tools_Dev\\DataTechnologies", "Tools_Dev\\2026 Sprint\\2026 Sprint 10") _, operation, payload = fake.calls[0] self.assertEqual(operation, "BatchImportWorkItemsToTfs") inner = json.loads(urllib.parse.unquote(payload["param"])) self.assertEqual(inner["SPName"], "[QuickTeam].[usp_BatchImportWorkFeaturesByFeatureIds]") self.assertEqual(inner["SQLParams"][0]["Value"], "11,22") self.assertEqual(payload["areaPath"], "Tools_Dev\\DataTechnologies") def test_push_without_features(self) -> None: with self.assertRaises(QtError): QuickTeamApi(FakeTransport()).push_tfs([], "p", "a", "i") def test_push_returns_status(self) -> None: fake = FakeTransport() fake.tfs_result = 0 self.assertEqual(QuickTeamApi(fake).push_tfs([1, 2], "p", "a", "i"), 0) def test_push_negative_status_is_failure(self) -> None: """返回值是状态码：0 = 成功，-1 = 失败（实测：项目名不存在时返回 -1）。""" fake = FakeTransport() fake.tfs_result = -1 with self.assertRaises(QtError) as ctx: QuickTeamApi(fake).push_tfs([1], "p", "a", "i") self.assertEqual(ctx.exception.category, Category.TFS) class DryRunTest(unittest.TestCase): def test_no_server_calls(self) -> None: fake = FakeTransport() api = QuickTeamApi(fake, dry_run=True) self.assertEqual(api.create_worksheet(), 0) api.fill_worksheet( 0, PROJECT, system_name="s", start_time="a", end_time="b", evaluator="e", technical_leader="t", customer_leader="c", belong_team="bt", content="ct", operator="op", ) api.import_excel(0, 0, "p") api.verify_worksheet(0, 0, next_operator="x", next_version_code="y", operator="o", percent=20) api.push_tfs([1], "p", "a", "i") self.assertEqual(fake.calls, [], "dry-run 不允许发出任何服务端调用") if __name__ == "__main__": unittest.main()