"""Accept #155's original fixture on the release-downloaded 0.1.36 Runtime, without editing it."""
from pathlib import Path
from datetime import datetime
import hashlib, json, re, shutil, sqlite3, subprocess

LAB=Path(__file__).resolve().parents[1]
ROOT=LAB/'evidence/launcher-0.1.36-retest'
E=ROOT/'runs'/datetime.now().strftime('%Y%m%d-%H%M%S-%f'); E.mkdir(parents=True)
CLI=LAB/'tools/cli/Scripts/foggy-runtime.exe'
URL='http://127.0.0.1:18136'
SOURCE=LAB/'issues/time-role-0.1.35'
MODEL='TimeRoleProbeQueryModel'

def save(path,obj):
    path.parent.mkdir(parents=True,exist_ok=True)
    path.write_text(json.dumps(obj,ensure_ascii=False,indent=2),encoding='utf-8')

def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest()

def call(label,ns,*args,expect=True):
    cmd=[str(CLI),'--output','json','--base-url',URL,'--namespace',ns,*map(str,args)]
    proc=subprocess.run(cmd,capture_output=True,text=True,encoding='utf-8',timeout=60)
    (E/(label+'.json')).write_text(proc.stdout,encoding='utf-8')
    if proc.stderr:(E/(label+'.stderr.txt')).write_text(proc.stderr,encoding='utf-8')
    obj=json.loads(proc.stdout)
    with (E/'commands.jsonl').open('a',encoding='utf-8') as out:
        out.write(json.dumps({'label':label,'args':cmd[1:],'exitCode':proc.returncode,'success':obj.get('success')},ensure_ascii=False)+'\n')
    if expect: assert proc.returncode==0 and obj.get('success'),(label,obj)
    return obj

tm=(SOURCE/'models/model/TimeRoleProbeModel.tm').read_text(encoding='utf-8')
qm=(SOURCE/'models/query/TimeRoleProbeQueryModel.qm').read_text(encoding='utf-8')
posted=json.loads((SOURCE/'remote-readback.json').read_text(encoding='utf-8'))['body'].replace('\r\n','\n')
assert [s.strip() for s in re.findall(r'```javascript\n(.*?)\n```',posted,re.S)][:2]==[tm.strip(),qm.strip()]
protected=[SOURCE/'models/model/TimeRoleProbeModel.tm',SOURCE/'models/query/TimeRoleProbeQueryModel.qm',SOURCE/'time-role-probe.sqlite',LAB/'data/Chinook_Sqlite.sqlite',*sorted((LAB/'models').rglob('*')),*sorted((LAB/'chapter-18/models').rglob('*'))]
before={str(p.relative_to(LAB)):digest(p) for p in protected if p.is_file()}
save(E/'original-hashes-before.json',before)
original_ns='time-role-repro-035'
call('wait-ready',original_ns,'wait-ready','--timeout-seconds','15')
cap=call('capabilities',original_ns,'capabilities')
assert cap['engine']=='java' and cap['data']['securityMode']=='none-dev-test-only'
db=E/'time-role-probe.sqlite';shutil.copyfile(SOURCE/'time-role-probe.sqlite',db)
assert digest(db)==digest(SOURCE/'time-role-probe.sqlite')
empty=E/'time-role-empty.sqlite'
with sqlite3.connect(empty) as con:
    con.execute('CREATE TABLE TimeRoleProbe (RecordId INTEGER PRIMARY KEY, BusinessAt DATETIME NOT NULL, LoadedAt DATETIME NOT NULL, CreatedAt DATETIME NOT NULL)')
for ds,file in [('issue155-original',db),('issue155-empty',empty)]:
    call(ds+'-add',original_ns,'datasources','add','--name',ds,'--type','sqlite','--jdbc-url','jdbc:sqlite:'+str(file).replace('\\','/'))
    call(ds+'-test',original_ns,'datasources','test',ds)

def write_models(label,source_tm=tm,source_qm=qm):
    path=E/'models'/label
    (path/'model').mkdir(parents=True); (path/'query').mkdir()
    (path/'model/TimeRoleProbeModel.tm').write_text(source_tm,encoding='utf-8')
    (path/'query/TimeRoleProbeQueryModel.qm').write_text(source_qm,encoding='utf-8')
    return path

def describe(label,ns):
    obj=call(label+'-json',ns,'models','describe',MODEL)
    md=call(label+'-markdown',ns,'models','describe',MODEL,'--format','markdown')
    content=md['data']['content'];(E/(label+'.md')).write_text(content,encoding='utf-8')
    return obj['data']['data']['fields'],content

def load(label,ns,path,ds='issue155-original'):
    call(label+'-bind',ns,'datasources','bind','--namespace',ns,'--data-source',ds)
    call(label+'-inspect',ns,'tables','inspect','--data-source',ds,'--table','TimeRoleProbe','--include-indexes')
    o=call(label+'-validate',ns,'models','validate','--models-dir',path,'--include-stack-trace')
    assert o['data']['valid'] and o['data']['invalidFiles']==0
    call(label+'-bundle',ns,'bundles','add','--name','issue155-'+label,'--path',path)
    call(label+'-refresh',ns,'models','refresh')
    return describe(label,ns)

expected={
    'businessAt':('business_date','business_axis_control'),
    'loadedAt':('system_time','technical_ingestion_only'),
    'createdAt':('business_date','explicit_business_axis'),
    'loadedTime$id':('system_time','technical_ingestion_dimension_only')}

def verify_metadata(fields,content,mapping):
    for name,(role,use) in mapping.items():
        assert fields[name]['timeRole']==role,(name,fields[name])
        assert fields[name]['recommendedUse']==use,(name,fields[name])
        line=next(s for s in content.splitlines() if s.startswith('| '+name+' |'))
        assert f'| {role} |' in line and use in line,(name,line)

path=write_models('original')
fields,content=load('original',original_ns,path)
verify_metadata(fields,content,expected)
actual={k:{'timeRole':fields[k]['timeRole'],'recommendedUse':fields[k]['recommendedUse']} for k in expected}
call('original-refresh-again',original_ns,'models','refresh')
again=describe('after-refresh',original_ns)
verify_metadata(*again,expected)
assert fields==again[0],'Metadata changed after refresh'
print('PASS: original issue configuration, JSON/Markdown and repeated refresh',flush=True)

empty_meta=load('empty','time-role-empty-036',path,ds='issue155-empty')
verify_metadata(*empty_meta,expected)
print('PASS: identical metadata on empty table',flush=True)

fallback_tm=tm
for role,use in list(expected.values())[:3]:
    fallback_tm=fallback_tm.replace(f"timeRole: '{role}', recommendedUse: '{use}'",'/* role/use omitted for fallback control */')
# Removing the entire annotation pair must not leave a comma-only invalid JavaScript object.
fallback_meta=load('fallback','time-role-fallback-036',write_models('fallback',fallback_tm))
for field,role in [('businessAt','business_date'),('loadedAt','business_date'),('createdAt','system_time')]:
    assert fallback_meta[0][field]['timeRole']==role,(field,fallback_meta[0][field])
print('PASS: absent annotations retain field-name fallback',flush=True)

custom_tm=tm.replace("timeRole: 'system_time', recommendedUse: 'technical_ingestion_only'", "timeRole: 'ingestion_checkpoint', recommendedUse: 'technical_ingestion_only'")
custom_meta=load('custom','time-role-custom-036',write_models('custom',custom_tm))
custom_expected={**expected,'loadedAt':('ingestion_checkpoint','technical_ingestion_only')}
verify_metadata(*custom_meta,custom_expected)
print('PASS: custom string role retained',flush=True)

alias_qm=qm.replace('{ref: probe.loadedTime}',"{ref: probe.loadedTime$id, name: 'ingestedAt', alias: 'ingestedAt'}")
alias_meta=load('alias','time-role-alias-036',write_models('alias',tm,alias_qm))
verify_metadata(*alias_meta,{'ingestedAt':expected['loadedTime$id']})
print('PASS: renamed self-dimension ID metadata',flush=True)

invalid_ns='time-role-invalid-036'
call('invalid-bind',invalid_ns,'datasources','bind','--namespace',invalid_ns,'--data-source','issue155-original')
bad_variants={
    'property-role-number':tm.replace("timeRole: 'system_time', recommendedUse: 'technical_ingestion_only'", "timeRole: 123, recommendedUse: 'technical_ingestion_only'"),
    'property-use-object':tm.replace("recommendedUse: 'technical_ingestion_only'", "recommendedUse: {purpose: 'ingestion'}"),
    'dimension-role-boolean':tm.replace("type: 'DATETIME', timeRole: 'system_time',", "type: 'DATETIME', timeRole: false,"),
    'dimension-use-array':tm.replace("recommendedUse: 'technical_ingestion_dimension_only'", "recommendedUse: ['ingestion']")}
rejected=[]
for name,source in bad_variants.items():
    bad=write_models(name,source)
    result=call(name,invalid_ns,'models','validate','--models-dir',bad,'--include-stack-trace',expect=False)
    data=result.get('data',{})
    assert not result.get('success') or (data.get('valid') is False and data.get('invalidFiles',0)>0),(name,result)
    diagnostic=json.dumps(result,ensure_ascii=False)
    needle='timeRole' if '-role-' in name else 'recommendedUse'
    assert needle in diagnostic and ('TimeRoleProbeModel' in diagnostic),name
    messages=result.get('diagnostics',{}).get('attributes',{}).get('validation',{}).get('errors',[])
    assert any('INVALID_TIME_METADATA' in d.get('message','') and needle in d['message'] and 'must be a string or null' in d['message'] for d in messages),name
    rejected.append({'case':name,'success':result.get('success'),'valid':data.get('valid'),'invalidFiles':data.get('invalidFiles'),'error':result.get('error'),'fieldDiagnostics':messages})
print('PASS: four non-string annotations rejected before registration',flush=True)

query_specs=[
    ('row',original_ns,{'columns':['recordId','businessAt','loadedAt','createdAt','loadedTime$id'],'limit':5},1),
    ('business-november',original_ns,{'columns':['recordId'],'slice':[{'field':'businessAt','op':'[)','value':['2012-11-01','2012-12-01']}],'limit':5},1),
    ('ingestion-november',original_ns,{'columns':['recordId'],'slice':[{'field':'loadedAt','op':'[)','value':['2012-11-01','2012-12-01']}],'limit':5},0),
    ('dimension-december',original_ns,{'columns':['recordId','loadedTime$id'],'slice':[{'field':'loadedTime$id','op':'[)','value':['2012-12-01','2012-12-02']}],'limit':5},1),
    ('renamed-dimension-december','time-role-alias-036',{'columns':['recordId','ingestedAt'],'slice':[{'field':'ingestedAt','op':'[)','value':['2012-12-01','2012-12-02']}],'limit':5},1)]
for name,ns,payload,count in query_specs:
    p=E/'queries'/(name+'.json');save(p,payload)
    call('query-'+name+'-validate',ns,'query','validate',MODEL,'--payload',p)
    result=call('query-'+name+'-execute',ns,'query','execute',MODEL,'--payload',p)
    assert len(result['data']['items'])==count,(name,result)
    if count:
        assert result['data']['items'][0]['recordId']==1
    if name=='row':
        row=result['data']['items'][0]
        assert row['businessAt'].startswith('2012-11-29') and row['createdAt'].startswith('2012-11-29')
        assert row['loadedAt'].startswith('2012-12-01') and row['loadedTime$id'].startswith('2012-12-01')
    if name=='renamed-dimension-december':assert result['data']['items'][0]['ingestedAt'].startswith('2012-12-01')
print('PASS: five date query/physical binding checks',flush=True)

after={str(p.relative_to(LAB)):digest(p) for p in protected if p.is_file()}
assert before==after
save(E/'original-hashes-after.json',after)
identity=json.loads((ROOT/'runtime-identity.json').read_text(encoding='utf-8-sig'))
assert identity['launcher']=='0.1.36' and identity['actualProcessUsesExpectedJar']
report={'issue':155,'result':'PASS','launcher':'0.1.36','cli':'0.1.25','runtime':URL,
    'sourceCommit':identity['sourceCommit'],'jarSha256':identity['jarSha256'],
    'originalIssueModelsMatchPostedBody':True,'actualOriginalMetadata':actual,
    'metadataScenarios':['original one-row table','same metadata after refresh','empty table','fallback when omitted','custom string role','renamed self-dimension ID'],
    'nonStringValidationRejections':rejected,'dateQueriesPassed':5,
    'originalIssueAndTeachingFilesUnchanged':True,'evidenceDirectory':str(E),
    'scope':'Isolated release runtime; no source build, no broad release/Console/multi-database acceptance, no GitHub comments'}
save(E/'acceptance.json',report);save(ROOT/'acceptance.json',report)
save(ROOT/'latest-run.json',{'evidenceDirectory':str(E)})
print(json.dumps(report,ensure_ascii=True,indent=2))