205 lines
7.1 KiB
Python
205 lines
7.1 KiB
Python
"""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<RowVo> rows) {
|
|
List<ItemVo> items = rows.stream()
|
|
.map(row -> toVo(row))
|
|
.collect(Collectors.toList());
|
|
int updated = fileResultMapper.update(null, new LambdaUpdateWrapper<FileResultEntity>()
|
|
.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<FileTaskEntity>()
|
|
.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<TaskFileEntity>()
|
|
.eq(TaskFileEntity::getId, taskId));
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
FIXTURE_PLAIN = """package com.example;
|
|
@Service
|
|
public class DemoService {
|
|
public int sum(List<Integer> 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()
|