import io
import os
from pathlib import Path
import sys
import time
from types import SimpleNamespace
from unittest import mock

import pytest
import requests
from urllib3.response import HTTPResponse

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


@pytest.fixture
def transport(monkeypatch):
    state = SimpleNamespace(replies=[], calls=[], responses=[])

    def send(adapter, request, **kwargs):
        state.calls.append((request, kwargs))
        assert state.replies, 'unexpected offline HTTP request'
        reply = state.replies.pop(0)
        if isinstance(reply, Exception):
            raise reply
        status, headers, body = reply
        response = requests.Response()
        response.status_code = status
        response.headers.update(headers)
        response.url = request.url
        response.request = request
        response.raw = HTTPResponse(io.BytesIO(body), preload_content=False)
        state.responses.append(response)
        return response

    monkeypatch.setattr(requests.adapters.HTTPAdapter, 'send', send)
    for key in ('http_proxy', 'https_proxy', 'all_proxy', 'HTTP_PROXY', 'HTTPS_PROXY', 'ALL_PROXY'):
        monkeypatch.setenv(key, 'http://ambient.invalid:8080')
    monkeypatch.setenv('NO_PROXY', '')
    monkeypatch.setenv('no_proxy', '')
    monkeypatch.setenv('NETRC', str(Path(__file__).with_name('missing-netrc')))
    return state


@pytest.fixture
def proxy_file(tmp_path, monkeypatch):
    path = tmp_path / 'routing-proxies.txt'
    path.write_text('http://configured.invalid:8080\n', encoding='ascii')
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_enabled', True)
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_file', str(path))
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_max_retries', 2)
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_retry_delay', 0)
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_timeout', 99)
    for name, value in (('path', None), ('mtime', None), ('entries', []), ('index', 0)):
        monkeypatch.setattr(scanner, '_api_proxy_cache_' + name, value)
    return path


@pytest.mark.parametrize('scheme', ['http', 'https'])
@pytest.mark.parametrize('stream', [False, True])
def test_direct_ignores_ambient_and_caller_proxies_across_redirects(transport, proxy_file, scheme, stream):
    transport.replies = [
        (302, {'Location': 'https://second.invalid/payload'}, b''),
        (307, {'Location': 'http://third.invalid/payload'}, b''),
        (200, {}, b'complete artifact'),
    ]
    environment = dict(os.environ)
    with mock.patch.object(scanner, 'next_api_proxy', side_effect=AssertionError('direct selected proxy')):
        response = scanner.api_request(
            'GET', scheme + '://first.invalid/payload', use_proxy=False,
            timeout=(3, 7), stream=stream,
            proxies={'https://second.invalid': 'http://caller.invalid:8080'},
        )
    assert len(response.history) == 2
    if stream:
        assert not response._content_consumed
        assert not response.raw.closed
        assert b''.join(response.iter_content(3)) == b'complete artifact'
    else:
        assert response.content == b'complete artifact'
    response.close()
    for request, options in transport.calls:
        assert requests.utils.select_proxy(request.url, options['proxies']) is None
        assert options['proxies'] == {'no_proxy': '*'}
        assert options['timeout'] == (3, 7)
        assert 'Proxy-Authorization' not in request.headers
    assert dict(os.environ) == environment


@pytest.mark.parametrize('verify,expected', [(None, 'fixture-ca.pem'), (True, 'fixture-ca.pem'), (False, False), ('explicit.pem', 'explicit.pem')])
def test_direct_preserves_requests_certificate_merge(transport, monkeypatch, verify, expected):
    monkeypatch.setenv('REQUESTS_CA_BUNDLE', 'fixture-ca.pem')
    transport.replies = [(200, {}, b'ok')]
    with scanner._direct_request('GET', 'https://fixture.invalid', verify=verify, cert=('cert.pem', 'key.pem')):
        pass
    options = transport.calls[0][1]
    assert options['verify'] == expected
    assert options['cert'] == ('cert.pem', 'key.pem')


@pytest.mark.parametrize('missing', [False, True])
def test_configured_proxy_file_failure_never_falls_back_direct(transport, proxy_file, missing):
    if missing:
        proxy_file.unlink()
    else:
        proxy_file.write_text('# empty\n', encoding='ascii')
    with pytest.raises(scanner.ApiRequestError, match='no valid proxies'):
        scanner.api_request('GET', 'https://fixture.invalid')
    assert not transport.calls
    transport.replies = [(200, {}, b'direct')]
    with scanner.api_request('GET', 'https://fixture.invalid', use_proxy=False) as response:
        assert response.content == b'direct'


def test_proxy_transport_failure_retries_file_proxy_without_direct_fallback(transport, proxy_file):
    transport.replies = [requests.exceptions.ProxyError('offline fixture')] * 2
    with pytest.raises(scanner.ApiRequestError, match='after 2 attempt'):
        scanner.api_request('GET', 'https://fixture.invalid')
    assert len(transport.calls) == 2
    for request, options in transport.calls:
        assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'


def test_metadata_file_proxy_survives_worker_no_proxy_and_redirects(transport, proxy_file, monkeypatch):
    monkeypatch.setenv('NO_PROXY', '*')
    monkeypatch.setenv('no_proxy', '*')
    transport.replies = [
        (302, {'Location': 'https://second.invalid/metadata'}, b''),
        (200, {}, b'{}'),
    ]
    with scanner.api_request('GET', 'https://first.invalid/metadata') as response:
        assert response.content == b'{}'
    assert len(transport.calls) == 2
    for request, options in transport.calls:
        assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'


@pytest.mark.parametrize('timeout,expected', [
    (30, (5, 30)),
    (15, (5, 15)),
    (3, (3, 3)),
    ((2, 30), (2, 30)),
    ((12, 7), (5, 7)),
    ([9, 15], (5, 15)),
    ((None, 30), (5, 30)),
    ((2, None), (2, None)),
    (None, (5, 5)),
])
def test_proxy_connect_cap_keeps_caller_read_timeout(transport, proxy_file, monkeypatch, timeout, expected):
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_timeout', 5)
    transport.replies = [(200, {}, b'{}')]
    with scanner.api_request('GET', 'https://fixture.invalid/metadata', timeout=timeout):
        pass
    request, options = transport.calls[0]
    assert options['timeout'] == expected
    assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'


def test_proxy_retry_budget_is_shared_and_clamps_read_timeout(transport, proxy_file, monkeypatch):
    clock = [0.0]
    waits = []
    calls = []
    monkeypatch.setattr(scanner.scan_config, 'api_proxy_timeout', 5)
    monkeypatch.setattr(scanner.time, 'monotonic', lambda: clock[0])

    def request(*args, **kwargs):
        calls.append(kwargs)
        clock[0] += 7
        raise requests.exceptions.ReadTimeout('offline fixture')

    def wait(seconds):
        waits.append(seconds)
        clock[0] += seconds

    monkeypatch.setattr(scanner.requests, 'request', request)
    monkeypatch.setattr(scanner, '_wait_or_raise_scan_slot_fatal', wait)
    with pytest.raises(scanner.ApiRequestError, match='deadline expired during retry'):
        scanner.api_request(
            'GET', 'https://fixture.invalid/metadata', timeout=30,
            max_retries=3, retry_delay=5, deadline=20,
        )
    assert [call['timeout'] for call in calls] == [(5, 20), (5, 8)]
    assert all(call['proxies']['https'] == 'http://configured.invalid:8080' for call in calls)
    assert waits == [5]


def test_direct_retries_deadline_and_response_cleanup(transport, proxy_file):
    transport.replies = [(503, {}, b''), requests.exceptions.ConnectionError('offline'), (200, {}, b'ok')]
    with mock.patch.object(scanner, 'next_api_proxy', side_effect=AssertionError('direct selected proxy')):
        response = scanner.api_request(
            'GET', 'https://fixture.invalid', use_proxy=False, stream=True,
            timeout=(2, 4), max_retries=3, retry_delay=0, deadline=time.monotonic() + 1,
        )
    assert transport.responses[0].raw.closed
    assert not response.raw.closed
    response.close()
    assert len(transport.calls) == 3
    assert all(0 < value <= 1 for _, options in transport.calls for value in options['timeout'])


def test_direct_cancellation_closes_response_and_expired_deadline_sends_nothing(transport):
    transport.replies = [(200, {}, b'ok')]
    with mock.patch.object(scanner, '_raise_if_scan_slot_fatal', side_effect=[None, scanner.ScanSlotFatalError('offline')]), \
            mock.patch.object(scanner._scan_slot_fatal_event, 'is_set', return_value=True):
        with pytest.raises(scanner.ScanSlotFatalError):
            scanner.api_request('GET', 'https://fixture.invalid', use_proxy=False, stream=True)
    assert transport.responses[0].raw.closed
    with pytest.raises(scanner.ApiRequestError, match='deadline expired'):
        scanner.api_request('GET', 'https://fixture.invalid', use_proxy=False, deadline=time.monotonic() - 1)
    assert len(transport.calls) == 1


@pytest.mark.parametrize('blob', [False, True])
def test_registry_metadata_uses_file_proxy_but_blob_and_401_retry_are_direct(transport, proxy_file, monkeypatch, blob):
    challenge = 'Bearer realm="https://auth.docker.io/token",service="registry.docker.io"'
    transport.replies = [
        (401, {'WWW-Authenticate': challenge}, b''),
        (200, {}, b'{"token":"synthetic-token"}'),
        (200, {}, b'{"schemaVersion":2}'),
    ]
    monkeypatch.setattr(scanner, 'docker_token_manager', scanner.DockerTokenManager())
    monkeypatch.setattr(scanner, '_require_docker_blob_url', lambda *a, **k: None)
    deadline = time.monotonic() + 10
    digest = 'sha256:' + 'a' * 64
    if blob:
        response, _ = scanner._docker_registry_blob_response('fixture/repo', digest, None, deadline)
        assert not response.raw.closed
        response.close()
    else:
        scanner.docker_registry_manifest('fixture/repo', digest, deadline=deadline)
    assert len(transport.calls) == 3
    for index, (request, options) in enumerate(transport.calls):
        selected = requests.utils.select_proxy(request.url, options['proxies'])
        assert selected == (None if blob and index != 1 else 'http://configured.invalid:8080')
        assert options['stream'] is True


def test_github_pool_forwards_content_optout_but_metadata_keeps_proxy(transport, proxy_file):
    transport.replies = [(200, {}, b'{"size":2,"encoding":"base64","content":"e30="}'), (200, {}, b'[]')]
    pool = scanner.GitHubTokenPool(token='synthetic-token')
    assert scanner.github_content_bytes({'url': 'https://fixture.invalid/content'}, pool) == b'{}'
    with pool.request('GET', 'https://fixture.invalid/commits'):
        pass
    assert [requests.utils.select_proxy(req.url, options['proxies']) for req, options in transport.calls] == [None, 'http://configured.invalid:8080']


def test_ci_download_direct_redirects_strip_credentials(transport, proxy_file, tmp_path, monkeypatch):
    transport.replies = [
        (302, {'Location': 'https://other.invalid/artifact'}, b''),
        (200, {}, b'zip fixture'),
    ]
    monkeypatch.setattr(scanner, 'require_private_directory', lambda *a, **k: None)
    monkeypatch.setattr(scanner, 'reject_reparse_components', lambda *a: None)
    monkeypatch.setattr(scanner, 'harden_private_file', lambda *a: None)
    monkeypatch.setattr(scanner, 'private_file_ready', lambda *a: True)
    monkeypatch.setattr(scanner, 'durable_replace', os.replace)
    destination = tmp_path / 'artifact'
    outcome = scanner.download_to_file(
        'https://fixture.invalid/artifact', str(destination), timeout=7,
        headers={'Authorization': 'synthetic-token', 'Private-Token': 'synthetic-token'},
    )
    assert outcome.ok and destination.read_bytes() == b'zip fixture'
    assert len(transport.calls) == 2
    for request, options in transport.calls:
        assert requests.utils.select_proxy(request.url, options['proxies']) is None
        assert options['timeout'] == 7
    assert 'Authorization' not in transport.calls[1][0].headers
    assert 'Private-Token' not in transport.calls[1][0].headers
    assert all(response.raw.closed for response in transport.responses)


@pytest.mark.parametrize('wrapper', [scanner.github_api_get, scanner.gitlab_api_get])
def test_ci_metadata_wrappers_keep_configured_proxy_even_when_streamed(transport, proxy_file, wrapper):
    transport.replies = [(200, {}, b'{}')]
    with wrapper('https://fixture.invalid/metadata', stream=True):
        pass
    request, options = transport.calls[0]
    assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'
    assert options['stream'] is True


def test_pypi_bulk_index_is_direct_without_touching_a_database(transport, proxy_file, tmp_path, monkeypatch):
    path = tmp_path / 'mock-index'
    path.touch()
    connection = mock.Mock()
    connection.execute.return_value.fetchone.side_effect = [None, (1,)]
    monkeypatch.setattr(scanner, '_pypi_index_path', lambda: str(path))
    monkeypatch.setattr(scanner.sqlite3, 'connect', mock.Mock(return_value=connection))
    monkeypatch.setattr(scanner, 'harden_private_file', lambda *a: None)
    transport.replies = [(200, {}, b'<a href="/simple/fixture/">fixture</a>')]
    assert scanner.load_pypi_project_index(request_timeout=7) == str(path)
    request, options = transport.calls[0]
    assert requests.utils.select_proxy(request.url, options['proxies']) is None
    assert options['timeout'] == 7 and options['stream'] is True
    assert transport.responses[0].raw.closed
    connection.close.assert_called_once()