import importlib.util
import io
import json
from pathlib import Path
import re
import subprocess
import urllib.error

import yaml

ROOT = Path(__file__).resolve().parents[1]
ADDON = ROOT / 'argocd/platform-addons/eso-token-renewer'
MINT = ROOT / 'scripts/openbao-eso-db-token-periodic-attended.sh'
spec = importlib.util.spec_from_file_location('renew', ADDON / 'renew.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)


class Response(io.BytesIO):
    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


def opener_for(ttls):
    seen = []

    def opener(request, timeout):
        token = request.headers['X-vault-token']
        seen.append((request.full_url, request.get_method(), token))
        ttl = ttls[token]
        if isinstance(ttl, Exception):
            raise ttl
        return Response(json.dumps({'auth': {'lease_duration': ttl, 'renewable': True}}).encode())
    return opener, seen


def lanes(tmp_path, tokens):
    for lane, token in tokens.items():
        (tmp_path / lane).mkdir()
        (tmp_path / lane / 'token').write_text(token + '\n')
    m.ROOT = tmp_path


def test_all_lanes_renewed_and_no_token_printed(tmp_path, capsys):
    lanes(tmp_path, {'core-hub': 'tok-a', 'tenant-engine': 'tok-b'})
    opener, seen = opener_for({'tok-a': 604800, 'tok-b': 604800})
    assert m.main(opener) == 0
    assert [s[1] for s in seen] == ['POST', 'POST']
    assert all(s[0].endswith('/v1/auth/token/renew-self') for s in seen)
    out = capsys.readouterr().out
    assert 'tok-a' not in out and 'tok-b' not in out
    assert [json.loads(line)['lane'] for line in out.splitlines()] == ['core-hub', 'tenant-engine']


def test_capped_ttl_and_http_errors_fail_the_job(tmp_path, capsys):
    lanes(tmp_path, {'a': 't1', 'b': 't2', 'c': 't3'})
    denied = urllib.error.HTTPError('u', 403, 'denied', {}, None)
    opener, _ = opener_for({'t1': 604800, 't2': 3600, 't3': denied})
    assert m.main(opener) == 1
    rows = {r['lane']: r for r in map(json.loads, capsys.readouterr().out.splitlines())}
    assert rows['a']['ok'] and rows['b']['error'] == 'ttl_below_threshold'
    assert rows['c']['error'] == 'http_403'


def test_no_lanes_is_a_failure(tmp_path):
    m.ROOT = tmp_path
    assert m.main() == 1


def test_cronjob_mounts_match_mint_script_lanes():
    docs = list(yaml.safe_load_all((ADDON / 'cronjob.yaml').read_text()))
    cron = next(d for d in docs if d['kind'] == 'CronJob')
    pod = cron['spec']['jobTemplate']['spec']['template']['spec']
    mounted = {v['name']: v['secret']['secretName'] for v in pod['volumes'] if 'secret' in v}
    script = MINT.read_text()
    minted = dict(re.findall(r'^\s+([a-z-]+)\) echo "(openbao-[a-z-]+-eso-token) ', script, re.M))
    assert mounted == minted
    assert pod['automountServiceAccountToken'] is False
    assert all(m_['readOnly'] for m_ in pod['containers'][0]['volumeMounts'])


def test_mint_script_uses_period_and_is_silent():
    script = MINT.read_text()
    assert 'exec >/dev/null 2>&1' in script
    assert '-period="$TOKEN_PERIOD"' in script and '-ttl=' not in script
    assert subprocess.run(['sh', '-n', str(MINT)]).returncode == 0
    assert subprocess.run(['sh', str(MINT), '--confirm', 'wrong', 'core-hub']).returncode == 2


def test_mint_policies_exist_or_are_known_external():
    script = MINT.read_text()
    policies = set()
    for group in re.findall(r'-eso-token ([a-z0-9,-]+)"', script):
        policies.update(group.split(','))
    declared = {p.stem for p in (ROOT / 'openbao/policies').glob('*.hcl')}
    # Configured by rapp-postgres scripts (configure-*-openbao.sh), not as .hcl here.
    external = {'external-secrets-canned-prompts-database', 'external-secrets-sbom-nexus-database',
                'credential-broker-tenant-engine-runtime', 'credential-broker-tenant-engine-migration'}
    assert policies - declared <= external