#!/usr/bin/env python3 """U58 公开入口：真实 RabbitMQ 4.4.0 broker 上的 exchange/路由、confirm/return、ack/prefetch、重投与死信实验。 broker 是本机独立 node（NODENAME=u58@localhost，端口 5679，mnesia/log 在 /tmp/u58-rabbit）， 由本脚本启动并在结束时停止与删除；不触碰用户其他 RabbitMQ 实例。RabbitMQ 发行包与 自带插件位于仓库外缓存 ~/.cache/software-engineering-study/tools/rabbitmq_server-4.4.0。 """ 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 from datetime import datetime ROOT = pathlib.Path(__file__).resolve().parent TOOLS = pathlib.Path.home() / '.cache/software-engineering-study/tools' RMQ = TOOLS / 'rabbitmq_server-4.4.0' JDK = pathlib.Path(os.environ.get('JAVA_HOME') or subprocess.check_output( ['/usr/libexec/java_home', '-v', '21'], text=True).strip()) PORT = 5679 NODENAME = 'u58@localhost' _LOG_ROOT = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u58' _LOG_ROOT.mkdir(parents=True, exist_ok=True) BASE = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=_LOG_ROOT)) ENV = dict(os.environ, JAVA_HOME=str(JDK), PATH='/opt/homebrew/bin:' + os.environ.get('PATH', ''), RABBITMQ_NODENAME=NODENAME, RABBITMQ_NODE_PORT=str(PORT), RABBITMQ_DIST_PORT='25679', RABBITMQ_MNESIA_BASE=str(BASE / 'mnesia'), RABBITMQ_LOG_BASE=str(BASE / 'log')) results = [] OBSERVED = [] started = datetime.now().astimezone().isoformat() def mark(line): OBSERVED.append(line) print(line, flush=True) def run(name, args, cwd=ROOT, timeout=600, expect=0, env=None): log = BASE / (name + '.log') try: proc = subprocess.run(args, cwd=cwd, env=env or ENV, text=True, capture_output=True, timeout=timeout) except subprocess.TimeoutExpired: raise SystemExit(name + ' 超时') combined = (proc.stdout or '') + (proc.stderr or '') log.write_text(combined) ok = proc.returncode == expect results.append({'name': name, 'command': args, 'exitCode': proc.returncode, 'expected': expect, 'ok': ok, 'observed': [l for l in combined.splitlines() if l.startswith('U58')]}) print(('PASS ' if ok else 'FAIL ') + name + ' exit=' + str(proc.returncode), flush=True) if not ok: print(combined[-3000:], file=sys.stderr) raise SystemExit(1) return combined def port_open(): try: with socket.create_connection(('127.0.0.1', PORT), 1): return True except OSError: return False def start_broker(): if not RMQ.exists(): print('U58 broker=BLOCKED rabbitmq-not-cached', flush=True) raise SystemExit(2) if port_open(): raise SystemExit(f'端口 {PORT} 已被占用：拒绝启动，避免影响非本实验的 RabbitMQ') run('broker-start', [str(RMQ / 'sbin/rabbitmq-server'), '-detached'], timeout=300) deadline = time.time() + 90 while time.time() < deadline: if port_open(): mark('U58 brokerStart=PASS port=' + str(PORT) + ' node=' + NODENAME) return time.sleep(1) raise SystemExit('RabbitMQ 未在 90 秒内监听 ' + str(PORT)) def stop_broker(): if port_open(): subprocess.run([str(RMQ / 'sbin/rabbitmqctl'), '-n', NODENAME, 'stop'], env=ENV, capture_output=True, text=True, timeout=120) deadline = time.time() + 60 while time.time() < deadline and port_open(): time.sleep(1) mark('U58 brokerStop=PASS stopped=' + str(not port_open())) def version_of(): out = subprocess.run([str(RMQ / 'sbin/rabbitmqctl'), '-n', NODENAME, 'status'], env=ENV, capture_output=True, text=True, timeout=120).stdout for line in out.splitlines(): if line.strip().startswith('RabbitMQ version'): return line.split(':', 1)[1].strip() return 'unknown' def mvn(): return shutil.which('mvn') or '/Users/gaochao/develop/maven-3.9.9/bin/mvn' try: start_broker() version = version_of() run('mvn-test', [mvn(), '-B', '-ntp', '-q', '-Dlab.amqp.port=' + str(PORT), 'clean', 'test'], timeout=900) 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']}) assert len(reports) == 1 and reports[0]['tests'] == '11', reports assert reports[0]['failures'] == '0' and reports[0]['errors'] == '0' mark('U58 tests=PASS count=11 failures=0 errors=0') deps = run('dependency', [mvn(), '-B', '-ntp', '-q', 'dependency:list', '-DoutputFile=/dev/stdout']) versions = {} for line in deps.splitlines(): m = re.match(r'\s*[\w.\-]+:([\w.\-]+):jar:([\w.\-]+):', line) if m and m.group(1) in ('amqp-client',): versions[m.group(1)] = m.group(2) # 测试输出在 mvn 的 stdout 里，需要先并入 OBSERVED for line in results[-2]['observed']: OBSERVED.append(line) missing = [m for m in ['U58 direct=PASS', 'U58 fanout=PASS', 'U58 topic=PASS', 'U58 mandatoryReturn=PASS', 'U58 publisherConfirm=PASS', 'U58 nackRequeue=PASS', 'U58 prefetch=PASS', 'U58 deadLetter=PASS', 'U58 unackedRecovery=PASS', 'U58 alternateExchange=PASS', 'U58 quorumQueue=PASS'] if m not in '\n'.join(OBSERVED)] assert not missing, missing finally: stop_broker() tracked = [ROOT / 'pom.xml', ROOT / 'run.py', ROOT / 'README.md'] tracked += sorted(ROOT.glob('src/**/*.java')) digests = {} for p in tracked: if p.exists(): digests[str(p.relative_to(ROOT))] = hashlib.sha256(p.read_bytes()).hexdigest() evidence = { 'ticket': 'U58 #95', 'startedAt': started, 'finishedAt': datetime.now().astimezone().isoformat(), 'input': 'python3 labs/rabbitmq-topology/run.py', 'exitCode': 0, 'environment': { 'jdk': subprocess.run([str(JDK / 'bin/java'), '-version'], capture_output=True, text=True, env=ENV).stderr.strip(), 'maven': subprocess.run([mvn(), '-v'], capture_output=True, text=True, env=ENV).stdout.splitlines()[0], 'broker': 'RabbitMQ ' + version + '（本机独立 node ' + NODENAME + '，端口 ' + str(PORT) + '，数据目录临时）', 'client': 'com.rabbitmq:amqp-client:' + versions.get('amqp-client', 'unknown'), 'os': platform.platform(), 'arch': platform.machine(), 'erlang': subprocess.run(['/opt/homebrew/bin/erl', '-version'], capture_output=True, text=True).stderr.strip(), }, 'reports': reports, 'testCount': 11, 'results': results, 'observed': OBSERVED, 'artifacts': digests, 'logs': str(BASE), 'cleanup': { 'broker': '本脚本启动的 RabbitMQ node 已 stop；mnesia/log 位于临时目录（本次运行的 ' + str(BASE) + '）', 'queues': '每个用例在结束时 queueDelete/exchangeDelete 自己的命名空间；默认交换机不做删除', }, 'limits': [ '单节点 RabbitMQ 4.4.0；集群、quorum 队列多数派、镜像/仲裁队列跨节点恢复未运行', '持久化只验证 persistent 消息 confirm 与单节点重启前行为；broker 崩溃/断电恢复未运行', '未运行 TLS、SASL 外部认证、 shovel/federation、stream 插件、优先级队列与惰性队列', 'prefetch 只验证单 channel basicQos；consumer 多 channel 分摊与公平分发未专项运行', ], } out = ROOT / 'evidence' / ('u58-' + 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=11', flush=True)