#!/usr/bin/env python3 """U38 公开入口：真实 seata-server 2.5.0 发行版（store.mode=db 指向本机 MySQL） + seata-all 2.5.0 编程式 API（AT/XA/TC 恢复）+ seata-spring-boot-starter 2.5.0 （Boot 2.7.18 官方基线）上的 TCC fence 与 @GlobalTransactional 实验。 server 进程由测试 JVM（study.seata.Harness）启动与回收；本入口验证 Surefire 报告、 观察行、依赖版本（org.apache.seata:* 必须精确 2.5.0）、SCA 2025.1.0.0 依赖解析， 并验证 study_u38_* 专用库已定向清理、18091 端口已释放、无 seata-server 进程残留、 ~/logs/seata 无新写入。发行包缓存在仓库外 ~/.cache/software-engineering-study/。 """ import glob import hashlib import json import os import pathlib import platform import re import shutil import signal import socket import subprocess import sys import tempfile import time import xml.etree.ElementTree as ET import zipfile from datetime import datetime ROOT = pathlib.Path(__file__).resolve().parent EVIDENCE = ROOT / 'evidence/u38-2026-10-29.json' # 双模块：core（seata-all 编程式，无 Spring）/ starter（Boot 2.7.18 集成） MODULES = { 'core': { 'study.seata.SeataAtTest': [ 'atCommitRollbackUndoLog', 'atGlobalLockConflict', 'atTimeoutAutoRollback', 'atDirtyWriteUnretriable', ], 'study.seata.SeataXaTest': [ 'xaCommitRollbackIsolation', ], 'study.seata.SeataRecoveryTest': [ 'tcRestartRecoversPendingRollback', ], }, 'starter': { 'study.seata.SeataTccFenceTest': [ 'tccFenceProtocol', ], 'study.seata.SeataStarterTest': [ 'starterGlobalTransactional', ], }, } EXPECTED_TEST_COUNT = 8 JAVA_MARKERS = [ 'U38 atCommitRollback=PASS', 'U38 atGlobalLock=PASS', 'U38 atTimeoutRollback=PASS', 'U38 atDirtyWrite=PASS', 'U38 xaIsolation=PASS', 'U38 tcRestartRecovery=PASS', 'U38 tccFence=PASS', 'U38 starterBoot2=PASS', ] CLEANUP_MARKER = 'U38 cleanup=PASS' EXPECTED_VERSIONS = { 'org.apache.seata:seata-all': '2.5.0', 'org.apache.seata:seata-spring-boot-starter': '2.5.0', 'com.mysql:mysql-connector-j': '9.7.0', 'org.springframework.boot:spring-boot': '2.7.18', 'com.zaxxer:HikariCP': '4.0.3', } SCA_RESOLVE_EXPECTED = { 'com.alibaba.cloud:spring-cloud-starter-alibaba-seata': '2025.1.0.0', 'org.apache.seata:seata-spring-boot-starter': '2.5.0', 'org.apache.seata:seata-all': '2.5.0', } SEATA_GROUP = 'org.apache.seata:' SEATA_VERSION = '2.5.0' PINNED_SEATA_JARS = { 'target/seata-server.jar': '1eee5700a5651ca6bedb4b36a3c10287244488d6cb0dbb00c60c17fd026c945a', 'lib/seata-core-2.5.0.jar': '6000b0640a4c0a74614b5e2bd785e1a45a91793af98c93e9307ce05b8f1b1d40', 'lib/asm-6.0.jar': 'dd8971c74a4e697899a8e95caae4ea8760ea6c486dc6b97b1795e75760420461', } TC_PORT = 18091 SEATA_HOME_DEFAULT = pathlib.Path.home() / ( '.cache/software-engineering-study/java-depth-impl/u38/dist/extracted-bin' '/apache-seata-2.5.0-incubating-bin/seata-server') def now(): return datetime.now().astimezone().isoformat() def write_json(path, value): """Replace the evidence atomically; a failed run must not leave old success.""" path.parent.mkdir(parents=True, exist_ok=True) temporary = None try: with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent, prefix=path.name + '.', delete=False) as stream: temporary = pathlib.Path(stream.name) json.dump(value, stream, ensure_ascii=False, indent=2) stream.write('\n') temporary.replace(path) finally: if temporary is not None: temporary.unlink(missing_ok=True) def file_sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def validate_runtime_jars(seata_home, env): """核对本地固定文件及 JAR 元数据；不将其表述为整包官方来源认证。""" checked = {} for relative, expected in PINNED_SEATA_JARS.items(): path = seata_home / relative if not path.is_file() or file_sha(path) != expected: raise ValueError('Seata 实验固定 JAR 缺失或 SHA-256 不匹配: ' + str(path)) checked[relative] = expected with zipfile.ZipFile(seata_home / 'target/seata-server.jar') as jar: manifest = jar.read('META-INF/MANIFEST.MF').decode('utf-8') if not re.search(r'^Implementation-Version: 2[.]5[.]0\s*$', manifest, re.MULTILINE): raise ValueError('server JAR 元数据版本不为 2.5.0') with zipfile.ZipFile(seata_home / 'lib/seata-core-2.5.0.jar') as jar: properties = jar.read('META-INF/maven/org.apache.seata/seata-core/pom.properties').decode('utf-8') if not re.search(r'^version=2[.]5[.]0\s*$', properties, re.MULTILINE): raise ValueError('core JAR 元数据版本不为 2.5.0') default_driver = pathlib.Path.home() / '.m2/repository/com/mysql/mysql-connector-j/9.7.0/mysql-connector-j-9.7.0.jar' driver = pathlib.Path(env.get('U38_MYSQL_DRIVER') or str(default_driver)).expanduser().resolve() if not driver.is_file(): raise ValueError('缺少本地 MySQL 驱动，可使用 U38_MYSQL_DRIVER 指定: ' + str(driver)) with zipfile.ZipFile(driver) as jar: manifest = jar.read('META-INF/MANIFEST.MF').decode('utf-8') if ('com/mysql/cj/jdbc/Driver.class' not in jar.namelist() or not re.search(r'^Implementation-Title: MySQL Connector/J\s*$', manifest, re.MULTILINE) or not re.search(r'^Implementation-Version: 9[.]7[.]0\s*$', manifest, re.MULTILINE)): raise ValueError('MySQL 驱动版本不为 9.7.0: ' + str(driver)) env['U38_MYSQL_DRIVER'] = str(driver) return {'seataJarsSha256': checked, 'mysqlDriver': {'path': str(driver), 'sha256': file_sha(driver), 'version': '9.7.0'}, 'limits': '只固定并核对列出的三个 JAR 及其局部元数据，未验证其他依赖或完整发行包来源。'} def read_runtime_observations(observation_dir): if observation_dir is None or not observation_dir.is_dir(): return {'mysql': [], 'listeners': []} return { 'mysql': [{'file': p.name, 'version': p.read_text(encoding='utf-8').strip()} for p in sorted(observation_dir.glob('mysql-*.txt'))], 'listeners': [{'file': p.name, 'output': p.read_text(encoding='utf-8').strip()} for p in sorted(observation_dir.glob('listener-*.txt'))], 'workdirs': [{'file': p.name, 'path': p.read_text(encoding='utf-8').strip()} for p in sorted(observation_dir.glob('workdir-*.txt'))], } def artifact_digests(): paths = [ROOT / name for name in ('pom.xml', 'run.py', 'README.md', 'core/pom.xml', 'starter/pom.xml')] for module in MODULES: paths += sorted(path for path in (ROOT / module / 'src').rglob('*') if path.is_file()) paths += sorted(path for path in (ROOT / 'loopback').rglob('*') if path.is_file() and '__pycache__' not in path.parts) return {str(path.relative_to(ROOT)): file_sha(path) for path in paths} def maven_binary(): supplied = os.environ.get('MAVEN_BIN') if supplied: return shutil.which(supplied) or supplied found = shutil.which('mvn') if found: return found candidates = sorted(glob.glob(os.path.expanduser( '~/.m2/wrapper/dists/apache-maven-3.9.9/*/bin/mvn'))) if candidates: return candidates[-1] raise RuntimeError('需要 Maven 3.9.x：设置 MAVEN_BIN 或在 PATH 中提供 mvn') def interrupted(signum, _frame): raise SystemExit(128 + signum) def failure_code(exc): if isinstance(exc, subprocess.TimeoutExpired): return 124 if isinstance(exc, SystemExit) and isinstance(exc.code, int) and exc.code: return exc.code if isinstance(exc, KeyboardInterrupt): return 130 return 1 def run(name, args, logs, env, results, timeout=900): log_path = logs / (name + '.log') started = now() process = None try: # Maven 会再启动 Surefire JVM，测试 JVM 又会启动 Seata。独立进程组让超时或 # 中断时能回收整棵实验进程，而不是只杀掉 Maven 留下子进程。 process = subprocess.Popen(args, cwd=ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True) stdout, stderr = process.communicate(timeout=timeout) except BaseException as exc: if process is not None: try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: pass try: process.communicate(timeout=10) except subprocess.TimeoutExpired: pass finally: # 即使组长先退出，也要处理可能仍在运行的子进程。 try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass process.communicate() exit_code = failure_code(exc) result = {'name': name, 'command': args, 'cwd': str(ROOT), 'startedAt': started, 'finishedAt': now(), 'exitCode': None, 'runnerExitCode': exit_code, 'timedOut': isinstance(exc, subprocess.TimeoutExpired), 'failure': str(exc), 'log': str(log_path), 'logSha256': None, 'observed': []} results.append(result) write_json(logs / (name + '.status.json'), result) print(name + ' exit=' + str(exit_code) + ': ' + str(exc), file=sys.stderr, flush=True) raise SystemExit(exit_code) combined = (stdout or '') + (stderr or '') log_path.write_text(combined, encoding='utf-8') result = {'name': name, 'command': args, 'cwd': str(ROOT), 'startedAt': started, 'finishedAt': now(), 'exitCode': process.returncode, 'runnerExitCode': process.returncode, 'timedOut': False, 'failure': None, 'log': str(log_path), 'logSha256': file_sha(log_path), 'observed': [line for line in combined.splitlines() if re.match(r'^U38 [A-Za-z][A-Za-z0-9]*=', line)]} results.append(result) write_json(logs / (name + '.status.json'), result) if process.returncode: print(name + ' exit=' + str(process.returncode) + '; raw log: ' + str(log_path) + '\n' + combined[-4000:], file=sys.stderr, flush=True) raise SystemExit(process.returncode) print('DONE ' + name + ' exit=0; log: ' + str(log_path), flush=True) return combined def validate_reports(logs, not_before_ns): reports = [] for module, expected_classes in MODULES.items(): report_dir = ROOT / module / 'target/surefire-reports' paths = sorted(report_dir.glob('TEST-*.xml')) expected_files = sorted('TEST-' + name + '.xml' for name in expected_classes) if [path.name for path in paths] != expected_files: raise ValueError(module + ' Surefire 报告缺失或多出报告: ' + str([path.name for path in paths]) + '; expected=' + str(expected_files)) for path in paths: if path.stat().st_mtime_ns < not_before_ns: raise ValueError('Surefire 报告不是本次测试生成: ' + str(path)) shutil.copyfile(path, logs / (module + '-' + path.name)) suite = ET.parse(path).getroot() class_name = suite.get('name') if suite.tag != 'testsuite' or class_name not in expected_classes: raise ValueError(module + ' Surefire testsuite 名称不匹配: ' + str(class_name)) report = {key: suite.attrib[key] for key in ('name', 'tests', 'failures', 'errors', 'skipped')} for field in ('failures', 'errors', 'skipped'): if int(report[field]) != 0: raise ValueError(class_name + ' ' + field + '=' + report[field]) cases = suite.findall('testcase') names = [case.attrib['name'] for case in cases] expected = expected_classes[class_name] if (len(expected) != int(report['tests']) or len(cases) != len(expected) or len(set(names)) != len(names) or set(names) != set(expected)): raise ValueError(class_name + ' Surefire 测试名/数量不匹配: ' + str(names) + '; expected=' + str(sorted(expected))) for case in cases: if case.get('classname') != class_name or any( case.find(tag) is not None for tag in ('failure', 'error', 'skipped')): raise ValueError('Surefire 用例失败、跳过或类名不匹配: ' + case.attrib['name']) report['testNames'] = names report['sha256'] = file_sha(path) report['module'] = module reports.append(report) write_json(logs / 'reports.json', reports) return reports def validate_observations(observed): for marker in JAVA_MARKERS: matches = [line for line in observed if line == marker] if len(matches) != 1: raise ValueError('观察行必须出现一次: ' + marker + '; actual=' + str(len(matches))) if len(observed) != len(JAVA_MARKERS): raise ValueError('存在未登记或重复的 U38 观察行: ' + str(observed)) def port_open(port): try: with socket.create_connection(('127.0.0.1', port), 1): return True except OSError: return False def validate_cleanup(logs, observation_dir, not_before_ns): """两模块各有本次报告；独占目录登记、清理结果与真实文件状态一致。""" if observation_dir is None or not observation_dir.is_dir(): raise ValueError('缺少本次临时目录登记') workdirs = {} for record in observation_dir.glob('workdir-*.txt'): match = re.fullmatch(r'workdir-([0-9a-f]{32})[.]txt', record.name) if not match or record.stat().st_mtime_ns < not_before_ns: raise ValueError('临时目录登记无效或不是本次产物: ' + str(record)) path = pathlib.Path(record.read_text(encoding='utf-8').strip()) if not path.is_absolute() or not path.name.startswith('u38-seata-'): raise ValueError('临时目录登记路径无效: ' + str(path)) workdirs[match.group(1)] = path if len(workdirs) != len(MODULES): raise ValueError('须登记两个模块各自的独占目录: ' + str(workdirs)) details = [] seen_ids = set() for module in MODULES: report = ROOT / module / 'target/seata-cleanup.json' if not report.is_file() or report.stat().st_mtime_ns < not_before_ns: raise ValueError('缺少本次模块清理报告: ' + str(report)) result = json.loads(report.read_text(encoding='utf-8')) run_id = result.get('runId') if run_id not in workdirs or run_id in seen_ids: raise ValueError('清理报告无法与本次独占目录唯一匹配: ' + str(report)) seen_ids.add(run_id) if result.get('workRoot') != str(workdirs[run_id]): raise ValueError('清理报告和登记路径不一致: ' + str(report)) if result.get('databasesRemoved') is not True or result.get('workDirRemoved') is not True: raise ValueError('本次资源清理失败，保留记录供核查: ' + str(result)) try: os.lstat(workdirs[run_id]) except FileNotFoundError: pass except OSError as exc: raise ValueError('无法确认临时目录已删除: ' + str(workdirs[run_id])) from exc else: raise ValueError('本次临时目录仍存在: ' + str(workdirs[run_id])) saved = logs / (module + '-seata-cleanup.json') shutil.copyfile(report, saved) details.append({'module': module, 'runId': run_id, 'workRoot': str(workdirs[run_id]), 'report': str(saved), 'sha256': file_sha(saved)}) if seen_ids != set(workdirs): raise ValueError('存在未被报告的临时目录登记') if port_open(TC_PORT): raise ValueError('端口 ' + str(TC_PORT) + ' 未释放') processes = subprocess.run(['/usr/bin/pgrep', '-f', 'seata-server.jar'], capture_output=True, text=True) if processes.returncode == 0: raise ValueError('仍存在 seata-server 进程: ' + processes.stdout.strip()) return details def recover_owned_databases(java, ownership_dir, env): """异常退出后仅删除带本次所有权标记的数据库；模糊所有权一律保留并报错。""" if ownership_dir is None or not ownership_dir.is_dir() or not any(ownership_dir.iterdir()): return '无已登记数据库' if not java: raise RuntimeError('存在所有权记录但找不到 Java；保留记录供人工清理: ' + str(ownership_dir)) classes = ROOT / 'core/target/classes' driver = pathlib.Path(env.get('U38_MYSQL_DRIVER') or str(pathlib.Path.home() / '.m2/repository/com/mysql/mysql-connector-j/9.7.0/mysql-connector-j-9.7.0.jar')) if not (classes / 'study/seata/Harness.class').is_file() or not driver.is_file(): raise RuntimeError('定向清理缺少已编译 Harness 或 MySQL 驱动；保留记录: ' + str(ownership_dir)) command = [java, '-Xms16m', '-Xmx64m', '-XX:ActiveProcessorCount=1', '-Du38.cleanupOnly=true', '-cp', str(classes) + os.pathsep + str(driver), 'study.seata.Harness', str(ownership_dir)] result = subprocess.run(command, env=env, cwd=ROOT, capture_output=True, text=True, timeout=30) output = (result.stdout + result.stderr).strip() if result.returncode: raise RuntimeError('定向清理失败，保留所有权记录 ' + str(ownership_dir) + ': ' + output) return output or '已核验本次数据库不存在' def validate_home_logs(not_before_ns): """seata 客户端/server 日志必须重定向到实验目录，禁止写 ~/logs/seata。""" home_seata = pathlib.Path.home() / 'logs/seata' if not home_seata.is_dir(): return {'homeDirWrites': 0, 'files': []} offenders = [str(path) for path in home_seata.rglob('*') if path.is_file() and path.stat().st_mtime_ns >= not_before_ns] if offenders: raise ValueError('seata 日志泄漏到 ~/logs/seata（log.home 未生效）: ' + str(offenders)) return {'homeDirWrites': 0, 'files': []} def read_dependencies(path): versions = {} for line in path.read_text(encoding='utf-8').splitlines(): fields = line.strip().split(' ', 1)[0].split(':') if len(fields) not in (5, 6) or fields[2] not in ('jar', 'pom', 'test-jar'): continue if fields[-1] not in ('compile', 'provided', 'runtime', 'test', 'system'): continue coordinate = ':'.join(fields[:2]) version = fields[-2] if coordinate in versions and versions[coordinate] != version: raise ValueError('依赖版本冲突: ' + coordinate) versions[coordinate] = version for coordinate, expected in EXPECTED_VERSIONS.items(): if versions.get(coordinate) != expected: raise ValueError('依赖版本不匹配: ' + coordinate + '=' + str(versions.get(coordinate)) + '; expected=' + expected) seata_versions = {coordinate: version for coordinate, version in versions.items() if coordinate.startswith(SEATA_GROUP)} wrong = {coordinate: version for coordinate, version in seata_versions.items() if version != SEATA_VERSION} if wrong: raise ValueError('org.apache.seata 工件必须全部为 ' + SEATA_VERSION + '（防旧版冒充）: ' + str(wrong)) return versions def read_sca_dependencies(path): versions = {} for line in path.read_text(encoding='utf-8').splitlines(): fields = line.strip().split(' ', 1)[0].split(':') if len(fields) not in (5, 6) or fields[2] not in ('jar', 'pom'): continue coordinate = ':'.join(fields[:2]) versions[coordinate] = fields[-2] for coordinate, expected in SCA_RESOLVE_EXPECTED.items(): if versions.get(coordinate) != expected: raise ValueError('SCA 2025.1.0.0 依赖解析不匹配: ' + coordinate + '=' + str(versions.get(coordinate)) + '; expected=' + expected) seata_versions = {coordinate: version for coordinate, version in versions.items() if coordinate.startswith(SEATA_GROUP)} wrong = {coordinate: version for coordinate, version in seata_versions.items() if version != SEATA_VERSION} if wrong: raise ValueError('SCA 解析中的 seata 工件必须全部为 ' + SEATA_VERSION + ': ' + str(wrong)) return versions def validate_pom(): parent = ET.parse(ROOT / 'pom.xml').getroot() ns = {'m': 'http://maven.apache.org/POM/4.0.0'} expected = { 'm:parent/m:version': '2.7.18', 'm:properties/m:java.version': '21', 'm:properties/m:seata.version': '2.5.0', 'm:properties/m:mysql.version': '9.7.0', 'm:properties/m:spring-cloud-alibaba.version': '2025.1.0.0', } for query, value in expected.items(): if parent.findtext(query, namespaces=ns) != value: raise ValueError('父 POM 版本不匹配: ' + query + '; expected=' + value) modules = [node.text for node in parent.findall('m:modules/m:module', ns)] if modules != ['core', 'starter']: raise ValueError('模块清单不匹配: ' + str(modules)) core = ET.parse(ROOT / 'core/pom.xml').getroot() starter = ET.parse(ROOT / 'starter/pom.xml').getroot() for pom, artifact in ((core, 'seata-lab-core'), (starter, 'seata-lab-starter')): if pom.findtext('m:artifactId', namespaces=ns) != artifact: raise ValueError('模块 artifactId 不匹配: ' + artifact) if pom.findtext('m:parent/m:version', namespaces=ns) != 'u38': raise ValueError(artifact + ' 父版本不匹配') return {'spring-boot': '2.7.18', 'seata': '2.5.0', 'mysql-connector-j': '9.7.0', 'spring-cloud-alibaba': '2025.1.0.0', 'java': '21', 'modules': 'core(seata-all)+starter(spring-boot-starter)'} def main(): signal.signal(signal.SIGTERM, interrupted) signal.signal(signal.SIGINT, interrupted) logs = None ownership_dir = None observation_dir = None java = None env = dict(os.environ) cleanup_detail = '未执行' evidence = { 'ticket': 'U38 #75', 'status': 'running', 'startedAt': now(), 'finishedAt': None, 'input': 'python3 labs/seata/run.py', 'exitCode': None, 'environment': {'os': platform.platform(), 'arch': platform.machine(), 'python': platform.python_version()}, 'expectedTestCount': EXPECTED_TEST_COUNT, 'expectedTests': {name: sorted(tests) for module in MODULES.values() for name, tests in module.items()}, 'reports': [], 'testCount': 0, 'observed': [], 'results': [], 'seataHome': str(SEATA_HOME_DEFAULT), 'scaResolution': None, 'artifacts': {}, 'logs': None, 'cleanup': {'processGroups': '尚未完成', 'resources': 'study_u38_* 专用库由测试 JVM shutdown hook 删除；' 'seata-server 进程与临时目录一并回收；run.py 复核端口与进程'}, 'limits': [ '实验 agent 将 Seata 2.5.0 RPC 监听地址改为 127.0.0.1；事务实现未修改，原版默认通配监听', 'TC 为单机 seata-server 2.5.0 发行版进程（store.mode=db，端口 18091）；无注册中心、无多 TC 集群', 'SAGA 状态机引擎、Redis/raft store、namingserver、生产参数调优为推演，未运行', 'starter 用例运行在 Seata 官方构建基线 Boot 2.7.18；课程 Boot 4.0.9 与 Seata 2.5.0 的组合未运行', 'SCA spring-cloud-starter-alibaba-seata 2025.1.0.0 仅做依赖解析，不等于该组合实跑通过', 'MySQL 实际版本记录于 runtimeObservations；主从/断电级崩溃恢复（kill -9 仅进程级）与生产容量不在证明范围', ], } exit_code = 1 try: # Invalidate success even if cache setup fails or this run is killed later. write_json(EVIDENCE, evidence) logs_root = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u38' logs_root.mkdir(parents=True, exist_ok=True) logs = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=logs_root)) ownership_dir = logs / 'ownership' ownership_dir.mkdir(mode=0o700) observation_dir = logs / 'observations' observation_dir.mkdir(mode=0o700) evidence['logs'] = str(logs) print('LOGS ' + str(logs), flush=True) if len(sys.argv) != 1: raise ValueError('本入口不接受参数；使用 python3 labs/seata/run.py') seata_home = pathlib.Path(os.environ.get('U38_SEATA_SERVER_DIR', str(SEATA_HOME_DEFAULT))) evidence['seataHome'] = str(seata_home) if not seata_home.is_dir() or not (seata_home / 'target/seata-server.jar').is_file(): raise ValueError('缺少 seata-server 2.5.0 发行包: ' + str(seata_home)) evidence['runtimeJars'] = validate_runtime_jars(seata_home, env) if len(JAVA_MARKERS) != EXPECTED_TEST_COUNT or sum( len(tests) for module in MODULES.values() for tests in module.values()) != EXPECTED_TEST_COUNT: raise ValueError('U38 测试/观察行清单尚未完整登记，必须有 ' + str(EXPECTED_TEST_COUNT) + ' 项') evidence['artifacts'] = artifact_digests() evidence['environment']['declaredVersions'] = validate_pom() env['U38_OWNERSHIP_DIR'] = str(ownership_dir) env['U38_OBSERVATION_DIR'] = str(observation_dir) if not env.get('JAVA_HOME') and sys.platform == 'darwin': java_home = run('java-home', ['/usr/libexec/java_home', '-v', '21'], logs, env, evidence['results'], timeout=15).strip() env['JAVA_HOME'] = java_home if env.get('JAVA_HOME'): env['PATH'] = str(pathlib.Path(env['JAVA_HOME']) / 'bin') + os.pathsep + env.get('PATH', '') java = shutil.which('java', path=env.get('PATH')) if not java: raise RuntimeError('需要 JDK 21：设置 JAVA_HOME 或在 PATH 中提供 java') version = run('jdk', [java, '-version'], logs, env, evidence['results'], timeout=15) evidence['environment']['jdk'] = version.strip() if not re.search(r'version "21[.\"]', version): raise ValueError('需要 JDK 21；实际版本见 jdk.log') mvn = maven_binary() maven = run('maven', [mvn, '-version'], logs, env, evidence['results'], timeout=30) evidence['environment']['maven'] = maven.strip() if not re.search(r'Apache Maven 3[.]9[.]', maven): raise ValueError('需要 Maven 3.9.x；实际版本见 maven.log') # 不自动下载依赖；缺失时先由用户授权下载，再单独准备本地缓存。 base = [mvn, '-o', '-B', '-ntp', '-q', '-f', str(ROOT / 'pom.xml')] test_started_ns = time.time_ns() run('test', base + ['clean', 'test', '-DskipTests=false', '-Dmaven.test.skip=false', '-Dmaven.clean.skip=false', '-Dmaven.test.failure.ignore=false', '-DfailIfNoTests=true'], logs, env, evidence['results'], timeout=1200) evidence['reports'] = validate_reports(logs, test_started_ns) runtime_observations = read_runtime_observations(observation_dir) if len(runtime_observations['mysql']) < 2 or len(runtime_observations['listeners']) < 3: raise ValueError('缺少两模块 MySQL 版本或 TC 启动/重启的回环监听观察') if len({item['version'] for item in runtime_observations['mysql']}) != 1: raise ValueError('两个模块观察到不同的 MySQL 版本') evidence['runtimeObservations'] = runtime_observations evidence['observed'] = evidence['results'][-1]['observed'] validate_observations(evidence['observed']) evidence['homeSeataLogs'] = validate_home_logs(test_started_ns) cleanup_reports = validate_cleanup(logs, observation_dir, test_started_ns) evidence['cleanup']['reports'] = cleanup_reports cleanup_detail = '两个模块的数据库与独占临时目录已核验清理' evidence['cleanup']['detail'] = cleanup_detail dependencies = logs / 'dependencies.txt' combined = [] for module in MODULES: module_file = logs / (module + '-dependencies.txt') run('dependencies-' + module, base + ['-pl', module, '-am', 'test-compile', 'dependency:list', '-DskipTests', '-DincludeScope=test', '-DappendOutput=false', '-DoutputFile=' + str(module_file)], logs, env, evidence['results']) combined.append(module_file.read_text(encoding='utf-8')) dependencies.write_text('\n'.join(combined), encoding='utf-8') evidence['environment']['dependencies'] = read_dependencies(dependencies) evidence['dependencyReport'] = {'path': str(dependencies), 'sha256': file_sha(dependencies)} sca_dependencies = logs / 'sca-dependencies.txt' sca_module_file = logs / ('starter-sca-dependencies.txt') run('sca-resolve', base + ['-pl', 'starter', '-am', '-Psca', 'test-compile', 'dependency:list', '-DskipTests', '-DincludeScope=test', '-DappendOutput=false', '-DoutputFile=' + str(sca_module_file)], logs, env, evidence['results']) sca_dependencies.write_text(sca_module_file.read_text(encoding='utf-8'), encoding='utf-8') evidence['scaResolution'] = { 'bom': 'com.alibaba.cloud:spring-cloud-alibaba-dependencies:2025.1.0.0', 'mode': 'dependency:list（仅解析，未运行）', 'verified': read_sca_dependencies(sca_dependencies), 'report': {'path': str(sca_dependencies), 'sha256': file_sha(sca_dependencies)}, } if artifact_digests() != evidence['artifacts']: raise ValueError('运行期间实验文件发生变化；请在修改结束后重新运行') if validate_runtime_jars(seata_home, env) != evidence['runtimeJars']: raise ValueError('运行期间固定 JAR 或驱动发生变化；请重新运行') evidence['testCount'] = sum(int(report['tests']) for report in evidence['reports']) if evidence['testCount'] != EXPECTED_TEST_COUNT: raise ValueError('测试总数不为 ' + str(EXPECTED_TEST_COUNT) + ': ' + str(evidence['testCount'])) evidence['status'] = 'passed' exit_code = 0 except BaseException as exc: exit_code = failure_code(exc) evidence['status'] = 'failed' evidence['failure'] = {'type': type(exc).__name__, 'message': str(exc)} print('FAILED exit=' + str(exit_code) + ': ' + str(exc) + '; logs: ' + str(logs), file=sys.stderr, flush=True) finally: handlers = {sig: signal.signal(sig, signal.SIG_IGN) for sig in (signal.SIGINT, signal.SIGTERM)} try: try: recovery_detail = recover_owned_databases(java, ownership_dir, env) except Exception as exc: recovery_detail = str(exc) evidence['status'] = 'failed' evidence['cleanup']['recoveryFailure'] = recovery_detail if exit_code == 0: exit_code = 1 print('FAILED cleanup: ' + recovery_detail, file=sys.stderr, flush=True) evidence['cleanup']['recovery'] = recovery_detail try: evidence['runtimeObservations'] = read_runtime_observations(observation_dir) except Exception as exc: evidence['status'] = 'failed' evidence['observationFailure'] = str(exc) if exit_code == 0: exit_code = 1 evidence['exitCode'] = exit_code evidence['finishedAt'] = now() evidence['observed'] = [line for result in evidence['results'] if result['name'] == 'test' for line in result['observed']] if evidence['status'] == 'passed': evidence['observed'].append(CLEANUP_MARKER) evidence['cleanup']['detail'] = cleanup_detail if not evidence['results']: evidence['cleanup']['processGroups'] = '未启动命令进程' else: evidence['cleanup']['processGroups'] = 'seata-server 由测试 JVM 直接管理并回收；mvn 进程由本入口同步等待' if logs is not None: write_json(logs / 'run.status.json', evidence) write_json(EVIDENCE, evidence) finally: for sig, handler in handlers.items(): signal.signal(sig, handler) if exit_code == 0: print('PASS U38 tests=' + str(evidence['testCount']) + ' cleanup=' + cleanup_detail + '; evidence: ' + str(EVIDENCE), flush=True) return exit_code if __name__ == '__main__': sys.exit(main())