#!/usr/bin/env python3
"""U34 公开入口：真实 Config Server HTTP、Git 与隔离 Kafka 的单模块测试。

只在完整测试、观察行、版本和进程组清理均通过后记录成功；失败与中断也保留日志。
"""
import glob
import hashlib
import json
import os
import pathlib
import platform
import re
import shutil
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
EVIDENCE = ROOT / 'evidence/u34-2026-10-27.json'
TEST_CLASS = 'study.configbus.ConfigBusHttpTest'
EXPECTED_TEST_COUNT = 6
EXPECTED_TESTS = {
    'clientLoadsGitProfileThroughRealHttpServer',
    'gitCommitDoesNotRefreshObjectsUntilTheLocalEndpointRuns',
    'busDestinationUpdatesOnlyBAndBroadcastConvergesBoth',
    'oneFailedConfigPathLeavesMixedVersionsAndExplicitRefreshRepairsIt',
    'invalidLazyTargetFailsAfterRefreshAndGitRevertRestoresWithANewVersion',
    'ciphertextAtRestDoesNotReplaceAuthenticationOrEnvironmentRedaction',
}
MARKERS = {
    'U34 initialRead=PASS',
    'U34 localRefresh=PASS',
    'U34 busPropagation=PASS',
    'U34 partialFailure=PASS',
    'U34 rollback=PASS',
    'U34 secretProtection=PASS',
}
EXPECTED_VERSIONS = {
    'org.springframework.boot:spring-boot': '4.0.9',
    'org.springframework.cloud:spring-cloud-config-server': '5.0.6',
    'org.springframework.cloud:spring-cloud-config-client': '5.0.6',
    'org.springframework.cloud:spring-cloud-bus': '5.0.4',
    'org.springframework.kafka:spring-kafka': '4.0.8',
    'org.springframework.kafka:spring-kafka-test': '4.0.8',
    'org.apache.kafka:kafka_2.13': '4.1.2',
    'org.apache.kafka:kafka-clients': '4.1.2',
}


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 artifact_digests():
    paths = [ROOT / name for name in ('pom.xml', 'run.py', 'README.md')]
    paths += sorted(path for path in (ROOT / 'src').rglob('*') if path.is_file())
    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 live_group_pids(pgid):
    # killpg(..., 0) also sees zombies; they cannot hold ports or do more work.
    snapshot = subprocess.run(['/bin/ps', '-axo', 'pid=,pgid=,stat='],
                              capture_output=True, text=True, check=True, timeout=5)
    members = []
    for line in snapshot.stdout.splitlines():
        pid, group, state = line.split()
        if int(group) == pgid and not state.startswith('Z'):
            members.append(int(pid))
    return members


def cleanup_group(process):
    # The leader can exit before its descendants, including on exit=0.
    for signum, grace in ((signal.SIGTERM, 2), (signal.SIGKILL, 5)):
        process.poll()
        if not live_group_pids(process.pid):
            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, logs, env, results, timeout=600):
    log_path = logs / (name + '.log')
    started = now()
    process = None
    failure = None
    cleanup_error = None
    exit_code = 0
    try:
        # A surviving descendant cannot keep a PIPE reader waiting indefinitely.
        with log_path.open('w', encoding='utf-8') as stream:
            try:
                process = subprocess.Popen(args, cwd=ROOT, env=env, text=True,
                                           stdout=stream, stderr=subprocess.STDOUT,
                                           start_new_session=True)
                code = process.wait(timeout=timeout)
                exit_code = code if code >= 0 else 128 - code
            except BaseException as exc:
                failure = exc
                exit_code = failure_code(exc)
            finally:
                # A second signal cannot interrupt bounded cleanup/status writing.
                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(encoding='utf-8', errors='replace') if log_path.exists() else ''
    result = {
        'name': name, 'command': args, 'cwd': str(ROOT),
        'startedAt': started, 'finishedAt': now(),
        '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,
        'log': str(log_path),
        'logSha256': file_sha(log_path) if log_path.exists() else None,
        'observed': [line for line in output.splitlines() if line.startswith('U34 ')],
    }
    results.append(result)
    write_json(logs / (name + '.status.json'), result)
    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('DONE ' + name + ' exit=0; log: ' + str(log_path), flush=True)
    return output


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 validate_reports(logs, not_before_ns):
    report_dir = ROOT / 'target/surefire-reports'
    paths = sorted(report_dir.glob('TEST-*.xml'))
    expected_file = 'TEST-' + TEST_CLASS + '.xml'
    if [path.name for path in paths] != [expected_file]:
        raise ValueError('Surefire 报告缺失或多出报告: ' + str([path.name for path in paths]))
    path = paths[0]
    if path.stat().st_mtime_ns < not_before_ns:
        raise ValueError('Surefire 报告不是本次测试生成: ' + str(path))
    shutil.copyfile(path, logs / path.name)
    suite = ET.parse(path).getroot()
    if suite.tag != 'testsuite' or suite.get('name') != TEST_CLASS:
        raise ValueError('Surefire testsuite 名称不匹配')
    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('Surefire ' + field + '=' + report[field])
    cases = suite.findall('testcase')
    names = [case.attrib['name'] for case in cases]
    if (len(EXPECTED_TESTS) != EXPECTED_TEST_COUNT or int(report['tests']) != EXPECTED_TEST_COUNT
            or len(cases) != EXPECTED_TEST_COUNT
            or len(set(names)) != len(names) or set(names) != EXPECTED_TESTS):
        raise ValueError('Surefire 测试名/数量不匹配: ' + str(names)
                         + '; expected=' + str(sorted(EXPECTED_TESTS)))
    for case in cases:
        if case.get('classname') != TEST_CLASS 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)
    write_json(logs / 'reports.json', [report])
    return [report]


def validate_observations(observed):
    for marker in MARKERS:
        matches = [line for line in observed if line == marker or line.startswith(marker + ' ')]
        if len(matches) != 1:
            raise ValueError('观察行必须出现一次: ' + marker + '; actual=' + str(len(matches)))
    if len(observed) != len(MARKERS):
        raise ValueError('存在未登记或重复的 U34 观察行: ' + str(observed))


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)
    return versions


def validate_pom():
    pom = ET.parse(ROOT / 'pom.xml').getroot()
    ns = {'m': 'http://maven.apache.org/POM/4.0.0'}
    expected = {
        'm:parent/m:version': '4.0.9',
        'm:properties/m:java.version': '21',
        'm:properties/m:spring-cloud.version': '2025.1.4',
    }
    for query, value in expected.items():
        if pom.findtext(query, namespaces=ns) != value:
            raise ValueError('POM 版本不匹配: ' + query + '; expected=' + value)
    return {'spring-boot': '4.0.9', 'spring-cloud': '2025.1.4', 'java': '21'}


def main():
    signal.signal(signal.SIGTERM, interrupted)
    signal.signal(signal.SIGINT, interrupted)
    logs = None
    evidence = {
        'ticket': 'U34 #71', 'status': 'running',
        'startedAt': now(), 'finishedAt': None,
        'input': 'python3 labs/config-bus/run.py', 'exitCode': None,
        'environment': {'os': platform.platform(), 'arch': platform.machine(),
                        'python': platform.python_version()},
        'expectedTestCount': EXPECTED_TEST_COUNT, 'expectedTestNames': sorted(EXPECTED_TESTS),
        'reports': [], 'testCount': 0, 'observed': [], 'results': [],
        'artifacts': {}, 'logs': None,
        'cleanup': {'processGroups': '尚未完成',
                    'resources': '测试 fixture 的 close 负责关闭 Spring Context/HTTP、Kafka 与临时 Git'},
        'limits': [
            'Config Server 与两个客户端是同一测试 JVM 内三个独立 Spring Context；不是三个 OS 进程',
            '真实 loopback HTTP、本地临时 Git 与隔离 embedded Kafka；故障注入代理是受控测试设施',
            '部分失败仅阻断 B 的 Config HTTP、消费 max-attempts=1；显式本地刷新恢复，无消费者离线或 Kafka 断线重放证明',
            '不证明生产配置刷新原子覆盖所有实例；无跨进程内存隔离、生产原子回滚或一致性认证',
            'TLS/mTLS、RabbitMQ、Nacos 配置服务、生产集群、OS 进程崩溃与断电恢复均未运行',
        ],
    }
    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/u34'
        logs_root.mkdir(parents=True, exist_ok=True)
        logs = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=logs_root))
        evidence['logs'] = str(logs)
        print('LOGS ' + str(logs), flush=True)
        if len(sys.argv) != 1:
            raise ValueError('本入口不接受参数；使用 python3 labs/config-bus/run.py')
        if len(EXPECTED_TESTS) != EXPECTED_TEST_COUNT or len(MARKERS) != EXPECTED_TEST_COUNT:
            raise ValueError('U34 测试/观察行清单尚未完整登记，必须有 ' + str(EXPECTED_TEST_COUNT) + ' 项')
        evidence['artifacts'] = artifact_digests()
        evidence['environment']['declaredVersions'] = validate_pom()
        env = dict(os.environ)
        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, '-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'])
        evidence['reports'] = validate_reports(logs, test_started_ns)
        evidence['observed'] = evidence['results'][-1]['observed']
        validate_observations(evidence['observed'])
        dependencies = logs / 'dependencies.txt'
        run('dependencies', base + ['dependency:list', '-DincludeScope=test',
                                   '-DappendOutput=false', '-DoutputFile=' + str(dependencies)],
            logs, env, evidence['results'])
        evidence['environment']['dependencies'] = read_dependencies(dependencies)
        evidence['dependencyReport'] = {'path': str(dependencies), 'sha256': file_sha(dependencies)}
        if artifact_digests() != evidence['artifacts']:
            raise ValueError('运行期间实验文件发生变化；请在修改结束后重新运行')
        evidence['testCount'] = sum(int(report['tests']) for report in evidence['reports'])
        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:
            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 not evidence['results']:
                evidence['cleanup']['processGroups'] = '未启动命令进程'
            elif any(result['cleanupError'] is not None for result in evidence['results']):
                evidence['cleanup']['processGroups'] = '进程组清理失败，详见 results.cleanupError'
            else:
                evidence['cleanup']['processGroups'] = '本次启动的全部进程组已定向回收'
            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 U34 tests=' + str(evidence['testCount']) + '; evidence: ' + str(EVIDENCE), flush=True)
    return exit_code


if __name__ == '__main__':
    sys.exit(main())