"""task-124 @Transactional 方法审计清单测试。 对应 plan 07 任务 124 的 8 条用例 + 2 条补强： 1. test_audit_covers_all_modules 覆盖全部业务模块 2. test_audit_classification 分类完整（必须/可移出） 3. test_audit_result_endpoints /result 相关方法都在清单 4. test_audit_python_facing Python 回调相关标记 5. test_audit_priority_order 按收益排序 6. test_audit_no_code_change 本任务零代码变更 7. test_audit_documented 文档产出 8. test_audit_repeatable 可重复执行 9. test_audit_detects_tx_method 识别 @Transactional 方法 10. test_audit_classifies_movable 事务内可移出段分类 """ from __future__ import annotations import json import subprocess import unittest from pathlib import Path from tx_audit import MODULES_ROOT, scan_modules, scan_text SCRIPTS_DIR = Path(__file__).resolve().parent REPO_ROOT = SCRIPTS_DIR.parent.parent DOC_PATH = REPO_ROOT / "backend-java" / "docs" / "transaction-boundary-audit.md" FIXTURE_TX = """package com.example; @Service public class DemoService { @Transactional public boolean submitResult(Long taskId, List rows) { List items = rows.stream() .map(row -> toVo(row)) .collect(Collectors.toList()); int updated = fileResultMapper.update(null, new LambdaUpdateWrapper() .eq(FileResultEntity::getId, taskId) .set(FileResultEntity::getStatus, "SUCCESS")); log.info("submitted taskId={} count={}", taskId, items.size()); Files.deleteIfExists(Path.of("/tmp/task-" + taskId)); return updated > 0; } public void noTxMethod(Long id) { fileTaskMapper.selectById(id); } } """ FIXTURE_WRITE_ONLY = """package com.example; @Service public class DemoService { @Transactional public int markFailed(Long id) { return fileTaskMapper.update(null, new LambdaUpdateWrapper() .eq(FileTaskEntity::getId, id) .set(FileTaskEntity::getStatus, "FAILED")); } } """ FIXTURE_LOCK = """package com.example; @Service public class DemoService { @Transactional public void finalizeTask(Long taskId) { TaskDistributedLockService.LockHandle handle = acquireTaskLock("MODULE", taskId); if (handle == null) { return; } try (handle) { taskMapper.update(null, new LambdaUpdateWrapper() .eq(TaskFileEntity::getId, taskId)); } } } """ FIXTURE_PLAIN = """package com.example; @Service public class DemoService { public int sum(List nums) { return nums.stream().mapToInt(Integer::intValue).sum(); } } """ class TxAuditDetectTest(unittest.TestCase): def test_audit_detects_tx_method(self): findings = scan_text(FIXTURE_TX, "DemoService.java") self.assertEqual(len(findings), 1) self.assertEqual(findings[0]["method"], "submitResult") self.assertFalse(findings[0]["class_level"]) def test_audit_classifies_movable(self): findings = scan_text(FIXTURE_TX, "DemoService.java") self.assertEqual(len(findings), 1) movable = findings[0]["movable"] self.assertIn("compute", movable) self.assertIn("assembly", movable) self.assertIn("log", movable) self.assertIn("cleanup", movable) self.assertIn("write", findings[0]["must_stay"]) self.assertGreater(findings[0]["priority"], 0) def test_audit_no_tx_not_reported(self): self.assertEqual(len(scan_text(FIXTURE_PLAIN, "Plain.java")), 0) def test_audit_write_only_no_movable(self): findings = scan_text(FIXTURE_WRITE_ONLY, "DemoService.java") self.assertEqual(len(findings), 1) self.assertIn("write", findings[0]["must_stay"]) self.assertEqual(findings[0]["movable"], []) def test_audit_lock_flagged_must_stay(self): findings = scan_text(FIXTURE_LOCK, "DemoService.java") self.assertEqual(len(findings), 1) self.assertIn("lock", findings[0]["must_stay"]) class TxAuditCoverageTest(unittest.TestCase): def test_audit_covers_all_modules(self): report = scan_modules(MODULES_ROOT) self.assertGreaterEqual(report.scanned_files, 600) self.assertGreater(len(report.findings), 50) modules = {f["module"] for f in report.findings} self.assertGreaterEqual(len(modules), 15) def test_audit_classification(self): for finding in scan_modules(MODULES_ROOT).findings: self.assertIn("must_stay", finding) self.assertIn("movable", finding) self.assertIsInstance(finding["must_stay"], list) self.assertIsInstance(finding["movable"], list) def test_audit_result_endpoints(self): report = scan_modules(MODULES_ROOT) result_methods = [ f for f in report.findings if "result" in f["method"].lower() or "submit" in f["method"].lower() ] self.assertGreater( len(result_methods), 0, "应找到 /result 相关 @Transactional 方法" ) modules_with_result = {f["module"] for f in result_methods} self.assertGreaterEqual(len(modules_with_result), 3) def test_audit_python_facing(self): report = scan_modules(MODULES_ROOT) flagged = [f for f in report.findings if f["python_facing"]] self.assertGreater(len(flagged), 0, "应标记 Python 回调相关方法") for f in flagged: name = f["method"].lower() self.assertTrue( any( k in name for k in ( "result", "submit", "upload", "chunk", "done", "ack", "report", ) ), f"python_facing 标记与命名不符: {f['method']}", ) def test_audit_priority_order(self): findings = scan_modules(MODULES_ROOT).findings priorities = [f["priority"] for f in findings] self.assertEqual(priorities, sorted(priorities, reverse=True)) def test_audit_repeatable(self): first = scan_modules(MODULES_ROOT).to_dict() second = scan_modules(MODULES_ROOT).to_dict() self.assertEqual( json.dumps(first, sort_keys=True), json.dumps(second, sort_keys=True) ) def test_audit_documented(self): self.assertTrue(DOC_PATH.is_file(), f"audit doc missing: {DOC_PATH}") text = DOC_PATH.read_text(encoding="utf-8") self.assertIn("事务", text) self.assertIn("必须", text) self.assertIn("可移出", text) def test_audit_no_code_change(self): changed = subprocess.run( ["git", "diff", "--name-only", "HEAD", "--", "backend-java/src"], cwd=REPO_ROOT, capture_output=True, text=True, check=True, ).stdout.strip() self.assertEqual(changed, "", f"生产代码被改动: {changed}") if __name__ == "__main__": unittest.main()