import importlib.util
import os
from pathlib import Path
import stat
import json
from types import SimpleNamespace

import pytest


ROOT=Path(__file__).resolve().parents[1]
spec=importlib.util.spec_from_file_location('trace_install',ROOT/'tools/icerti_offline/trace_install.py')
install=importlib.util.module_from_spec(spec);spec.loader.exec_module(install)


def test_image_code_readable_with_private_umask(tmp_path):
    source=tmp_path/'source';source.mkdir()
    (source/'module').mkdir();(source/'module/main.py').write_text('value=1\n')
    (source/'module/main.py').chmod(0o600)
    old=os.umask(0o077)
    try:install.image_payload(source,tmp_path/'build/payload',['module/main.py'])
    finally:os.umask(old)
    assert stat.S_IMODE((tmp_path/'build').stat().st_mode)==0o700
    for p in ('build/payload','build/payload/module'):
        assert stat.S_IMODE((tmp_path/p).stat().st_mode)==0o755
    assert stat.S_IMODE((tmp_path/'build/payload/module/main.py').stat().st_mode)==0o644
    assert stat.S_IMODE((source/'module/main.py').stat().st_mode)==0o600


@pytest.mark.parametrize('name',['../outside.py','/outside.py'])
def test_image_payload_rejects_path_escape(tmp_path,name):
    with pytest.raises(ValueError,match='TRACE_PAYLOAD_PATH'):
        install.image_payload(tmp_path,tmp_path/'dest',[name])


def test_export_uses_actual_mount_not_server_alias():
    row={'Config':{'Env':['CONTEXTRA_ARTIFACT_ROOT=/srv/artifacts']},
         'Mounts':[{'Type':'bind','Source':'/disk/site/artifacts','Destination':'/srv/artifacts'}]}
    assert str(install.export_directory(row))=='/disk/site/diagnostics/exports'
    row['Mounts']=[]
    with pytest.raises(ValueError,match='TRACE_ARTIFACT_MOUNT_REQUIRED'):install.export_directory(row)


def test_hook_compatibility_still_rejects_changed_function():
    spec=importlib.util.spec_from_file_location('trace_payload',ROOT/'tools/icerti_offline/trace_payload.py')
    payload=importlib.util.module_from_spec(spec);spec.loader.exec_module(payload)
    with pytest.raises(ValueError,match='TRACE_BASELINE_FUNCTION_CHANGED'):
        payload.compatible('def completed():\n return 1\n','def completed():\n return 2\n','completed')


def test_rollback_once_preserves_data_and_targets_only_changed_services(tmp_path,monkeypatch):
    i=install.Installer.__new__(install.Installer)
    i.root=tmp_path;i.tools=tmp_path/'tools';i.state=tmp_path/'state'
    i.tools.mkdir();i.state.mkdir();(tmp_path/'tool-backup').mkdir()
    (i.tools/'installed-files.json').write_text('{}')
    (tmp_path/'tool-backup/installed-files.json').write_text('{}')
    old={'physical':'old-compose'};new={'physical':'new-compose'};current=[new]
    services=('analytics','ai-control-sidecar','ai-gateway','redis','postgresql')
    before={k:{'id':k,'image':'old-'+k,'running':True} for k in services}
    plan={'active':old,'protected':{},'before':before,'toolExisted':['installed-files.json'],'session':None}
    for name,data in [('plan',plan),('activation',{'after':new}),('tools-installed',{'inventoryHash':install.sha(i.tools/'installed-files.json')})]:
        (tmp_path/(name+'.json')).write_text(json.dumps(data))
    rows={k:{'Id':k,'Image':('new-'+k if k in services[:2] else 'old-'+k),
             'State':{'Running':False,'ExitCode':0,'Health':{'Status':'healthy'}}} for k in services}
    i.rows=lambda:rows;i.guard_payload=lambda:None
    i.failure_evidence=lambda action:commands.append(['capture',action])
    i.cb=SimpleNamespace(active=lambda:current[0],inspect=lambda:None,compose_args=lambda p:['compose',p])
    i.s=SimpleNamespace(verify_tools=lambda:None)
    writes=[];commands=[]
    def atomic(path,value,**kwargs):
        writes.append(str(path))
        if path==i.state/'active.json':current[0]=value
    i.fp=SimpleNamespace(atomic=atomic)
    i.save=lambda name,value:(tmp_path/(name+'.json')).write_text(json.dumps(value))
    def command(args,*a,**kw):
        commands.append(args)
        if args[:3]==['compose','old-compose','start']:
            for k in services[:2]:rows[k]['Image']='old-'+k;rows[k]['State']['Running']=True
        return b''
    monkeypatch.setattr(install,'command',command)
    i.rollback()
    assert current[0]==old and (tmp_path/'rollback-complete.json').exists()
    assert writes==[str(i.tools/'installed-files.json'),str(i.state/'active.json')]
    assert commands[-1]==['docker','start','ai-gateway']
    assert commands[0]==['capture','before-rollback']
    assert all('redis' not in cmd and 'postgresql' not in cmd for cmd in commands)
    previous=len(commands)
    with pytest.raises(ValueError,match='ALREADY_ATTEMPTED'):i.rollback()
    assert len(commands)==previous


@pytest.mark.parametrize('code,oom,logs,okay',[
    (0,False,'',True),
    (143,False,'Shutting down\nApplication shutdown complete\nFinished server process',True),
    (143,False,'Shutting down',False),
    (137,False,'Application shutdown complete',False),
    (143,True,'Shutting down\nApplication shutdown complete\nFinished server process',False),
])
def test_sigterm_requires_current_transition_shutdown_evidence(tmp_path,monkeypatch,code,oom,logs,okay):
    i=install.Installer.__new__(install.Installer);i.root=tmp_path;i.save=lambda *args:None
    (tmp_path/'transition.json').write_text('{}')
    calls=[]
    def run(args,**kwargs):
        calls.append(args);return SimpleNamespace(returncode=0,stdout=logs.encode(),stderr=b'')
    monkeypatch.setattr(install.subprocess,'run',run)
    rows={'ai-control-sidecar':{'Id':'fixture','State':{'Running':False,'ExitCode':code,'OOMKilled':oom}}}
    if okay:i.check_stopped(rows,['ai-control-sidecar'])
    else:
        with pytest.raises(ValueError,match='STOP_INCOMPLETE'):i.check_stopped(rows,['ai-control-sidecar'])
    if calls:assert calls[0][2]=='--since'


@pytest.mark.parametrize('kind',['absent','retained','partial','changed','symlink'])
def test_retained_policy_pair_never_overwrites(tmp_path,kind):
    src,applied=tmp_path/'src.json',tmp_path/'applied.json'
    settings={'enabled':True}
    if kind!='absent':src.write_text(json.dumps(settings))
    if kind in ('retained','changed'):applied.write_text(json.dumps(settings if kind=='retained' else {'enabled':False}))
    if kind=='symlink':applied.symlink_to(src)
    before={p:p.read_bytes() for p in (src,applied) if p.exists()}
    if kind in ('absent','retained'):
        assert install.policy_pair(src,applied,settings)==(kind=='retained')
    else:
        with pytest.raises(ValueError,match='TRACE_POLICY'):install.policy_pair(src,applied,settings)
    assert all(p.read_bytes()==v for p,v in before.items())


def test_failure_evidence_masks_container_logs(tmp_path,monkeypatch):
    i=install.Installer.__new__(install.Installer);i.root=tmp_path
    i.rows=lambda:{'analytics':{'Id':'fixture','Image':'old','State':{'Running':False,'ExitCode':1}}}
    saved=[];i.save=lambda name,data:saved.append(data)
    monkeypatch.setattr(install.subprocess,'run',lambda *a,**k:SimpleNamespace(stdout=b'password=secret123\nAuthorization: Bearer sensitive123\nRuntimeError: BOOT_FAILURE',stderr=b''))
    i.failure_evidence('fixture')
    encoded=json.dumps(saved)
    assert 'secret123' not in encoded and 'sensitive123' not in encoded
    assert 'BOOT_FAILURE' in encoded and 'ExitCode' in encoded