from __future__ import annotations

import asyncio
from collections.abc import Generator
from pathlib import Path
from unittest.mock import patch

import httpx
import pygit2
import pytest
from httpx import ASGITransport

from pygitweb.change_queue import CHANGE_QUEUE
from pygitweb.config import settings
from pygitweb.dependencies import is_loopback_host
from pygitweb.main import app


def _create_initial_commit(repo: pygit2.Repository, repo_dir: Path) -> None:
    (repo_dir / "README.md").write_text("seed\n", encoding="utf-8")
    index = repo.index
    index.add("README.md")
    index.write()
    tree = index.write_tree()
    sig = pygit2.Signature("tester", "tester@example.com")
    repo.create_commit("HEAD", sig, sig, "initial", tree, [])


@pytest.fixture(scope="module")
def updates_env(tmp_path_factory: pytest.TempPathFactory) -> Generator[dict[str, str], None, None]:
    root: Path = tmp_path_factory.mktemp("updates-routes")
    alpha_dir: Path = root / "alpha"
    alpha_dir.mkdir()
    _create_initial_commit(pygit2.init_repository(str(alpha_dir), bare=False), alpha_dir)
    beta_dir: Path = root / "group" / "beta"
    beta_dir.mkdir(parents=True)
    _create_initial_commit(pygit2.init_repository(str(beta_dir), bare=False), beta_dir)
    with (
        patch.object(settings, "PROJECTROOT", str(root)),
        patch.object(settings, "PROJECTS_LIST", str(root)),
        patch.object(settings, "PROJECT_MAXDEPTH", 3),
        patch.object(settings, "STRICT_EXPORT", False),
        patch.object(settings, "EXPORT_OK", ""),
        patch.object(settings, "LIST_ALL", True),
        patch.object(settings, "AUTH", False),
        patch.object(settings, "MAXLOAD", None),
    ):
        yield {"root": str(root), "alpha": "alpha", "beta": "group/beta"}


def _client() -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test")


class TestUpdatesLongPoll:
    def test_updates_returns_200_empty_after_timeout(self, updates_env: dict[str, str]) -> None:
        async def scenario() -> tuple[int, bytes]:
            async with _client() as client:
                with patch.object(CHANGE_QUEUE, "DEFAULT_TIMEOUT_SECONDS", 0.05):
                    resp = await client.get(
                        f"/project/{updates_env['alpha']}",
                        params={"a": "log", "updates": "true"},
                    )
                return resp.status_code, resp.content

        status, body = asyncio.run(scenario())
        assert status == 200
        assert body == b""

    def test_updates_returns_200_empty_when_shutting_down(self, updates_env: dict[str, str]) -> None:
        async def scenario() -> tuple[int, bytes]:
            async with _client() as client:
                app.state.shutting_down = True
                try:
                    resp = await client.get(
                        f"/project/{updates_env['alpha']}",
                        params={"a": "log", "updates": "true"},
                    )
                    return resp.status_code, resp.content
                finally:
                    app.state.shutting_down = False

        status, body = asyncio.run(scenario())
        assert status == 200
        assert body == b""

    def test_updates_returns_data_after_notify(self, updates_env: dict[str, str]) -> None:
        async def scenario() -> tuple[int, str]:
            async with _client() as client:
                poll: asyncio.Task[httpx.Response] = asyncio.create_task(
                    client.get(
                        f"/project/{updates_env['alpha']}",
                        params={"a": "log", "updates": "true"},
                    )
                )
                await asyncio.sleep(0.05)
                notify_resp = await client.post(
                    "/_internal/notify",
                    params={"project": updates_env["alpha"]},
                )
                assert notify_resp.status_code == 200
                assert notify_resp.json()["waiters_woken"] == 1
                resp: httpx.Response = await poll
                return resp.status_code, resp.text

        status, body = asyncio.run(scenario())
        assert status == 200
        assert "Log" in body

    def test_updates_with_pf_wakes_on_any_matching_project(self, updates_env: dict[str, str]) -> None:
        async def scenario() -> int:
            async with _client() as client:
                poll: asyncio.Task[httpx.Response] = asyncio.create_task(
                    client.get(
                        f"/project/{updates_env['alpha']}",
                        params={"a": "heads", "updates": "true", "pf": "group"},
                    )
                )
                await asyncio.sleep(0.05)
                notify_resp = await client.post(
                    "/_internal/notify",
                    params={"project": updates_env["beta"]},
                )
                assert notify_resp.status_code == 200
                assert notify_resp.json()["waiters_woken"] == 1
                return (await poll).status_code

        assert asyncio.run(scenario()) == 200

    def test_updates_rejected_for_unsupported_action(self, updates_env: dict[str, str]) -> None:
        async def scenario() -> int:
            async with _client() as client:
                resp = await client.get(
                    f"/project/{updates_env['alpha']}",
                    params={"a": "tree", "updates": "true"},
                )
                return resp.status_code

        assert asyncio.run(scenario()) == 400

    def test_notify_unknown_project_returns_404(self, updates_env: dict[str, str]) -> None:
        async def scenario() -> int:
            async with _client() as client:
                resp = await client.post("/_internal/notify", params={"project": "no-such-project"})
                return resp.status_code

        assert asyncio.run(scenario()) == 404

    def test_notify_rejects_non_loopback_client(self, updates_env: dict[str, str]) -> None:
        del updates_env

        async def scenario() -> int:
            transport = ASGITransport(app=app, client=("203.0.113.1", 999))
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                resp = await client.post("/_internal/notify", params={"project": "alpha"})
                return resp.status_code

        assert asyncio.run(scenario()) == 403


class TestLoopbackHost:
    def test_is_loopback_host(self) -> None:
        assert is_loopback_host("127.0.0.1")
        assert is_loopback_host("::1")
        assert is_loopback_host("localhost")
        assert not is_loopback_host("203.0.113.1")
        assert not is_loopback_host(None)
        assert not is_loopback_host("")