import os
from pathlib import Path
import sys
import tempfile
from unittest import mock

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'app'))
import janitor
import scanner
from test_docker_staging_bounds import command_harness


REAL_SHARED_OWNERS = scanner._shared_staging_owners
REAL_WRITE_OWNER = scanner.write_temp_owner
REAL_CLEANUP = scanner.cleanup_command_work_dir
REAL_ATOMIC_WRITE = scanner.atomic_write_private_json


@pytest.fixture
def owned_runner(command_harness, monkeypatch):
    state = command_harness
    state.work_root = state.command_dir.parent
    state.command_dir = state.work_root / 'trufflehog-run-command'
    state.blobs = state.work_root / 'docker-layer-shared'
    for path in (state.work_root, state.blobs):
        scanner.ensure_private_directory(str(path))
    monkeypatch.setattr(scanner, '_runtime_initialized', True)
    monkeypatch.setattr(scanner.scan_config, 'work_dir', str(state.work_root))
    monkeypatch.setattr(scanner, '_shared_staging_owners', REAL_SHARED_OWNERS)
    monkeypatch.setattr(scanner, 'write_temp_owner', REAL_WRITE_OWNER)
    monkeypatch.setattr(scanner, 'require_git_clone_launch_authority', lambda cmd: None)
    state.parent = scanner.serialize_process_identity(scanner.current_process_identity())
    REAL_WRITE_OWNER(str(state.blobs), required=True)
    state.cleanups = mock.Mock(wraps=REAL_CLEANUP)
    monkeypatch.setattr(scanner, 'cleanup_command_work_dir', state.cleanups)

    def create_command_dir():
        scanner.ensure_private_directory(str(state.command_dir))
        REAL_WRITE_OWNER(str(state.command_dir), required=True)
        return str(state.command_dir)

    monkeypatch.setattr(scanner, 'create_command_work_dir', create_command_dir)
    original_launch = scanner.OwnedProcess
    state.identities = []
    state.pending_at_launch = []

    def launch(command, **options):
        state.pending_at_launch.append(all(
            scanner.read_private_json(str(root / scanner.TEMP_OWNER_FILE)).get('child_pid', 'absent') is None
            for root in (state.command_dir, state.blobs)
        ))
        process = original_launch(command, **options)
        identity = {'pid': 41001 + len(state.identities), 'creation_time': f'fixture-{len(state.identities)}',
                    'executable': state.parent['executable']}
        state.identities.append(identity)
        process.pid = identity['pid']
        process.payload_identity = identity
        process.returncode = None
        state.children = {process.pid}
        return process

    monkeypatch.setattr(scanner, 'OwnedProcess', launch)

    def identity_state(pid, created, executable):
        if pid is None or not created or not executable:
            return 'unknown'
        return 'alive' if pid in state.children or pid == state.parent['pid'] else 'dead'

    monkeypatch.setattr(scanner, 'exact_process_identity_state', identity_state)
    state.marker = lambda root=state.blobs: scanner.read_private_json(str(root / scanner.TEMP_OWNER_FILE))
    state.command = lambda: ['fixture', 'clone', '--no-checkout', '--no-recurse-submodules', '--',
                             'https://fixture.invalid/repo.git', str(state.blobs / 'repo')]
    return state


@pytest.mark.parametrize('native', [False, True])
def test_retained_roots_cannot_be_reaped_with_dead_scanner_and_live_child(owned_runner, monkeypatch, native):
    state = owned_runner
    state.unkillable = True
    checks = ['', 'TruffleHog staging limit exceeded'] if native else ['TruffleHog staging limit exceeded']
    monkeypatch.setattr(scanner, '_check_command_staging', mock.Mock(side_effect=checks))
    with pytest.raises(scanner.ScanSlotFatalError):
        with scanner.run_command_streamed(state.command(), 30, staging_roots=(str(state.blobs),), native_git_clone=native):
            pass
    state.cleanups.assert_not_called()
    state.slot.release.assert_not_called()
    assert state.pending_at_launch == [True]
    marker = state.marker()
    assert marker['owner_pid'] == marker['parent_pid'] == state.parent['pid']
    assert marker['child_pid'] == state.identities[0]['pid']
    monkeypatch.setattr(janitor, 'exact_process_identity_state',
                        lambda pid, *_: 'alive' if pid in state.children else 'dead')
    valid, reason = janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                            [state.parent['executable']], 0)
    assert not valid and reason == 'child_live_or_unknown'
    report = janitor.run_janitor_pass(str(state.work_root), [state.parent['executable']], minimum_age_sec=0)
    assert report['considered'] >= 2
    assert report['removed'] == 0
    assert state.blobs.is_dir() and state.command_dir.is_dir()
    # Direct shared-root callers (including Docker cleanup) cannot remove it either.
    REAL_CLEANUP(str(state.blobs))
    assert state.blobs.is_dir() and (state.blobs / scanner.TEMP_OWNER_FILE).is_file()
    monkeypatch.setattr(janitor, 'exact_process_identity_state', lambda *_: 'dead')
    assert janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                  [state.parent['executable']], 0) == (True, 'eligible')


def test_sequential_children_refresh_identity_and_restore_parent_lifetime(owned_runner, monkeypatch):
    state = owned_runner
    active = []

    def while_active():
        marker = state.marker()
        active.append((marker['child_pid'], marker['child_creation_time']))
        with mock.patch.object(janitor, 'exact_process_identity_state',
                               side_effect=lambda pid, *_: 'alive' if pid in state.children else 'dead'):
            assert not janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                               [state.parent['executable']], 0)[0]
        state.completed = True

    state.on_wait = while_active
    for _ in range(2):
        state.completed = False
        with scanner.run_command_streamed(['fixture'], 30, staging_roots=(str(state.blobs),)):
            pass
        assert state.blobs.is_dir()
        marker = state.marker()
        assert marker['owner_pid'] == state.parent['pid']
        assert 'child_pid' not in marker
        with mock.patch.object(janitor, 'exact_process_identity_state', return_value='alive'):
            assert not janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                               [state.parent['executable']], 0)[0]
    assert len(set(active)) == 2
    assert state.pending_at_launch == [True, True]
    REAL_CLEANUP(str(state.blobs))
    assert not state.blobs.exists()


@pytest.mark.parametrize('publication', ['command', 'shared'])
def test_failed_child_publication_leaves_pending_marker_when_termination_is_unknown(owned_runner, monkeypatch, publication):
    state = owned_runner
    state.unkillable = True

    def write(path, value):
        command_failure = publication == 'command' and value.get('owner_pid') in state.children
        shared_failure = publication == 'shared' and value.get('child_pid') in state.children
        if command_failure or shared_failure:
            raise OSError('synthetic publication failure')
        return REAL_ATOMIC_WRITE(path, value)

    monkeypatch.setattr(scanner, 'atomic_write_private_json', write)
    with pytest.raises(scanner.ScanSlotFatalError):
        with scanner.run_command_streamed(['fixture'], 30, staging_roots=(str(state.blobs),)):
            pass
    state.cleanups.assert_not_called()
    marker = state.marker()
    assert 'child_pid' in marker and marker['child_pid'] is None
    with mock.patch.object(janitor, 'exact_process_identity_state', return_value='dead'):
        assert not janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                           [state.parent['executable']], 0)[0]
    REAL_CLEANUP(str(state.blobs))
    assert state.blobs.is_dir()


def test_restore_failure_does_not_leak_slot_or_delete_unconfirmed_marker(owned_runner, monkeypatch):
    state = owned_runner
    state.completed = True

    def write(path, value):
        if (state.identities and scanner.canonical_path(path) == scanner.canonical_path(state.blobs / scanner.TEMP_OWNER_FILE)
                and 'child_pid' not in value):
            raise OSError('synthetic restore failure')
        return REAL_ATOMIC_WRITE(path, value)

    monkeypatch.setattr(scanner, 'atomic_write_private_json', write)
    with pytest.raises(RuntimeError, match='restore shared staging ownership'):
        with scanner.run_command_streamed(['fixture'], 30, staging_roots=(str(state.blobs),)):
            pass
    state.slot.release.assert_called_once()
    assert state.marker()['child_pid'] == state.identities[0]['pid']
    assert state.blobs.is_dir()


@pytest.mark.parametrize('state_value', ['alive', 'unknown', 'reused'])
def test_janitor_never_accepts_unconfirmed_recorded_child(owned_runner, monkeypatch, state_value):
    state = owned_runner
    marker = dict(state.marker(), child_pid=41001, child_creation_time='child', child_executable=state.parent['executable'])
    monkeypatch.setattr(janitor, 'exact_process_identity_state',
                        lambda pid, *_: state_value if pid == 41001 else 'dead')
    assert janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                  [state.parent['executable']], 0) == (False, 'child_live_or_unknown')


def test_child_executable_must_remain_authorized(owned_runner, monkeypatch):
    state = owned_runner
    marker = dict(state.marker(), child_pid=41001, child_creation_time='child',
                  child_executable=str(state.work_root / 'unapproved.exe'))
    monkeypatch.setattr(janitor, 'exact_process_identity_state', lambda *_: 'dead')
    assert janitor.validate_marker(str(state.work_root), str(state.blobs), marker,
                                  [state.parent['executable']], 0) == (False, 'child_executable_unapproved')


@pytest.mark.parametrize('change', [
    {'schema': 99}, {'relative_path': 'other'}, {'parent_pid': 123}, {'owner_pid': 123},
    {'child_pid': None}, {'child_pid': 41001, 'child_creation_time': 'child', 'child_executable': sys.executable},
])
def test_shared_root_validation_rejects_untrusted_or_busy_markers(owned_runner, monkeypatch, change):
    state = owned_runner
    marker_path = str(state.blobs / scanner.TEMP_OWNER_FILE)
    REAL_ATOMIC_WRITE(marker_path, dict(state.marker(), **change))
    state.children = {41001}
    write = mock.Mock(side_effect=AssertionError('validation must not write'))
    monkeypatch.setattr(scanner, 'atomic_write_private_json', write)
    with pytest.raises(RuntimeError):
        REAL_SHARED_OWNERS((str(state.blobs),))
    write.assert_not_called()


def test_outside_and_unmarked_roots_are_not_adopted(owned_runner):
    state = owned_runner
    with pytest.raises(RuntimeError, match='escapes'):
        REAL_SHARED_OWNERS((str(state.work_root),))
    unmarked = state.work_root / 'unmarked'
    scanner.ensure_private_directory(str(unmarked))
    with pytest.raises(RuntimeError, match='no authenticated owner'):
        REAL_SHARED_OWNERS((str(unmarked),))
    with tempfile.TemporaryDirectory(dir=state.work_root.parent) as outside:
        scanner.harden_private_directory(outside)
        with pytest.raises(RuntimeError, match='escapes'):
            REAL_SHARED_OWNERS((outside,))


@pytest.mark.parametrize('guard', ['require_private_directory', 'require_private_file', 'read_private_json'])
def test_private_directory_marker_and_json_checks_fail_closed(owned_runner, monkeypatch, guard):
    state = owned_runner
    monkeypatch.setattr(scanner, guard, mock.Mock(side_effect=OSError('synthetic private-file rejection')))
    writer = mock.Mock(side_effect=AssertionError('untrusted root must not be marked'))
    monkeypatch.setattr(scanner, 'atomic_write_private_json', writer)
    with pytest.raises((OSError, RuntimeError)):
        REAL_SHARED_OWNERS((str(state.blobs),))
    writer.assert_not_called()


def test_nested_staging_resolves_only_its_existing_owned_parent(owned_runner):
    state = owned_runner
    nested = state.blobs / 'nested'
    scanner.ensure_private_directory(str(nested))
    owners = REAL_SHARED_OWNERS((str(nested), str(state.blobs)))
    assert len(owners) == 1
    assert owners[0][0] == scanner.canonical_path(state.blobs)
    assert not (nested / scanner.TEMP_OWNER_FILE).exists()


@pytest.mark.parametrize('relative_parent', ['', 'tmp'])
def test_actual_docker_layer_prefix_is_enumerated_but_reaped_only_after_death_and_age(owned_runner, monkeypatch, relative_parent):
    state = owned_runner
    parent = state.work_root / relative_parent
    scanner.ensure_private_directory(str(parent))
    candidate = parent / 'docker-layer-aged'
    scanner.ensure_private_directory(str(candidate))
    REAL_WRITE_OWNER(str(candidate), required=True)
    marker_path = str(candidate / scanner.TEMP_OWNER_FILE)
    marker = dict(scanner.read_private_json(marker_path), child_pid=41001,
                  child_creation_time='synthetic-child', child_executable=state.parent['executable'])
    REAL_ATOMIC_WRITE(marker_path, marker)
    payload = candidate / 'blob-0000'
    payload.write_bytes(b'synthetic retained payload')
    scanner.harden_private_file(str(payload))
    cursor = janitor._MemoryCursorStore()
    try:
        candidates = list(janitor.iter_candidates(str(state.work_root), janitor.JanitorBudget(), cursor))
    finally:
        cursor.close()
    assert scanner.canonical_path(candidate) in {scanner.canonical_path(row[3]) for row in candidates}
    monkeypatch.setattr(janitor, 'exact_process_identity_state', lambda *_: 'dead')
    assert janitor.validate_marker(str(state.work_root), str(candidate), marker,
                                   [state.parent['executable']], 3600) == (False, 'too_young')
    young = janitor.run_janitor_pass(str(state.work_root), [state.parent['executable']], minimum_age_sec=3600)
    assert young['considered'] >= 1 and young['removed'] == 0 and payload.is_file()
    marker['created_at'] = '2020-01-01T00:00:00+00:00'
    REAL_ATOMIC_WRITE(marker_path, marker)
    monkeypatch.setattr(janitor, 'exact_process_identity_state', lambda pid, *_: 'alive' if pid == 41001 else 'dead')
    live = janitor.run_janitor_pass(str(state.work_root), [state.parent['executable']], minimum_age_sec=3600)
    assert live['considered'] >= 1 and live['removed'] == 0 and payload.is_file()
    monkeypatch.setattr(janitor, 'exact_process_identity_state', lambda *_: 'dead')
    assert janitor.validate_marker(str(state.work_root), str(candidate), marker,
                                   [state.parent['executable']], 3600) == (True, 'eligible')
    dead = janitor.run_janitor_pass(str(state.work_root), [state.parent['executable']], minimum_age_sec=3600)
    assert dead['removed'] == 1 and not candidate.exists()


@pytest.mark.parametrize('relative', ['docker-layerish-unapproved', 'hg/docker-layer-wrong-layout'])
def test_docker_prefix_expansion_does_not_admit_other_names_or_layouts(owned_runner, monkeypatch, relative):
    state = owned_runner
    candidate = state.work_root / relative
    scanner.ensure_private_directory(str(candidate.parent))
    scanner.ensure_private_directory(str(candidate))
    REAL_WRITE_OWNER(str(candidate), required=True)
    marker_path = str(candidate / scanner.TEMP_OWNER_FILE)
    marker = scanner.read_private_json(marker_path)
    marker['created_at'] = '2020-01-01T00:00:00+00:00'
    REAL_ATOMIC_WRITE(marker_path, marker)
    monkeypatch.setattr(janitor, 'exact_process_identity_state', lambda *_: 'dead')
    report = janitor.run_janitor_pass(str(state.work_root), [state.parent['executable']], minimum_age_sec=3600)
    assert report['removed'] == 0 and candidate.is_dir()