from __future__ import annotations

import gzip
import socket
import subprocess
import threading
import time
from collections.abc import Generator
from pathlib import Path
from unittest.mock import patch

import pygit2
import pytest
import uvicorn
from fastapi.testclient import TestClient

from pygitweb.auth_config import AuthConfig, LocalUser
from pygitweb.config import settings
from pygitweb.conftest import set_client_session
from pygitweb.main import app
from pygitweb.permissions import PermissionsMap
from pygitweb.smart_http import _decode_request_body, _pkt_flush, _pkt_line, _service_advertisement

_LOCAL_ADMIN = AuthConfig(
	auth_mode="local",
	local_users=[LocalUser(user="alice", password="secret")],
	oauth_permissions=PermissionsMap.default_access(),
)


@pytest.fixture
def local_auth_config() -> Generator[AuthConfig, None, None]:
	with (
		patch("pygitweb.auth_config.auth_config", _LOCAL_ADMIN),
		patch("pygitweb.auth.auth_config", _LOCAL_ADMIN),
		patch("pygitweb.main.auth_config", _LOCAL_ADMIN),
	):
		yield _LOCAL_ADMIN


def _run_git(repo_dir: Path, *args: str) -> str:
	completed = subprocess.run(
		[settings.GIT, *args],
		cwd=repo_dir,
		capture_output=True,
		text=True,
		check=True,
	)
	return completed.stdout.strip()


def _seed_bare_repo(root: Path, bare_dir: Path, message: str = "initial") -> None:
	work_dir = root / "seed-work"
	work_dir.mkdir()
	_run_git(work_dir, "init", "-b", "main")
	_run_git(work_dir, "config", "user.name", "tester")
	_run_git(work_dir, "config", "user.email", "tester@example.com")
	(work_dir / "README.md").write_text(f"{message}\n", encoding="utf-8")
	_run_git(work_dir, "add", "README.md")
	_run_git(work_dir, "commit", "-m", message)
	_run_git(work_dir, "remote", "add", "origin", str(bare_dir))
	_run_git(work_dir, "push", "-u", "origin", "main")


@pytest.fixture(scope="class")
def http_repo_env(tmp_path_factory: pytest.TempPathFactory) -> Generator[dict[str, str], None, None]:
	root = tmp_path_factory.mktemp("smart-http")
	repo_dir = root / "demo"
	pygit2.init_repository(str(repo_dir), bare=True)
	_seed_bare_repo(root, repo_dir)
	with (
		patch.object(settings, "PROJECTROOT", str(root)),
		patch.object(settings, "PROJECTS_LIST", str(root)),
		patch.object(settings, "PROJECT_MAXDEPTH", 2),
		patch.object(settings, "STRICT_EXPORT", False),
		patch.object(settings, "EXPORT_OK", ""),
		patch.object(settings, "LIST_ALL", True),
		patch.object(settings, "AUTH", False),
	):
		yield {"project": "demo", "repo_dir": str(repo_dir), "root": str(root)}


class TestSmartHttpHelpers:
	def test_pkt_line_and_flush(self) -> None:
		assert _pkt_line(b"hello\n") == b"000ahello\n"
		assert _pkt_flush() == b"0000"

	def test_decode_request_body(self) -> None:
		raw = b"0014command=ls-refs\n0000"
		assert _decode_request_body(raw, None) == raw
		assert _decode_request_body(raw, "identity") == raw
		assert _decode_request_body(gzip.compress(raw), "gzip") == raw

	def test_decode_request_body_rejects_invalid_gzip(self) -> None:
		with pytest.raises(Exception) as exc_info:
			_decode_request_body(b"not-gzip", "gzip")
		assert exc_info.value.status_code == 400

	def test_decode_request_body_rejects_unknown_encoding(self) -> None:
		with pytest.raises(Exception) as exc_info:
			_decode_request_body(b"payload", "br")
		assert exc_info.value.status_code == 415

	def test_service_advertisement_matches_git(self, http_repo_env: dict[str, str]) -> None:
		direct = subprocess.run(
			[settings.GIT, "upload-pack", "--advertise-refs", "."],
			cwd=http_repo_env["repo_dir"],
			capture_output=True,
			check=True,
		).stdout
		body = _service_advertisement("git-upload-pack", http_repo_env["project"])
		assert body.startswith(_pkt_line(b"# service=git-upload-pack\n"))
		assert body.endswith(direct)
		assert _pkt_flush() in body


class TestSmartHttpRoutes:
	def test_upload_pack_info_refs(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		resp = client.get(
			f"/http/{http_repo_env['project']}/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 200
		assert resp.headers["content-type"].startswith("application/x-git-upload-pack-advertisement")
		assert b"# service=git-upload-pack" in resp.content
		assert b"refs/heads/" in resp.content

	def test_accepts_git_suffix(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		resp = client.get(
			f"/http/{http_repo_env['project']}.git/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 200

	def test_unknown_project_404(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		resp = client.get(
			"/http/missing/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 404

	def test_upload_pack_accepts_gzip_body(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		request_body = b"0000"
		headers = {
			"Content-Type": "application/x-git-upload-pack-request",
			"Git-Protocol": "version=2",
		}
		plain = client.post(
			f"/http/{http_repo_env['project']}/git-upload-pack",
			content=request_body,
			headers=headers,
		)
		gzipped = client.post(
			f"/http/{http_repo_env['project']}/git-upload-pack",
			content=gzip.compress(request_body),
			headers={**headers, "Content-Encoding": "gzip"},
		)
		assert plain.status_code == gzipped.status_code
		assert plain.content == gzipped.content

	def test_receive_pack_requires_auth_when_enabled(
		self,
		http_repo_env: dict[str, str],
		local_auth_config: AuthConfig,
	) -> None:
		del local_auth_config
		client = TestClient(app)
		with patch.object(settings, "AUTH", True):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
			)
		assert resp.status_code == 401

	def test_receive_pack_allowed_when_authenticated(
		self,
		http_repo_env: dict[str, str],
		local_auth_config: AuthConfig,
	) -> None:
		del local_auth_config
		client = TestClient(app)
		with patch.object(settings, "AUTH", True):
			set_client_session(client, "alice")
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
			)
		assert resp.status_code == 200
		assert resp.headers["content-type"].startswith("application/x-git-receive-pack-advertisement")

	def test_clone_and_push_over_http(self, http_repo_env: dict[str, str]) -> None:
		with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
			sock.bind(("127.0.0.1", 0))
			port = sock.getsockname()[1]

		config = uvicorn.Config(
			app,
			host="127.0.0.1",
			port=port,
			log_level="error",
			ws="none",
		)
		server = uvicorn.Server(config)
		thread = threading.Thread(target=server.run, daemon=True)
		thread.start()
		deadline = time.time() + 5.0
		while not server.started and time.time() < deadline:
			time.sleep(0.01)
		assert server.started

		clone_dir = Path(http_repo_env["root"]) / "clone"
		remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git"
		try:
			_run_git(Path(http_repo_env["root"]), "clone", remote_url, str(clone_dir))
			_run_git(clone_dir, "config", "user.name", "tester")
			_run_git(clone_dir, "config", "user.email", "tester@example.com")
			(clone_dir / "new.txt").write_text("pushed\n", encoding="utf-8")
			_run_git(clone_dir, "add", "new.txt")
			_run_git(clone_dir, "commit", "-m", "add new")
			_run_git(clone_dir, "push", "origin", "HEAD")
			repo = pygit2.Repository(http_repo_env["repo_dir"])
			commit = repo.head.peel()
			assert commit.message.strip() == "add new"
			tree = commit.tree
			assert tree["new.txt"].data.decode("utf-8") == "pushed\n"
		finally:
			server.should_exit = True
			thread.join(timeout=5.0)