#!/usr/bin/env python3 """U89 公开入口：真实嵌入式 HiveMQ CE 2026.5 broker 上的设备影子、命令状态与跨语言契约实验。 实验不使用任何外部服务：每个测试在 127.0.0.1 的随机端口上启动一个真实的 HiveMQ Community Edition broker（含 Netty TCP 监听器与文件持久化目录）， 用 HiveMQ MQTT Client 1.4.1 与原始 MQTT 5 报文探针建立真实 TCP 连接。 入口只运行本仓库测试，结束时回收自己创建的进程组。 """ import glob import hashlib import json import os import pathlib import platform import re import signal import subprocess import sys import tempfile import time import xml.etree.ElementTree as ET from datetime import datetime ROOT = pathlib.Path(__file__).resolve().parent EXPECTED = { 'study.iot.command.CommandShadowTest': 13, 'study.iot.command.EndToEndCommandTest': 2, } MARKERS = [ 'U89 lifecycle=PASS', 'U89 duplicate=PASS', 'U89 duplicateInFlight=PASS', 'U89 lateAck=PASS', 'U89 expired=PASS', 'U89 shadowConflict=PASS', 'U89 shadowJson=PASS', 'U89 shadowDelete=PASS', 'U89 shadowMerge=PASS', 'U89 offlineQueue=PASS', 'U89 crossLanguage=PASS', 'U89 pythonAck=PASS', 'U89 malformedRejected=PASS', 'U89 e2e=PASS', 'U89 e2eOffline=PASS', ] DIGEST_INCLUDE = { 'hivemq-community-edition-embedded', 'hivemq-extension-sdk', 'hivemq-mqtt-client', 'netty-codec-mqtt', 'netty-handler', 'jackson-databind', 'jackson-core', 'jackson-annotations', } def maven_binary(): found = os.environ.get('MAVEN_BIN') or subprocess.run( ['/usr/bin/which', 'mvn'], capture_output=True, text=True).stdout.strip() 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 SystemExit('需要 Maven 3.9.9：设置 MAVEN_BIN 或在 PATH 中提供 mvn') def interrupted(signum, _frame): raise SystemExit(128 + signum) signal.signal(signal.SIGTERM, interrupted) signal.signal(signal.SIGINT, interrupted) jdk = pathlib.Path(os.environ.get('JAVA_HOME') or subprocess.check_output( ['/usr/libexec/java_home', '-v', '21'], text=True).strip()) env = dict(os.environ, JAVA_HOME=str(jdk), PATH=str(jdk / 'bin') + os.pathsep + os.environ.get('PATH', '')) logs_root = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u89' logs_root.mkdir(parents=True, exist_ok=True) logs = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=logs_root)) started = datetime.now().astimezone().isoformat() results = [] mvn = maven_binary() def live_group_pids(pgid): snapshot = subprocess.run(['/bin/ps', '-axo', 'pid=,pgid=,stat='], capture_output=True, text=True, check=True, timeout=5) return [int(pid) for pid, group, state in (line.split() for line in snapshot.stdout.splitlines()) if int(group) == pgid and not state.startswith('Z')] def cleanup_group(process): for signum, grace in ((signal.SIGTERM, 2), (signal.SIGKILL, 5)): process.poll() remaining = live_group_pids(process.pid) if not remaining: return try: os.killpg(process.pid, signum) except ProcessLookupError: return deadline = time.monotonic() + grace while time.monotonic() < deadline: process.poll() if not live_group_pids(process.pid): return time.sleep(0.05) remaining = live_group_pids(process.pid) if remaining: raise RuntimeError('进程组仍有存活进程: ' + str(remaining)) def run(name, args, timeout=900, cwd=ROOT): log_path = logs / (name + '.log') process = None failure = None cleanup_error = None exit_code = 0 try: with log_path.open('w') as stream: try: process = subprocess.Popen(args, cwd=cwd, env=env, text=True, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True) return_code = process.wait(timeout=timeout) exit_code = return_code if return_code >= 0 else 128 - return_code except BaseException as exc: failure = exc if isinstance(exc, subprocess.TimeoutExpired): exit_code = 124 elif isinstance(exc, SystemExit) and isinstance(exc.code, int): exit_code = exc.code elif isinstance(exc, KeyboardInterrupt): exit_code = 130 else: exit_code = 1 finally: handlers = {sig: signal.signal(sig, signal.SIG_IGN) for sig in (signal.SIGINT, signal.SIGTERM)} try: if process is not None: cleanup_group(process) except Exception as exc: cleanup_error = str(exc) exit_code = exit_code or 1 finally: for sig, handler in handlers.items(): signal.signal(sig, handler) except OSError as exc: failure = exc exit_code = 1 output = log_path.read_text(errors='replace') if log_path.exists() else '' result = {'name': name, 'command': args, 'exitCode': process.returncode if process is not None else None, 'runnerExitCode': exit_code, 'timedOut': isinstance(failure, subprocess.TimeoutExpired), 'failure': str(failure) if failure is not None else None, 'cleanupError': cleanup_error, 'observed': [line for line in output.splitlines() if line.startswith('U89 ')]} results.append(result) (logs / (name + '.status.json')).write_text( json.dumps(result, ensure_ascii=False, indent=1) + '\n') if exit_code: print(name + ' exit=' + str(exit_code) + '; raw log: ' + str(log_path) + ('; cleanup: ' + cleanup_error if cleanup_error else '') + '\n' + output[-3500:], file=sys.stderr, flush=True) raise SystemExit(exit_code) print('PASS ' + name, flush=True) return output def validate_reports(reports): by_class = {} for report in reports: name = report['name'] if name in by_class: raise ValueError('重复 Surefire 报告: ' + name) for field in ('failures', 'errors', 'skipped'): if int(report[field]) != 0: raise ValueError(name + ' ' + field + '=' + report[field]) by_class[name] = int(report['tests']) if by_class != EXPECTED: raise ValueError('Surefire 计数不匹配: ' + str(by_class) + '; expected=' + str(EXPECTED)) test_count = sum(by_class.values()) if test_count != 15: raise ValueError('测试总数应为 15: ' + str(test_count)) return test_count version = run('jdk', [str(jdk / 'bin/java'), '-version']) assert re.search(r'version "21[.]', version), version maven_version = run('maven', [mvn, '-version']) run('test-iot-command', [mvn, '-B', '-ntp', '-q', 'clean', 'test']) reports = [] for path in sorted((ROOT / 'target/surefire-reports').glob('TEST-*.xml')): root = ET.parse(path).getroot() reports.append({key: root.attrib[key] for key in ['name', 'tests', 'failures', 'errors', 'skipped']}) (logs / 'reports.json').write_text(json.dumps(reports, ensure_ascii=False, indent=1) + '\n') try: test_count = validate_reports(reports) except (KeyError, ValueError) as exc: (logs / 'validation-failure.json').write_text( json.dumps({'failure': str(exc), 'exitCode': 1}, ensure_ascii=False) + '\n') raise SystemExit('Surefire 验证失败: ' + str(exc) + '; reports: ' + str(logs / 'reports.json')) def dependencies(): tree = run('dependency', [mvn, '-B', '-ntp', '-q', 'dependency:list', '-DoutputFile=/dev/stdout']) versions = {} for line in tree.splitlines(): match = re.match(r'\s*[\w.\-]+:([\w.\-]+):(?:jar|pom):([\w.\-]+):', line) if match and match.group(1) in DIGEST_INCLUDE: versions[match.group(1)] = match.group(2) return versions dependency_versions = dependencies() observed = '\n'.join(line for result in results for line in result['observed']) for marker in MARKERS: assert marker in observed, '缺少观察行 ' + marker tracked = [ROOT / 'pom.xml', ROOT / 'run.py', ROOT / 'README.md', ROOT / 'device_client.py'] tracked += sorted(ROOT.glob('src/**/*.java')) tracked += sorted(ROOT.glob('src/**/*.xml')) digests = {} for path in tracked: if path.exists(): digests[str(path.relative_to(ROOT))] = hashlib.sha256(path.read_bytes()).hexdigest() evidence = { 'ticket': 'U89 #126', 'startedAt': started, 'finishedAt': datetime.now().astimezone().isoformat(), 'input': 'python3 labs/iot-command/run.py', 'exitCode': 0, 'environment': { 'jdk': version.strip(), 'maven': maven_version.strip(), 'os': platform.platform(), 'arch': platform.machine(), 'python': platform.python_version(), 'broker': 'HiveMQ Community Edition 2026.5（嵌入式，真实 Netty MQTT 监听器与文件持久化）', 'client': 'HiveMQ MQTT Client 1.4.1（MQTT 5）与 device_client.py（Python 标准库独立实现）', 'dependencies': dependency_versions, }, 'reports': reports, 'testCount': test_count, 'results': results, 'artifacts': digests, 'logs': str(logs), 'cleanup': { 'processes': '本入口创建的进程组已回收；每个用例的 EmbeddedHiveMQ 在 finally 中 stop+close，临时数据目录删除', 'externalServices': '无外部服务；broker 监听 127.0.0.1 随机端口，仅使用临时目录与合成数据', }, 'limits': [ '影子与命令状态机是单进程 Java 教学实现，效果写本地文件；未接入生产数据库事务、多节点命令服务或真实云 IoT 平台影子', '跨语言客户端为 Python 标准库实现，只覆盖契约的编解码与摘要一致性，不含固件 C/C++ 或 C# 实现', 'TLS/mTLS、增强认证、WebSocket、外部凭据源与 broker 集群/桥接未运行', '时间过期用可控 Clock 制造，不是真实等待数分钟；断线恢复是正常 disconnect 后重连，不是 kill -9/断电', '未做容量与并发压测；两路并发更新影子按顺序模拟，不是真实多线程争用', ], } out = ROOT / 'evidence' / ('u89-' + datetime.now().strftime('%Y-%m-%d') + '.json') out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(evidence, ensure_ascii=False, indent=1) + '\n') print('WROTE ' + str(out.relative_to(ROOT)) + ' tests=' + str(test_count), flush=True)