import importlib.util
from pathlib import Path
import sys
import tomllib

ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'deploy'))
spec=importlib.util.spec_from_file_location('deployment_v2',ROOT/'deploy/deploy.py')
deployment=importlib.util.module_from_spec(spec);sys.modules[spec.name]=deployment;spec.loader.exec_module(deployment)

def test_units_have_no_sudo_or_root_runtime():
    for mode,executor,user in [('serve',False,'aim-web'),('worker',False,'aim-web'),('executor',True,'svc_bf-ansible')]:
        text=deployment.unit_text(user,991,Path('/etc/ansible/scripts/config/webgui.toml'),mode,executor=executor,executor_group=1001 if executor else None,supplementary_group=991 if executor else None,executor_local_home='/home/svc_bf-ansible' if executor else None)
        assert f'User={user}\n'in text
        assert 'NoNewPrivileges=true' in text
        assert 'CapabilityBoundingSet=\n'in text
        assert 'CAP_SETUID'not in text and 'sudo'not in text
        assert 'RuntimeDirectory=aim-web-executor' in text if executor else 'RuntimeDirectory='not in text
        if executor: assert 'RuntimeDirectoryMode=0711' in text
        if executor:
            assert 'Group=1001\n' in text
            assert 'SupplementaryGroups=991\n' in text
            assert 'Environment=HOME=/var/lib/aim-web-executor' in text
            assert 'ReadWritePaths=/var/lib/aim-web-executor /var/lib/aim-web-executor/.ansible/tmp /home/svc_bf-ansible/.ansible/tmp -/etc/ansible/inventories' in text
            assert 'core-staging-check' in text


def test_no_private_core_import_and_no_old_stage_bridge():
    text=(ROOT/'deploy/deploy.py').read_text()
    assert 'from aim.config'not in text and 'sys.path.insert'not in text
    assert 'key_export.py'not in text
    assert 'core_transport="stdio"'in text
    assert '--migrate-core'in text
    assert "'db','migrate'"in text
    assert text.index("os.replace(stage, target)")<text.index('atomic_bytes(EXECUTOR_UNIT')


def test_site_profile_and_versions():
    from aim_webgui.config import Settings
    from aim_webgui import __version__,SCHEMA_VERSION,HTTP_API_VERSION
    settings=Settings.load(ROOT/'deploy/webgui.example.toml')
    assert settings.core_transport=='unix' and settings.core_executor_user=='svc_bf-ansible'
    assert settings.host=='127.0.0.1' and settings.execution_require_approval is False
    assert len(settings.execution_playbooks)==14
    assert __version__=='2.1.0rc9' and SCHEMA_VERSION==5 and HTTP_API_VERSION==2
    metadata=tomllib.loads((ROOT/'pyproject.toml').read_text())
    assert metadata['tool']['aim-web']['http-api']==2


def test_stage_and_rollback_preserve_modes(tmp_path):
    target=tmp_path/'file';deployment.atomic_bytes(target,b'first',0o640)
    deployment.atomic_bytes(target,b'second',0o600)
    assert target.read_bytes()==b'second' and target.stat().st_mode&0o777==0o600
    link=tmp_path/'link';link.symlink_to(target)
    import pytest
    with pytest.raises(ValueError):deployment.atomic_bytes(link,b'unsafe')
    assert target.read_bytes()==b'second'


def test_release_manifest_detects_changed_or_traversing_payload(tmp_path):
    import hashlib
    import pytest
    files={'pyproject.toml':b'project','deploy/deploy.py':b'installer','src/aim_webgui/__init__.py':b'version'}
    for name,data in files.items():
        p=tmp_path/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(data)
    manifest=tmp_path/'MANIFEST.sha256'
    original=''.join(hashlib.sha256(data).hexdigest()+'  '+name+'\n'for name,data in files.items())
    manifest.write_text(original);deployment.verify_release(tmp_path)
    dotted=''.join(hashlib.sha256(data).hexdigest()+'  ./'+name+'\n' for name,data in files.items())
    manifest.write_text(dotted);deployment.verify_release(tmp_path)
    manifest.write_text(original)
    (tmp_path/'pyproject.toml').write_bytes(b'changed')
    with pytest.raises(ValueError,match='integrity'):deployment.verify_release(tmp_path)
    manifest.write_text('0'*64+'  ../outside\n')
    with pytest.raises(ValueError,match='manifest'):deployment.verify_release(tmp_path)


def test_home_and_direct_profile_defaults():
    from aim_webgui.config import Settings
    from dataclasses import replace
    import pytest
    settings=Settings.load(ROOT/'deploy/webgui.example.toml')
    assert settings.core_home==Path('/var/lib/aim-web-executor')
    with pytest.raises(ValueError):replace(settings,core_home=settings.state_dir).validate()
    browsing=Settings.load(ROOT/'deploy/profiles/direct-npm.toml')
    assert not browsing.execution_enabled and not browsing.credentials_enabled


def test_executor_identity_preserves_primary_gid_and_scopes_web_group():
    text=(ROOT/'deploy/deploy.py').read_text()
    assert 'os.setgid(account.pw_gid)' in text
    assert "os.getgrouplist(user,account.pw_gid) + [web.pw_gid]" in text
    executor=(ROOT/'src/aim_webgui/core/executor.py').read_text()
    assert 'os.chown(parent,-1,client_gid)' not in executor
    assert "stat.S_IMODE(st.st_mode)!=0o711" in executor
    assert 'os.chown(path,-1,client_gid)' in executor


def test_executor_socket_readiness_waits_for_type_simple_bind(monkeypatch,tmp_path):
    attempts=[]
    def fake_validate(*args,**kwargs):
        attempts.append(1)
        if len(attempts)<3:
            raise ValueError('Managed executor socket is missing or a symlink')
    class Result:
        returncode=0
    monkeypatch.setattr(deployment,'validate_executor_runtime_permissions',fake_validate)
    monkeypatch.setattr(deployment.subprocess,'run',lambda *a,**k: Result())
    monkeypatch.setattr(deployment.time,'sleep',lambda *_: None)
    deployment.wait_executor_runtime_permissions('aim-web','svc-bf',tmp_path/'core.sock',timeout=1)
    assert len(attempts)==3


def test_executor_socket_readiness_fails_if_service_exits(monkeypatch,tmp_path):
    import pytest
    monkeypatch.setattr(deployment,'validate_executor_runtime_permissions',lambda *a,**k: (_ for _ in ()).throw(ValueError('missing socket')))
    class Result:
        returncode=3
    monkeypatch.setattr(deployment.subprocess,'run',lambda *a,**k: Result())
    with pytest.raises(ValueError,match='exited before'):
        deployment.wait_executor_runtime_permissions('aim-web','svc-bf',tmp_path/'core.sock',timeout=1)

def test_restore_guard_probes_restored_adapter_as_executor_before_start(monkeypatch,tmp_path):
    d=deployment.Deployment(tmp_path,'root',0);calls=[]
    monkeypatch.setattr(d,'as_executor',lambda user,cmd,**kw:calls.append((user,cmd)))
    assert d.restored_core_compatible({'venv':str(tmp_path/'old'),'executor_user':'nobody'})
    assert calls[0][0]=='nobody' and str(tmp_path/'old/bin/python')==str(calls[0][1][0])
    assert 'core_transport="stdio"'in calls[0][1][3]
    def fail(*args,**kwargs):raise deployment.subprocess.CalledProcessError(1,['fixture'])
    monkeypatch.setattr(d,'as_executor',fail)
    assert not d.restored_core_compatible({'venv':str(tmp_path/'old')})