import importlib.util,json,hashlib,shutil,sys,uuid,os
from pathlib import Path
import pytest,yaml
SOURCE=Path(__file__).resolve().parents[1]/'scripts/task_execution_plan.py'
spec=importlib.util.spec_from_file_location('task_execution_plan',SOURCE)
m=importlib.util.module_from_spec(spec);sys.modules[spec.name]=m;spec.loader.exec_module(m)
PLATFORM=Path(os.environ.get('PLATFORM_PLAN_TEST_SOURCE_ROOT',str(SOURCE.parents[1])))
SECRETS=Path(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT',str(PLATFORM.parent/'secrets-engine')))
@pytest.fixture
def compiler(tmp_path):
    repo=tmp_path/'platform';(repo/'scripts').mkdir(parents=True);(repo/'credential-change-requests').mkdir();(repo/'openbao/policies').mkdir(parents=True)
    shutil.copyfile(PLATFORM/'scripts/credential-change.py',repo/'scripts/credential-change.py')
    ccr=yaml.safe_load((PLATFORM/'credential-change-requests/CCR-2026-0002-issue-core-ingestion-api-key.yaml').read_text())
    ccr['status']='approved';ccr['access_frontdoor'].update(readiness='approved-pending-apply',resolvable=False)
    ccrfile=repo/'credential-change-requests/CCR-2026-0002-fixture.yaml';ccrfile.write_text(yaml.safe_dump(ccr))
    policy=repo/ccr['openbao']['policy_file'];shutil.copyfile(PLATFORM/ccr['openbao']['policy_file'],policy)
    catalogdir=tmp_path/'catalog';catalogdir.mkdir();catalog=yaml.safe_load((SECRETS/'catalog/issue-core-ingestion-api-key.yaml').read_text())
    catalog['delivery_auth']['management']='existing'
    catalogfile=catalogdir/'issue-core-ingestion-api-key.yaml';catalogfile.write_text(yaml.safe_dump(catalog))
    paths=[SOURCE,repo/'scripts/credential-change.py',ccrfile,policy,catalogfile,*[SECRETS/'src/secrets_engine'/name for name in ['catalog.py','__init__.py','errors.py','redact.py']]]
    pins=tuple((str(p),m.tagged(p.read_bytes())) for p in paths)
    return m.Compiler(m.OwnerConfig(repo,SECRETS,ccrfile.name,catalogfile.name,catalogdir,pins))

def request(c):
    value=c.proposal();value.pop('catalog_target');value.update(operation_id=str(uuid.uuid4()),request_digest='sha256:'+'a'*64);return value

def test_compiles_real_owner_generated_bodies_and_fixed_six_calls(compiler):
    req=request(compiler);plan=compiler.compile(req)
    calls=[compiler.before_call(plan,i) for i in range(6)]
    assert [x.method for x in calls]==['GET','PUT','GET','GET','PUT','GET']
    assert len({x.path for x in calls})==2
    policy=json.loads(calls[1].payload)['policy'];assert 'capabilities = ["read"]' in policy
    role=json.loads(calls[4].payload);assert role['policies']=='workload-kv-read-issue-core-runtime'
    assert role['bound_service_account_namespaces']==['external-secrets']
    assert len(plan.content_digest)==64 and not plan.content_digest.startswith('sha256:')
    assert all(x.payload_sha256.startswith('sha256:') and len(x.call_digest)==64 for x in calls)

@pytest.mark.parametrize('field,value',[('action','apply'),('kind','service-jwt-login'),('catalog_id','another'),('ccr_sha256','sha256:'+'b'*64),('catalog_sha256','sha256:'+'b'*64),('request_digest','a'*64)])
def test_altered_bindings_refuse(compiler,field,value):
    req=request(compiler);req[field]=value
    with pytest.raises(m.Refused):compiler.compile(req)

@pytest.mark.parametrize('ordinal',[False,0.0,'0',-1,6])
def test_fixed_ordinals_refuse(compiler,ordinal):
    plan=compiler.compile(request(compiler))
    with pytest.raises(m.Refused):compiler.before_call(plan,ordinal)

@pytest.mark.parametrize('field,value',[('method','DELETE'),('path','platform/data/workloads/keycape/x'),('payload_sha256','sha256:'+'c'*64)])
def test_caller_operation_cannot_change_target_body_or_method(compiler,field,value):
    req=request(compiler);req['operations'][1][field]=value
    with pytest.raises(m.Refused):compiler.compile(req)

@pytest.mark.parametrize('which',['ccr','catalog','policy','validator'])
def test_owner_source_drift_between_calls_refuses(compiler,which):
    plan=compiler.compile(request(compiler));compiler.before_call(plan,0)
    path={'ccr':compiler.ccr,'catalog':compiler.catalog,'policy':next(Path(p) for p in compiler.pins if p.endswith('.hcl')),'validator':compiler.validator}[which]
    path.write_bytes(path.read_bytes()+b'\n# drift\n')
    with pytest.raises(m.Refused):compiler.before_call(plan,1)

@pytest.mark.parametrize('change',['extra','budget','request'])
def test_direct_plan_constructor_cannot_bypass_recompilation(compiler,change):
    plan=compiler.compile(request(compiler));data=plan.metadata()
    if change=='extra':data['caller_hcl']='path "*" {}'
    elif change=='budget':data['task_token_max_uses']=999
    else:data['request']['operations'][3]['payload_sha256']='sha256:'+'d'*64
    with pytest.raises(m.Refused):compiler.before_call(m.Plan(m.encoded(data)),3)


def repin(c,path):
    pins=dict(c.config.source_pins);pins[str(path)]=m.tagged(path.read_bytes())
    return m.Compiler(m.OwnerConfig(c.config.platform_repo,c.config.secrets_repo,c.config.ccr_filename,c.config.catalog_filename,c.config.catalog_directory,tuple(pins.items())))

@pytest.mark.parametrize('status',['in_flight','proposed','applied','verified','active','cancelled'])
def test_even_re_pinned_nonapproved_owner_record_refuses(compiler,status):
    ccr=yaml.safe_load(compiler.ccr.read_text());ccr['status']=status;compiler.ccr.write_text(yaml.safe_dump(ccr))
    with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()


def test_source_role_policy_escalation_refuses_even_when_re_pinned(compiler):
    ccr=yaml.safe_load(compiler.ccr.read_text());ccr['openbao']['auth']['policies']=['root'];compiler.ccr.write_text(yaml.safe_dump(ccr))
    with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()


def test_catalog_hash_agrees_with_actual_consumer_unicode_canonicalization(compiler):
    catalog=yaml.safe_load(compiler.catalog.read_text());catalog['consumers'][0]['purpose']='café';compiler.catalog.write_text(yaml.safe_dump(catalog))
    compiler=repin(compiler,compiler.catalog)
    from secrets_engine.execution_contract import catalog_target_digest
    proposed=compiler.proposal();assert proposed['catalog_sha256']==catalog_target_digest(proposed['catalog_target'])


def test_caller_file_path_and_policy_fields_are_refused(compiler):
    req=request(compiler);req['file']='/etc/passwd';req['policy']='path "*" {}'
    with pytest.raises(m.Refused):compiler.compile(req)


def test_actual_inflight_ccr42_is_not_an_admitted_target():
    actual=yaml.safe_load((PLATFORM/'credential-change-requests/CCR-2026-0042-forgejo-netkingdom-oidc-consumer-reader.yaml').read_text())
    assert actual['status']=='in_flight' and 'auth' not in actual['openbao']

@pytest.mark.parametrize('kind',['symlink','hardlink','writable'])
def test_owner_policy_file_indirection_or_permissions_refuse(compiler,tmp_path,kind):
    req=request(compiler)
    path=next(Path(p) for p in compiler.pins if p.endswith('.hcl'))
    if kind=='symlink':
        copy=tmp_path/'outside-policy';copy.write_bytes(path.read_bytes());path.unlink();path.symlink_to(copy)
    elif kind=='hardlink':
        import os
        os.link(path,tmp_path/'policy-link')
    else:path.chmod(0o666)
    with pytest.raises(m.Refused):compiler.compile(req)


def test_re_pinned_policy_body_escalation_refuses(compiler):
    path=next(Path(p) for p in compiler.pins if p.endswith('.hcl'))
    path.write_text('path "*" { capabilities = ["sudo", "read", "update"] }\n')
    with pytest.raises(m.Refused):repin(compiler,path).proposal()


def test_duplicate_owner_yaml_key_refuses_even_when_re_pinned(compiler):
    compiler.ccr.write_bytes(compiler.ccr.read_bytes()+b'\nstatus: approved\n')
    with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()


def test_unconfirmed_claim_binding_refuses_even_when_re_pinned(compiler):
    ccr=yaml.safe_load(compiler.ccr.read_text());ccr['openbao']['auth']['bound_claims_confirmed']=False
    compiler.ccr.write_text(yaml.safe_dump(ccr))
    with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()


def test_catalogue_engine_managed_approle_is_not_silently_generic_apply(compiler):
    data=yaml.safe_load(compiler.catalog.read_text());data['delivery_auth']['management']='engine'
    compiler.catalog.write_text(yaml.safe_dump(data))
    with pytest.raises(m.Refused):repin(compiler,compiler.catalog).proposal()


def test_oidc_role_body_reuses_owner_auth_payload_and_existing_mount(compiler):
    ccr=yaml.safe_load(compiler.ccr.read_text())
    source=yaml.safe_load((PLATFORM/'credential-change-requests/CCR-2026-0034.yaml').read_text())
    ccr['openbao']['auth']=source['openbao']['auth'];ccr['openbao']['auth']['role']='synthetic-operator-workload-kv-read'
    ccr['openbao']['auth']['policies']=[ccr['openbao']['policy_name']]
    compiler.ccr.write_text(yaml.safe_dump(ccr));compiler=repin(compiler,compiler.ccr)
    plan=compiler.compile(request(compiler));call=compiler.before_call(plan,4)
    role=json.loads(call.payload)
    assert call.path=='auth/netkingdom/role/synthetic-operator-workload-kv-read'
    assert role['role_type']=='oidc' and role['policies']=='workload-kv-read-issue-core-runtime'
    assert role['bound_claims']==source['openbao']['auth']['bound_claims']