#!/usr/bin/env python3 """U88 公开入口：真实嵌入式 HiveMQ CE 2026.5 broker 上的设备身份、会话与 QoS 实验。 实验不使用任何外部服务：每个测试在 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.DeviceIdentityTest': 6, 'study.iot.MqttSessionTest': 9, } MARKERS = [ 'U88 identity=PASS', 'U88 identityReject=PASS', 'U88 anonymousReject=PASS', 'U88 aclPublish=PASS', 'U88 will=PASS', 'U88 embeddedDefaultAllowAll=PASS', 'U88 session=PASS', 'U88 cleanStart=PASS', 'U88 qosOffline=PASS', 'U88 qos2=PASS', 'U88 dupRedelivery=PASS', 'U88 retained=PASS', 'U88 retainedRestart=PASS', 'U88 qosDowngrade=PASS', 'U88 takeover=PASS', ] DIGEST_INCLUDE = { 'hivemq-community-edition-embedded', 'hivemq-extension-sdk', 'hivemq-mqtt-client', 'netty-codec-mqtt', 'netty-handler', } 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/u88' 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('U88 ')]} 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-device', [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'] 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': 'U88 #125', 'startedAt': started, 'finishedAt': datetime.now().astimezone().isoformat(), 'input': 'python3 labs/iot-device/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）', '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': [ '单机嵌入式 CE broker；无集群、桥接、共享订阅跨节点与生产容量压测', 'TLS/mTLS、WebSocket 监听器、客户端证书与外部凭据提供方（OAuth/JWT/数据库认证）未运行', '权限模型是教学扩展：用户名/密码与设备 clientId 绑定，不是生产设备证书或动态注册流程', 'broker 重启是正常 stop 后重开，不是 kill -9、断电或文件系统损坏恢复', 'QoS 2 已验证协议握手与单次投递，但未覆盖跨 broker 集群去重与业务副作用 exactly-once', ], } out = ROOT / 'evidence' / ('u88-' + 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)