#!/usr/bin/env python3
"""Verify source-generated root entrypoints with real local processes only."""
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import socket
import subprocess
import tempfile
import time
import urllib.error
import urllib.request

ROOT = Path(__file__).resolve().parents[1]


def run(args, cwd, env):
    result = subprocess.run(args, cwd=cwd, env=env, capture_output=True, text=True, timeout=120)
    if result.returncode:
        output = re.sub(r'https://[^/@\s]+:[^/@\s]+@', 'https://', result.stdout + result.stderr)
        raise RuntimeError(f'{args[0]} failed: {output[-8000:]}')
    return result.stdout


def free_port():
    with socket.socket() as s:
        s.bind(('127.0.0.1', 0))
        return s.getsockname()[1]


def request(url, data=None, headers=None, method=None):
    try:
        req = urllib.request.Request(url, data=data, headers=headers or {}, method=method)
        with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(req, timeout=0.4) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read()


def verify_contract_changes(cli, project, env):
    api = project/'api/service.api'
    logic = project/'internal/logic/greet/greet.go'
    original = logic.read_bytes()
    api.write_text(api.read_text().replace('GreetReq', 'HelloReq'))
    def sources():
        return {str(p.relative_to(project)): p.read_bytes() for p in project.rglob('*') if p.is_file()}
    before = sources()
    preview = run([str(cli), 'api', 'generate', '--dry-run'], project, env)
    assert 'ERROR [logicSignature]' in preview and '*types.HelloReq' in preview
    assert before == sources(), 'preview modified project'
    for generated in (False, True):
        if generated:
            output = run([str(cli), 'api', 'generate'], project, env)
            assert 'ERROR [logicSignature]' in output and logic.read_bytes() == original
        before = sources()
        result = subprocess.run([str(cli), 'api', 'generate', '--check'], cwd=project, env=env, capture_output=True, text=True, timeout=30)
        assert result.returncode != 0 and 'ERROR [logicSignature]' in result.stdout
        assert 'internal/logic/greet/greet.go:' in result.stdout and before == sources()
    logic.write_bytes(original.replace(b'GreetReq', b'HelloReq'))
    run([str(cli), 'api', 'generate', '--check'], project, env)
    run(['go', 'test', '-mod=mod', './...'], project, env)
    run(['go', 'build', '.'], project, env)


def serve(args, project, env, transport, sig, strict_exit, contract=False, logs=None):
    port = free_port()
    env = dict(env, GO_OMNI_HTTP_ADDR=f'127.0.0.1:{port}', GO_OMNI_TRANSPORT=transport)
    log_path = (logs or project.parent) / f'{transport}-{sig.name}-{strict_exit}-{contract}.log'
    with log_path.open('wb') as log:
        proc = subprocess.Popen(args, cwd=project, env=env, stdout=log, stderr=log, start_new_session=True)
        try:
            deadline = time.monotonic() + 30
            base = f'http://127.0.0.1:{port}'
            while time.monotonic() < deadline:
                if proc.poll() is not None:
                    raise RuntimeError('server exited before readiness: ' + log_path.read_text()[-2000:])
                try:
                    if request(base + '/health/ready')[0] == 200:
                        break
                except (OSError, urllib.error.URLError):
                    pass
                time.sleep(0.05)
            else:
                raise RuntimeError('server readiness timed out')
            if contract:
                status, body = request(base + '/check?kind=fixed')
                assert status == 400 and json.loads(body) == {'code': 400002, 'msg': '用户名格式不正确', 'data': None}
                assert json.loads(request(base + '/missing')[1])['msg'] == '接口不存在'
                assert request(base + '/check', method='POST')[0] == 405
                assert json.loads(request(base + '/raw')[1]) == {'bytes': 3}
                assert request(base + '/download') == (200, b'file-content')
                assert request(base + '/callback', b'{ "raw" : 1 }', {'X-Signature': 'valid'}) == (200, b'ACK:{ "raw" : 1 }')
                assert request(base + '/callback', b'x' * 65, {'X-Signature': 'valid'})[0] == 413
                upload = b'--omniBoundary\r\nContent-Disposition: form-data; name="file"; filename="test.txt"\r\nContent-Type: text/plain\r\n\r\nhello\r\n--omniBoundary--\r\n'
                status, body = request(base + '/upload', upload, {'Content-Type': 'multipart/form-data; boundary=omniBoundary'})
                assert status == 200 and json.loads(body)['data'] == {'bytes': 5}
                # Exercise malformed multipart over real HTTP on both transports.
                for broken in (upload.split(b'hello')[0] + b'private file bytes', b'--omniBoundary\r\nContent-Disposition: form-data'):
                    status, body = request(base + '/upload', broken, {'Content-Type': 'multipart/form-data; boundary=omniBoundary'})
                    assert status == 400 and json.loads(body) == {'code': 400004, 'msg': '上传内容不完整', 'data': None}
            else:
                status, body = request(base + '/api/greet/Alice')
                assert status == 200 and json.loads(body)['data']['message'] == 'Hello Alice'
                assert request(base + '/missing')[0] == 404
            os.killpg(proc.pid, sig)
            exit_code = proc.wait(timeout=10)
            if strict_exit:
                assert exit_code == 0, exit_code
            # A go run parent can exit from the signal; its compiled child must also release its listener.
            deadline = time.monotonic() + 5
            while time.monotonic() < deadline:
                try:
                    with socket.create_connection(('127.0.0.1', port), timeout=0.1):
                        pass
                except OSError:
                    break
                time.sleep(0.05)
            else:
                raise RuntimeError('listener remains after shutdown')
        finally:
            try:
                os.killpg(proc.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            proc.wait(timeout=10)


def main():
    started = time.monotonic()
    env = dict(os.environ, GOWORK='off', GOFLAGS='')
    # Prevent inherited project configuration from binding a real server or loading secrets.
    env = {k: v for k, v in env.items() if not k.startswith('GO_OMNI_')}
    with tempfile.TemporaryDirectory(prefix='omni-managed-') as temp:
        work = Path(temp)
        cli = work / 'go-omni'
        project = work / 'service'
        run(['go', 'build', '-o', str(cli), './cmd/go-omni'], ROOT, env)
        run([str(cli), 'new', '--framework-dir', str(ROOT), '--module', 'example.com/managedSmoke', str(project)], ROOT, env)
        run(['go', 'mod', 'tidy'], project, env)
        run(['go', 'build', '.'], project, env)
        assert (project / 'main.go').is_file() and not (project / 'cmd/server/main.go').exists()
        before = {str(p.relative_to(project)): hashlib.sha256(p.read_bytes()).hexdigest() for p in project.rglob('*.go')}
        run([str(cli), 'api', 'generate'], project, env)
        run([str(cli), 'api', 'generate', '--check'], project, env)
        after = {str(p.relative_to(project)): hashlib.sha256(p.read_bytes()).hexdigest() for p in project.rglob('*.go')}
        assert before == after
        verify_contract_changes(cli, project, env)
        serve(['go', 'run', '.'], project, env, 'nethttp', signal.SIGTERM, False)
        for transport in ('nethttp', 'native'):
            for sig in (signal.SIGINT, signal.SIGTERM):
                serve([str(project / 'managedSmoke')], project, env, transport, sig, True)
        example = ROOT/'examples/managedService'
        run(['go', 'build', '-o', str(work/'managedExample'), '.'], example, env)
        for transport in ('nethttp', 'native'):
            serve([str(work/'managedExample')], example, env, transport, signal.SIGTERM, True, contract=True, logs=work)
    sources = ['cmd/go-omni/apigen/generate.go', 'cmd/go-omni/apigen/templates.go', 'cmd/go-omni/apigen/startup.go', 'cmd/go-omni/apigen/managed_test.go', 'cmd/go-omni/apigen/contractChanges.go', 'cmd/go-omni/apigen/logicCheck.go', 'cmd/go-omni/apigen/write.go', 'cmd/go-omni/api.go', 'config/config.go', 'httpx/transfer.go', 'httpx/multipart.go', 'scripts/verifyManagedProject.py']
    sources += [str(p.relative_to(ROOT)) for p in (ROOT/'cmd/go-omni/scaffold/templates').glob('api-*.tmpl')]
    sources += [str(p.relative_to(ROOT)) for p in (ROOT/'cmd/go-omni/apigen/testdata/managed').glob('*.go')]
    report = {'passed': True, 'mode': 'localSource', 'durationSeconds': round(time.monotonic()-started, 2), 'checks': ['rootGoRun', 'rootGoBuild', 'stableRegeneration', 'checkMainOwnership', 'contractChangePreview', 'logicSignatureCheckPersists', 'businessSourcePreserved', 'fixedConsumerBuild', 'nethttpSIGINT', 'nethttpSIGTERM', 'nativeSIGINT', 'nativeSIGTERM', 'listenerReleased', 'temporaryProcessesRemoved', 'nethttpGeneratedTransfers', 'nativeGeneratedTransfers', 'malformedMultipart400', 'goOwnedErrors', 'default404And405'], 'sourceHashes': {p: hashlib.sha256((ROOT/p).read_bytes()).hexdigest() for p in sources}}
    (ROOT/'performance/managedProjectLocal.json').write_text(json.dumps(report, indent=2)+'\n')
    print('PASS generated root project: go run/build, regeneration, nethttp/native signals; temporary files removed')


if __name__ == '__main__':
    main()