#!/usr/bin/env python3
"""U24: fixed-version conditional wiring, real Actuator readiness and ES8 adaptation."""
import hashlib,json,os,pathlib,platform,re,subprocess,time
from datetime import datetime
ROOT=pathlib.Path(__file__).resolve().parent
JAVA_HOME=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(JAVA_HOME),PATH=str(JAVA_HOME/'bin')+os.pathsep+os.environ.get('PATH',''))
VERSION=subprocess.check_output([str(JAVA_HOME/'bin/java'),'-version'],stderr=subprocess.STDOUT,text=True,env=ENV)
assert re.search(r'version "21[.]',VERSION),VERSION
LOGS=pathlib.Path.home()/'.cache/software-engineering-study/java-depth-impl/u24/runtime';LOGS.mkdir(parents=True,exist_ok=True)
RESULTS=[]
INVOCATION=datetime.now().astimezone().isoformat()
def command(name,args,markers=(),expected=0):
    started=datetime.now().astimezone().isoformat()
    p=subprocess.run([str(a) for a in args],cwd=ROOT,env=ENV,text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=600)
    (LOGS/(name+'.log')).write_text(p.stdout)
    assert p.returncode==expected,(name,p.returncode,p.stdout[-7000:])
    for marker in markers:assert marker in p.stdout,(name,marker,p.stdout[-7000:])
    RESULTS.append({'name':name,'invocationId':INVOCATION,'startedAt':started,'finishedAt':datetime.now().astimezone().isoformat(),'exitCode':p.returncode,'input':[str(a) for a in args]})
    print('PASS',name,flush=True)
    return p.stdout

# Runtime fixtures are separate from the exact auto-configuration probes.
import base64,contextlib,http.server,threading,urllib.request,urllib.error
class DependencyHandler(http.server.BaseHTTPRequestHandler):
    healthy=True
    def log_message(self,*args):pass
    def do_GET(self):
        if self.path=='/dependency':
            self.send_response(200 if self.healthy else 503);self.end_headers();self.wfile.write(b'up' if self.healthy else b'down')
        else:
            body=json.dumps({'traceparent':self.headers.get('traceparent',''),'customizer':self.headers.get('X-U24-Client','')}).encode();self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(body)
@contextlib.contextmanager
def dependency():
    service=http.server.ThreadingHTTPServer(('127.0.0.1',0),DependencyHandler);thread=threading.Thread(target=service.serve_forever,name='u24-owned-dependency');thread.start()
    try:yield 'http://127.0.0.1:'+str(service.server_port)
    finally:
        service.shutdown();service.server_close();thread.join(timeout=3);assert not thread.is_alive(),'dependency fixture leaked'
@contextlib.contextmanager
def server(name,dependency_url,extra=(),ready=True,jar=None):
    jar=jar or ROOT/'server/target/readiness-server-1.jar';args=[JAVA_HOME/'bin/java','-jar',jar,'--lab.dependency-url='+dependency_url+'/dependency','--lab.downstream-url='+dependency_url+'/echo',*extra];log=LOGS/(name+'.log');started=datetime.now().astimezone().isoformat()
    with log.open('w') as output:
        process=subprocess.Popen([str(a) for a in args],cwd=ROOT,env=ENV,text=True,stdout=output,stderr=subprocess.STDOUT)
        try:
            deadline=time.monotonic()+25
            while time.monotonic()<deadline:
                text=log.read_text();business=re.search(r'BUSINESS_PORT (\d+)',text);management=re.search(r'MANAGEMENT_PORT (\d+)',text)
                if business and ('READY' if ready else 'STARTED') in text:break
                if process.poll() is not None:raise AssertionError((name,process.returncode,text[-6000:]))
                time.sleep(.04)
            else:raise AssertionError((name,'startup timed out',log.read_text()[-6000:]))
            biz='http://127.0.0.1:'+business[1];mgmt='http://127.0.0.1:'+(management[1] if management else business[1]);yield biz,mgmt,process
        finally:
            if process.poll() is None:
                process.terminate()
                try:process.wait(timeout=10)
                except subprocess.TimeoutExpired:process.kill();process.wait(timeout=5);raise AssertionError(name+' required forced cleanup kill')
            assert process.poll() is not None
    RESULTS.append({'name':name,'invocationId':INVOCATION,'startedAt':started,'finishedAt':datetime.now().astimezone().isoformat(),'input':[str(a) for a in args],'exitCode':process.returncode,'reaped':True})
def request(origin,path,user=None,headers=None):
    actual={'Accept':'application/json',**(headers or {})}
    if user:actual['Authorization']='Basic '+base64.b64encode((':'.join(user)).encode()).decode()
    try:
        with urllib.request.urlopen(urllib.request.Request(origin+path,headers=actual),timeout=5) as response:return response.status,json.loads(response.read() or b'{}')
    except urllib.error.HTTPError as error:
        body=error.read();return error.code,json.loads(body) if body else {}
OPS=('ops','ops-password');READER=('reader','reader-password')

def probe_cases():
    command('probe-build',['mvn','-q','-pl','probe','test'])
    observed={'conditions':json.loads((ROOT/'probe/target/probe-results.json').read_text()),
              'messaging':json.loads((ROOT/'probe/target/messaging-results.json').read_text()),
              'session':json.loads((ROOT/'probe/target/session-results.json').read_text())}
    conditions=observed['conditions'];messaging=observed['messaging'];session=observed['session']
    assert conditions['jdbc']['mysql'] and conditions['jdbc']['userBackoff'] and conditions['jdbc']['twoCandidatesNoTemplate'],conditions['jdbc']
    assert conditions['redis']['ping'] and conditions['redis']['factoryBackoff'] and conditions['redis']['differentNameDoesNotBackoff'],conditions['redis']
    assert conditions['mongodb']['ping'] and conditions['mongodb']['userBackoff'] and conditions['mongodb']['writes']==0,conditions['mongodb']
    assert conditions['cache']['withoutEnableCaching'] and conditions['cache']['simple'] and conditions['cache']['userBackoff'] and conditions['cache']['redisRead'] and conditions['cache']['ownedKeyDeleted'],conditions['cache']
    for broker in ('kafka','amqp','jms','pulsar'):
        assert messaging[broker]['brokerMessages']==0,messaging[broker]
    assert messaging['jms']['noFactoryNoTemplate'] and messaging['amqp']['twoFactoriesNoTemplate'],messaging
    assert messaging['restClient']['prototypeBuilder'] and messaging['restClient']['userBuilderBackoff'],messaging['restClient']
    assert session['session']['cookieSaveRead'] and session['session']['ownedKeysDeleted'] and session['session']['ownedKeys']==1,session['session']
    assert session['sessionIndexed']['configureActionNone'] and session['sessionIndexed']['ownedKeysDeleted'],session['sessionIndexed']
    assert session['sessionTakeover']['userBackoff'] and session['sessionClassCondition']['missingClassSkipped'],session
    RESULTS[-1]['observed']=observed
    return observed

def es8_case():
    command('es8-dependency-tree',['mvn','-f','es8/pom.xml','dependency:tree','-Dincludes=org.springframework.boot:spring-boot-elasticsearch,co.elastic.clients:*'],markers=('elasticsearch-java:jar:8.19.23',))
    out=command('es8-run',['mvn','-q','-f','es8/pom.xml','compile','spring-boot:run'],markers=('RESULT ',))
    line=next(line for line in out.splitlines() if line.startswith('RESULT '))
    observed=json.loads(line[7:])
    assert observed['serverVersion'].startswith('8.19.') and observed['bootDefaultRejected'] and observed['manualClientBean'] and observed['transportClosed'] and observed['writes']==0,observed
    RESULTS[-1]['observed']=observed
    return observed

def runtime_cases():
    with dependency() as url:
        with server('default-security',url,['--management.endpoints.web.exposure.include=health']) as (biz,mgmt,process):
            status,body=request(mgmt,'/actuator/health');assert status==200,(status,body)
            status,probe=request(mgmt,'/actuator/health/readiness');assert status==200 and probe['status']=='UP',(status,probe)
            assert request(mgmt,'/actuator/info')[0]==401
            assert request(mgmt,'/actuator/info',OPS)[0]==404
            status,state=request(biz,'/lab/state',OPS);assert status==200 and state['defaultManagementChain'] and state['defaultUsers'],(status,state)
            RESULTS.append({'name':'default-security-observed','invocationId':INVOCATION,'state':state,'health':body,'readinessProbe':probe})
        with server('custom-security',url,['--lab.custom-security=true','--management.endpoints.web.exposure.include=health,info,metrics,env,auditevents,labcontrol','--management.endpoint.health.probes.add-additional-paths=true']) as (biz,mgmt,process):
            assert request(mgmt,'/actuator/info')[0]==401
            assert request(mgmt,'/actuator/info',('ops','wrong'))[0]==401
            assert request(mgmt,'/actuator/info',READER)[0]==403
            status,info=request(mgmt,'/actuator/info',OPS);assert status==200 and info['build']['artifact']=='readiness-server',(status,info)
            _,state=request(biz,'/lab/state',OPS);assert not state['defaultManagementChain'] and not state['defaultUsers'],state
            assert sorted(state['controlOperations'])==['READ','WRITE'],state
            status,env=request(mgmt,'/actuator/env',OPS);assert status==200 and 'synthetic-private-do-not-return' not in json.dumps(env),env
            _,health=request(mgmt,'/actuator/health');assert 'components' not in health,health
            _,health_reader=request(mgmt,'/actuator/health',READER);assert 'components' not in health_reader,health_reader
            _,health_ops=request(mgmt,'/actuator/health',OPS);assert 'components' in health_ops,health_ops
            status,readyz=request(biz,'/readyz');assert status==200 and readyz['status']=='UP',(status,readyz)
            status,livez=request(biz,'/livez');assert status==200 and livez['status']=='UP',(status,livez)
            for _ in range(3):assert request(biz,'/lab/count',OPS)[0]==200
            status,metric=request(mgmt,'/actuator/metrics/u24.orders.accepted',OPS);assert status==200 and metric['measurements'][0]['value']==3,(status,metric)
            request(biz,'/lab/audit',OPS)
            status,audits=request(mgmt,'/actuator/auditevents',OPS);assert status==200,(status,audits)
            types={event['type'] for event in audits['events']};assert {'AUTHENTICATION_SUCCESS','AUTHENTICATION_FAILURE','AUTHORIZATION_FAILURE','U24_ORDER_REPLAY'}<=types,types
            trace='11111111111111111111111111111111';parent='2222222222222222'
            status,call=request(biz,'/lab/call',OPS,{'traceparent':'00-'+trace+'-'+parent+'-01'});assert status==200 and call['traceId']==trace,call
            assert call['downstream']['traceparent'].split('-')[1]==trace and call['downstream']['customizer']=='auto-customizer',call
            _,bare=request(biz,'/lab/call?bare=true',OPS);assert bare['downstream']['traceparent']=='' and bare['downstream']['customizer']=='',bare
            _,spans=request(biz,'/lab/spans',OPS);matched=[span for span in spans if span['traceId']==trace];assert any(span['kind']=='SERVER' and span['parentId']==parent and span['ended'] for span in matched),matched
            assert any(span['kind']=='CLIENT' and span['ended'] for span in matched),matched
            RESULTS.append({'name':'custom-runtime-observed','invocationId':INVOCATION,'state':state,'info':info,'metric':metric,'auditTypes':sorted(types),'call':call,'bareCall':bare,'readyz':readyz,'livez':livez,'sdkSpans':matched})

def readiness_timeline():
    with dependency() as url:
        release=LOGS/('release-'+datetime.now().strftime('%H%M%S')+'.token');release.unlink(missing_ok=True)
        try:
            with server('readiness-timeline',url,['--management.server.port=0','--lab.release-file='+str(release),'--management.endpoints.web.exposure.include=health'],ready=False) as (biz,mgmt,process):
                assert mgmt!=biz,(biz,mgmt)
                status,probe=request(mgmt,'/actuator/health/readiness');assert status==503 and probe['status']=='OUT_OF_SERVICE',(status,probe)
                status,liveness=request(mgmt,'/actuator/health/liveness');assert status==200 and liveness['status']=='UP',(status,liveness)
                status,opened=request(biz,'/lab/open',OPS);assert status==200 and opened['readiness']=='REFUSING_TRAFFIC',(status,opened)
                assert request(biz,'/lab/gated',OPS)[0]==503
                _,state=request(biz,'/lab/state',OPS);events=state['events']
                assert 'business-listening' in events and 'management-listening' in events and 'started' in events and 'ready' not in events,events
                assert events.index('business-listening')<events.index('started') and events.index('management-listening')<events.index('started'),events
                release.write_text('release')
                log=LOGS/'readiness-timeline.log';deadline=time.monotonic()+25
                while time.monotonic()<deadline:
                    if 'READY' in log.read_text():break
                    time.sleep(.04)
                else:raise AssertionError('released runner did not reach ready')
                status,probe=request(mgmt,'/actuator/health/readiness');assert status==200 and probe['status']=='UP',(status,probe)
                assert request(biz,'/lab/gated',OPS)[0]==200
                _,state=request(biz,'/lab/state',OPS);events=state['events']
                assert events.index('started')<events.index('ready') and 'runner-exit' in events,events
                RESULTS.append({'name':'readiness-timeline-observed','invocationId':INVOCATION,'businessPort':biz,'managementPort':mgmt,'refusingProbe':probe,'events':events})
        finally:
            release.unlink(missing_ok=True)

def write_evidence():
    tree=command('server-dependency-tree',['mvn','-pl','server','dependency:tree','-Dincludes=org.springframework.boot:spring-boot-starter-parent,org.springframework.security:*,io.micrometer:*,io.opentelemetry:opentelemetry-sdk-trace'],markers=('spring-boot-starter-actuator:jar:4.0.9',))
    def version(artifact):
        match=re.search(re.escape(artifact)+r':jar:([0-9][^:]+)',tree);assert match,(artifact,'version not resolved')
        return match.group(1)
    (ROOT/'evidence').mkdir(exist_ok=True)
    evidence={'ticket':'U24 #61','generatedAt':datetime.now().astimezone().isoformat(),
      'environment':{'os':platform.platform(),'arch':platform.machine(),'jdk':VERSION.strip(),'boot':'4.0.9','framework':'7.0.10','security':version('org.springframework.security:spring-security-config'),'micrometerCore':version('io.micrometer:micrometer-core'),'micrometerTracingBridgeOtel':version('io.micrometer:micrometer-tracing-bridge-otel'),'opentelemetrySdkTrace':version('io.opentelemetry:opentelemetry-sdk-trace')},
      'input':'JAVA_HOME=$(/usr/libexec/java_home -v21) python3 labs/boot-readiness/run.py',
      'latestInvocation':{'id':INVOCATION,'finishedAt':datetime.now().astimezone().isoformat()},
      'artifactSha256':hashlib.sha256((ROOT/'server/target/readiness-server-1.jar').read_bytes()).hexdigest(),
      'runnerSha256':hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(),
      'exitCode':0,'rawLogs':str(LOGS),'results':RESULTS,
      'limits':['Loopback real Boot/Security/Actuator/Micrometer/OTel SDK only; no external Collector, registry or K8s probe service is contacted',
        'Probes use a real dependency fixture and a controlled runner release file; timings are local bounded fixtures, not production startup SLAs',
        'No real Kafka/RabbitMQ/JMS/Pulsar broker, ES writes, Mongo writes or user Redis data are touched: probe fixtures prove wiring/back-off only',
        'Metrics/audit/tracing observations cover this ticket assertions, not cardinality budgets or retention policies of a production platform']}
    (ROOT/'evidence/u24-2026-10-26.json').write_text(json.dumps(evidence,ensure_ascii=False,indent=2)+'\n')
    print('PASS U24, evidence written',flush=True)

if __name__=='__main__':
    probe_cases()
    es8_case()
    command('server-build',['mvn','-q','-pl','server','package'])
    runtime_cases()
    readiness_timeline()
    write_evidence()