"""Offline Docker HTTP boundary tests; no Docker daemon or model calls."""
import asyncio
import copy
import json
import os

import httpx
import pytest
import yaml
import socket

from control.provisioner import OwnershipError, Provisioner, ProvisioningError, container_name

ALICE = {'id': 'a' * 32, 'username': 'alice'}
BOB = {'id': 'b' * 32, 'username': 'bob'}


class DockerBoundary:
    def __init__(self):
        self.containers, self.networks, self.created, self.deleted = {}, {}, [], []
        self.fail_start = False
        self.restarted = []
        self.fail_restart = False

    def __call__(self, request):
        path, method = request.url.path, request.method
        if path == '/networks/create':
            spec = json.loads(request.content)
            name = spec['Name']
            self.networks[name] = dict(spec, Id='net-' + name, Internal=False)
            return httpx.Response(201, json={'Id': 'net-' + name})
        if path.startswith('/networks/'):
            key = path.split('/')[2]
            net = self.networks.get(key) or next((n for n in self.networks.values() if n['Id'] == key), None)
            if method == 'DELETE' and net:
                del self.networks[net['Name']]
                return httpx.Response(204)
            return httpx.Response(200, json=net) if net else httpx.Response(404)
        if path == '/containers/create':
            name = request.url.params['name']
            if name in self.containers:
                return httpx.Response(409)
            spec = json.loads(request.content)
            self.created.append(spec)
            self.containers[name] = {'Id': 'id-' + name, 'Config': copy.deepcopy(spec),
                'HostConfig': copy.deepcopy(spec['HostConfig']),
                'State': {'Running': False}, 'NetworkSettings': {'Ports': {}}}
            return httpx.Response(201, json={'Id': 'id-' + name})
        key = path.split('/')[2]
        info = self.containers.get(key) or next((c for c in self.containers.values() if c['Id'] == key), None)
        if not info:
            return httpx.Response(404)
        if method == 'GET':
            return httpx.Response(200, json=info)
        if method == 'POST':
            if path.endswith('/restart'):
                self.restarted.append(key)
                if self.fail_restart:
                    self.fail_restart = False
                    return httpx.Response(500)
            elif not path.endswith('/start'):
                raise AssertionError(path)
            if self.fail_start:
                self.fail_start = False
                return httpx.Response(500)
            info['State']['Running'] = True
            info['NetworkSettings']['Ports'] = {'8642/tcp': [{'HostIp': '127.0.0.1', 'HostPort': str(31000 + len(self.containers))}]}
            return httpx.Response(204)
        self.deleted.append((key, dict(request.url.params)))
        name = next(n for n, c in self.containers.items() if c is info)
        del self.containers[name]
        return httpx.Response(204)


@pytest.fixture
def boundary(tmp_path, monkeypatch):
    monkeypatch.delenv('HERMES_UPSTREAM_URL', raising=False)
    monkeypatch.delenv('HERMES_UPSTREAM_MODEL', raising=False)
    docker, health_requests = DockerBoundary(), []
    def health(request):
        health_requests.append(request)
        assert request.url.host == '127.0.0.1'
        assert request.url.path == '/v1/models'
        return httpx.Response(200, json={'data': []})
    dc = httpx.AsyncClient(base_url='http://docker', transport=httpx.MockTransport(docker))
    hc = httpx.AsyncClient(transport=httpx.MockTransport(health))
    p = Provisioner(tmp_path / 'private', docker_client=dc, health_client=hc,
        upstream_api_key='runtime-provider-secret', runtime_uid=os.geteuid() or 10000,
        runtime_gid=os.getegid() or 10000, ready_timeout=.1, poll_interval=.001)
    yield p, docker, health_requests
    asyncio.run(dc.aclose())
    asyncio.run(hc.aclose())


def test_account_isolation_and_runtime_configuration(boundary):
    p, docker, health = boundary
    async def exercise():
        alice, bob = await p.ensure(ALICE), await p.ensure(BOB)
        assert await p.ready(ALICE)
        return alice, bob
    alice, bob = asyncio.run(exercise())
    assert alice['container_name'] == container_name(ALICE)
    assert alice['hermes_url'].startswith('http://127.0.0.1:')
    assert alice['hermes_api_key'] != bob['hermes_api_key']
    assert len(alice['hermes_api_key']) >= 32
    for user, spec, connection in zip((ALICE, BOB), docker.created, (alice, bob)):
        assert spec['Labels'] == {'hope.managed': 'true', 'hope.user_id': user['id']}
        assert spec['User'].split(':')[0] != '0'
        assert spec['Entrypoint'] == ['/opt/hermes/.venv/bin/python']
        env = dict(entry.split('=', 1) for entry in spec['Env'])
        assert env['HERMES_UPSTREAM_API_KEY'] == 'runtime-provider-secret'
        assert env['API_SERVER_KEY'] == connection['hermes_api_key']
        assert env['API_SERVER_ENABLED'] == 'true'
        assert env['API_SERVER_HOST'] == '127.0.0.1'
        assert int(env['API_SERVER_PORT']) == int(connection['hermes_url'].rsplit(':', 1)[1])
        host = spec['HostConfig']
        assert host['NetworkMode'] == 'host'
        assert 'PortBindings' not in host
        assert 'ExposedPorts' not in spec
        assert host['Memory'] > 0 and host['NanoCpus'] > 0 and host['PidsLimit'] > 0
        assert host['Privileged'] is False and host['Init'] is True
        assert host['CapDrop'] == ['ALL']
        assert host['SecurityOpt'] == ['no-new-privileges:true']
        assert len(host['Binds']) == 2
        assert all(user['id'] in mount and 'docker.sock' not in mount for mount in host['Binds'])
        private, home, workspace = p._paths(user)
        text = (home / 'config.yaml').read_text()
        config = json.loads(text)
        assert config['model'] == {'default': 'auto', 'provider': 'custom:hope-upstream'}
        assert config['providers']['hope-upstream'] == {'api': 'http://192.168.86.5:2099/v1',
            'key_env': 'HERMES_UPSTREAM_API_KEY', 'transport': 'codex_responses', 'default_model': 'auto',
            'models': {'auto': {'supports_vision': True}}, 'discover_models': False}
        assert config['agent']['reasoning_effort'] == 'none'
        assert config['terminal'] == {'backend': 'local', 'cwd': '/workspace'}
        assert {'memory', 'browser', 'web', 'terminal', 'file', 'skills'} <= set(config['platform_toolsets']['api_server'])
        assert config['gateway']['api_server']['max_concurrent_runs'] == 1
        assert 'runtime-provider-secret' not in text and connection['hermes_api_key'] not in text
        assert (private / 'api-key').stat().st_mode & 0o777 == 0o600
        assert private.stat().st_mode & 0o777 == 0o700
        assert workspace != home
    assert all(req.headers['Authorization'].startswith('Bearer ') for req in health)


def test_retries_restart_reuse_key_and_preserve_home(boundary):
    p, docker, _ = boundary
    docker.fail_start = True
    with pytest.raises(ProvisioningError, match='HTTP 500'):
        asyncio.run(p.ensure(ALICE))
    private, home, workspace = p._paths(ALICE)
    key = (private / 'api-key').read_text().strip()
    (workspace / 'notes.txt').write_text('private work')
    (home / 'config.yaml').write_text(json.dumps({'retained': True}))
    result = asyncio.run(p.ensure(ALICE))
    assert result['hermes_api_key'] == key
    assert len(docker.created) == 1
    restarted = Provisioner(p.state_dir, docker_client=p.docker, health_client=p.health,
        upstream_api_key='runtime-provider-secret', runtime_uid=p.runtime_uid, runtime_gid=p.runtime_gid)
    assert asyncio.run(restarted.ensure(ALICE)) == result
    assert yaml.safe_load((home / 'config.yaml').read_text()) == {
        'retained': True, 'gateway': {'api_server': {'host': '127.0.0.1', 'port': p._ports()[ALICE['id']]}}}
    asyncio.run(restarted.remove(ALICE))
    asyncio.run(restarted.remove(ALICE))
    assert docker.deleted == [('id-' + container_name(ALICE), {'force': 'true', 'v': 'false'})]
    assert (workspace / 'notes.txt').read_text() == 'private work'
    assert (private / 'api-key').read_text().strip() == key
    assert not asyncio.run(restarted.ready(ALICE))


@pytest.mark.parametrize('labels', [{}, {'hope.managed': 'false', 'hope.user_id': ALICE['id']},
    {'hope.managed': 'true', 'hope.user_id': BOB['id']}])
def test_no_adoption_or_deletion_of_mismatched_labels(boundary, labels):
    p, docker, _ = boundary
    docker.containers[container_name(ALICE)] = {'Id': 'uncontrolled', 'Config': {'Labels': labels}, 'State': {'Running': True}}
    with pytest.raises(OwnershipError):
        asyncio.run(p.ensure(ALICE))
    with pytest.raises(OwnershipError):
        asyncio.run(p.remove(ALICE))
    assert not asyncio.run(p.ready(ALICE))
    assert docker.deleted == [] and docker.created == []
    assert not p.state_dir.exists()


def test_bounded_readiness_failure_retains_container_for_retry(boundary):
    p, docker, _ = boundary
    bad_health = httpx.AsyncClient(transport=httpx.MockTransport(lambda req: httpx.Response(503)))
    p.health = bad_health
    with pytest.raises(ProvisioningError, match='readiness timed out'):
        asyncio.run(p.ensure(ALICE))
    assert len(docker.created) == 1 and not docker.deleted
    assert container_name(ALICE) in docker.containers
    assert p._read_key(ALICE)
    asyncio.run(bad_health.aclose())


def test_untrusted_paths_and_public_bind_fail_closed(boundary):
    p, docker, _ = boundary
    for user in ({'id': '../bob', 'username': 'alice'}, {'id': ALICE['id'], 'username': '../bob'}):
        with pytest.raises(ValueError):
            asyncio.run(p.ensure(user))
    asyncio.run(p.ensure(ALICE))
    info = docker.containers[container_name(ALICE)]
    info['Config']['Env'] = [e.replace('API_SERVER_HOST=127.0.0.1', 'API_SERVER_HOST=0.0.0.0')
        for e in info['Config']['Env']]
    assert not asyncio.run(p.ready(ALICE))
    with pytest.raises(ProvisioningError, match='loopback'):
        asyncio.run(p.ensure(ALICE))


def test_owned_container_cannot_rotate_missing_private_secret(boundary):
    p, docker, _ = boundary
    asyncio.run(p.ensure(ALICE))
    private, _, _ = p._paths(ALICE)
    key_path = private / 'api-key'
    assert key_path.resolve() == p.state_dir / ALICE['id'] / 'api-key'
    key_path.unlink()
    with pytest.raises(ProvisioningError, match='no persisted API credential'):
        asyncio.run(p.ensure(ALICE))
    assert len(docker.created) == 1


@pytest.mark.parametrize('models', [
    ['auto', 'text-only'],
    {'auto': {'supports_vision': False, 'context_window': 32768}, 'text-only': {'supports_vision': False}},
])
def test_auto_vision_migration_preserves_settings_and_reloads_once(boundary, models):
    p, docker, _ = boundary
    original = asyncio.run(p.ensure(ALICE))
    private, home, workspace = p._paths(ALICE)
    config = yaml.safe_load((home / 'config.yaml').read_text())
    config['providers']['hope-upstream']['models'] = models
    config['model']['default'] = 'text-only'
    config['agent']['reasoning_effort'] = 'high'
    config['retained'] = {'private': True}
    (home / 'config.yaml').write_text(yaml.safe_dump(config))
    (workspace / 'work.txt').write_text('retained')
    assert not asyncio.run(p.ready(ALICE))
    assert asyncio.run(p.ensure(ALICE)) == original
    result = yaml.safe_load((home / 'config.yaml').read_text())
    expected = copy.deepcopy(config)
    if isinstance(models, list):
        expected['providers']['hope-upstream']['models'] = {m: {} for m in models}
    expected['providers']['hope-upstream']['models']['auto']['supports_vision'] = True
    assert result == expected
    assert 'supports_vision' not in result['model']
    assert docker.restarted == ['id-' + container_name(ALICE)]
    assert not docker.deleted and len(docker.created) == 1
    assert (workspace / 'work.txt').read_text() == 'retained'
    assert not p._reload_marker(ALICE).exists()
    assert asyncio.run(p.ready(ALICE))
    asyncio.run(p.ensure(ALICE))
    assert len(docker.restarted) == 1


def test_failed_reload_is_retried_after_control_restart(boundary):
    p, docker, _ = boundary
    original = asyncio.run(p.ensure(ALICE))
    _, home, _ = p._paths(ALICE)
    config = yaml.safe_load((home / 'config.yaml').read_text())
    config['providers']['hope-upstream']['models'] = ['auto']
    (home / 'config.yaml').write_text(yaml.safe_dump(config))
    docker.fail_restart = True
    with pytest.raises(ProvisioningError, match='HTTP 500'):
        asyncio.run(p.ensure(ALICE))
    assert p._reload_marker(ALICE).exists()
    assert not asyncio.run(p.ready(ALICE))
    restarted = Provisioner(p.state_dir, docker_client=p.docker, health_client=p.health,
        upstream_api_key='runtime-provider-secret', runtime_uid=p.runtime_uid, runtime_gid=p.runtime_gid)
    assert asyncio.run(restarted.ensure(ALICE)) == original
    assert len(docker.restarted) == 2 and not docker.deleted
    assert not p._reload_marker(ALICE).exists()


def test_vision_patch_skips_unverified_models_and_preserves_metadata():
    for config in ({'retained': True}, {'providers': {'hope-upstream': {'models': ['text-only']}}}):
        original = copy.deepcopy(config)
        assert Provisioner._declare_auto_vision(config) is False
        assert config == original
    for models in ('auto', [None], {'auto': None}):
        with pytest.raises(ValueError):
            Provisioner._declare_auto_vision({'providers': {'hope-upstream': {'models': models}}})


def test_upstream_runtime_overrides(monkeypatch):
    monkeypatch.setenv('HERMES_UPSTREAM_URL', 'http://127.0.0.1:9999/v1')
    monkeypatch.setenv('HERMES_UPSTREAM_MODEL', 'served-model')
    config = Provisioner.config()
    assert config['model']['default'] == 'served-model'
    provider = config['providers']['hope-upstream']
    assert provider['api'] == 'http://127.0.0.1:9999/v1'
    assert provider['models'] == ['served-model']
    assert provider['default_model'] == 'served-model'


def test_migrate_legacy_container_preserves_state_and_key(boundary):
    p, docker, _ = boundary
    old = asyncio.run(p.ensure(ALICE))
    _, home, workspace = p._paths(ALICE)
    (workspace / 'retained.txt').write_text('account work')
    (home / 'config.yaml').write_text('custom: retained\ngateway:\n  api_server:\n    host: 0.0.0.0\n    port: 8642\n')
    info = docker.containers[container_name(ALICE)]
    info['HostConfig']['NetworkMode'] = 'hope-hermes-net-' + ALICE['id']
    net = info['HostConfig']['NetworkMode']
    docker.networks[net] = {'Id': net, 'Name': net, 'Labels': info['Config']['Labels']}
    assert not asyncio.run(p.ready(ALICE))
    result = asyncio.run(p.ensure(ALICE))
    assert result == old
    assert len(docker.created) == 2 and len(docker.deleted) == 1
    assert not docker.networks
    assert (workspace / 'retained.txt').read_text() == 'account work'
    config = yaml.safe_load((home / 'config.yaml').read_text())
    assert config['custom'] == 'retained'
    assert config['gateway']['api_server'] == {'host': '127.0.0.1', 'port': p._ports()[ALICE['id']]}
    asyncio.run(p.ensure(ALICE))
    assert len(docker.created) == 2


def test_ports_unique_concurrent_persisted_and_skip_busy_port(boundary):
    p, docker, _ = boundary
    with socket.socket() as occupied:
        occupied.bind(('127.0.0.1', 0))
        start = occupied.getsockname()[1]
        p.api_port_range = (start, min(start + 10, 65535))
        async def create():
            return await asyncio.gather(p.ensure(ALICE), p.ensure(BOB))
        alice, bob = asyncio.run(create())
    ports = p._ports()
    assert ports[ALICE['id']] != start
    assert ports[ALICE['id']] != ports[BOB['id']]
    assert (p.state_dir / 'api-ports.json').stat().st_mode & 0o777 == 0o600
    restarted = Provisioner(p.state_dir, docker_client=p.docker, health_client=p.health,
        upstream_api_key='runtime-provider-secret', runtime_uid=p.runtime_uid, runtime_gid=p.runtime_gid)
    assert asyncio.run(restarted.ensure(ALICE)) == alice


def test_port_exhaustion_and_corrupt_registry_fail_closed(boundary):
    p, docker, _ = boundary
    with socket.socket() as occupied:
        occupied.bind(('127.0.0.1', 0))
        p.api_port_range = (occupied.getsockname()[1], occupied.getsockname()[1])
        with pytest.raises(ProvisioningError, match='No available'):
            asyncio.run(p.ensure(ALICE))
    p._directory(p.state_dir)
    p._write_private(p.state_dir / 'api-ports.json', json.dumps({ALICE['id']: 20000, BOB['id']: 20000}))
    with pytest.raises(ProvisioningError, match='Duplicate'):
        asyncio.run(p.ensure(ALICE))
    assert not docker.created


def test_migration_start_failure_retries_without_losing_data(boundary):
    p, docker, _ = boundary
    result = asyncio.run(p.ensure(ALICE))
    docker.containers[container_name(ALICE)]['HostConfig']['NetworkMode'] = 'legacy'
    docker.fail_start = True
    with pytest.raises(ProvisioningError, match='HTTP 500'):
        asyncio.run(p.ensure(ALICE))
    assert asyncio.run(p.ensure(ALICE)) == result
    assert len(docker.created) == 2


def test_webui_migration_unique_ports_passwords_and_cookies(boundary, monkeypatch):
    p, docker, _ = boundary
    original = asyncio.run(p.ensure(ALICE))
    p.webui_enabled = True
    p.webui_host = '127.0.0.1'
    p.image = 'hope-hermes-webui:62e4d7b'
    async def healthy(connection):
        return True
    monkeypatch.setattr(p, '_healthy', healthy)
    alice, bob = asyncio.run(p.ensure(ALICE)), asyncio.run(p.ensure(BOB))
    assert alice == original
    assert len(docker.created) == 3
    a = dict(e.split('=', 1) for e in docker.created[1]['Env'])
    b = dict(e.split('=', 1) for e in docker.created[2]['Env'])
    assert a['HERMES_WEBUI_PORT'] != b['HERMES_WEBUI_PORT']
    assert a['HERMES_WEBUI_PASSWORD'] != b['HERMES_WEBUI_PASSWORD']
    assert a['HERMES_WEBUI_COOKIE_NAME'] != b['HERMES_WEBUI_COOKIE_NAME']
    assert a['HERMES_WEBUI_PASSWORD'] != alice['hermes_api_key']
    assert a['HERMES_WEBUI_GATEWAY_API_KEY'] == alice['hermes_api_key']
    assert a['HERMES_WEBUI_GATEWAY_BASE_URL'] == alice['hermes_url']
    assert a['HERMES_WEBUI_SECURE'] == '1'
    assert a['HERMES_WEBUI_CHAT_BACKEND'] == 'gateway'
    password = (p.state_dir / ALICE['id'] / 'webui-password')
    assert password.stat().st_mode & 0o777 == 0o600
    asyncio.run(p.ensure(ALICE))
    assert len(docker.created) == 3


def test_webui_password_symlink_fails_closed(boundary, monkeypatch):
    p, docker, _ = boundary
    asyncio.run(p.ensure(ALICE))
    private, _, _ = p._paths(ALICE)
    (private / 'webui-password').symlink_to(private / 'api-key')
    p.webui_enabled = True
    p.webui_host = '127.0.0.1'
    with pytest.raises(ProvisioningError, match='symlink'):
        asyncio.run(p.ensure(ALICE))
    assert len(docker.created) == 1


@pytest.mark.parametrize('password', ['x', 'short passphrase'])
def test_webui_accepts_user_chosen_short_password(boundary, monkeypatch, password):
    p, docker, _ = boundary
    asyncio.run(p.ensure(ALICE))
    private, _, _ = p._paths(ALICE)
    (private / 'webui-password').write_text(password + '\n')
    p.webui_enabled = True
    p.webui_host = '127.0.0.1'
    async def healthy(connection):
        return True
    monkeypatch.setattr(p, '_healthy', healthy)
    asyncio.run(p.ensure(ALICE))
    env = dict(entry.split('=', 1) for entry in docker.created[-1]['Env'])
    assert env['HERMES_WEBUI_PASSWORD'] == password


@pytest.mark.parametrize('password', ['', '   ', 'first\nsecond'])
def test_webui_rejects_blank_or_multiline_password_before_recreation(boundary, password):
    p, docker, _ = boundary
    asyncio.run(p.ensure(ALICE))
    private, _, _ = p._paths(ALICE)
    (private / 'webui-password').write_text(password)
    p.webui_enabled = True
    p.webui_host = '127.0.0.1'
    with pytest.raises(ProvisioningError, match='nonempty and single-line'):
        asyncio.run(p.ensure(ALICE))
    assert not docker.deleted