import asyncio
import json
import os
import sys
from collections.abc import Generator
from pathlib import Path
from unittest.mock import AsyncMock, patch

import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient

from pygitweb.auth import ACCESS_TOKEN_COOKIE_NAME
from pygitweb.auth_config import AuthConfig, LocalUser, init_auth_config, write_auth_config
from pygitweb.auth_oauth import (
	OAUTH_STATE_COOKIE_NAME,
	_exchange_code,
	_fetch_github_profile,
	_fetch_google_profile,
	_fetch_profile,
	_seal_oauth_state,
	is_oauth_email_allowed,
)
from pygitweb.config import Settings, settings
from pygitweb.conftest import clear_client_cookies
from pygitweb.gravatar import gravatar_url
from pygitweb.main import app
from pygitweb.password_hash import hash_password, verify_password
from pygitweb.permissions import PERMISSION_ADD_PROJECTS, PermissionsMap
from pygitweb.sessions import get_session


class _DummyResponse:
	def __init__(self, payload: object, is_success: bool = True) -> None:
		self._payload = payload
		self.is_success = is_success

	def raise_for_status(self) -> None:
		return None

	def json(self) -> object:
		return self._payload


_LOCAL_ADMIN = AuthConfig(
	auth_mode="local",
	local_users=[LocalUser(user="admin", password="secret")],
)


@pytest.fixture
def client() -> Generator[TestClient, None, None]:
	with TestClient(app) as c:
		yield c


@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 test_token_disabled(client: TestClient) -> None:
	with patch.object(settings, "AUTH", False):
		r = client.post("/token", data={"username": "a", "password": "b"})
		assert r.status_code == 400
		assert r.json()["detail"] == "Authentication is disabled"


def test_token_success(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		r = client.post("/token", data={"username": "admin", "password": "secret"})
	assert r.status_code == 200
	data = r.json()
	assert data["token_type"] == "bearer"
	assert isinstance(data["access_token"], str)
	assert len(data["access_token"]) >= 32
	assert data["access_token"] != "admin"


def test_token_wrong_password(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		r = client.post("/token", data={"username": "admin", "password": "wrong"})
	assert r.status_code == 400


def test_users_me_with_bearer(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		tok = client.post("/token", data={"username": "admin", "password": "secret"}).json()["access_token"]
		r = client.get("/users/me", headers={"Authorization": f"Bearer {tok}"})
		assert r.status_code == 200
		assert r.json()["username"] == "admin"


def test_login_form_sets_cookie_and_users_me(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		r = client.post(
			"/login",
			data={"username": "admin", "password": "secret", "next": "/"},
			follow_redirects=False,
		)
		assert r.status_code == 303
		r2 = client.get("/users/me")
		assert r2.status_code == 200
		assert r2.json()["username"] == "admin"


def test_logout_revokes_cookie_session(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		client.post(
			"/login",
			data={"username": "admin", "password": "secret", "next": "/"},
			follow_redirects=False,
		)
		assert client.get("/users/me").status_code == 200
		r_out = client.get("/logout", params={"next": "/"}, follow_redirects=False)
		assert r_out.status_code == 303
		assert client.get("/users/me").status_code == 401


def test_logout_with_bearer_revokes_token(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		tok = client.post("/token", data={"username": "admin", "password": "secret"}).json()["access_token"]
		headers = {"Authorization": f"Bearer {tok}"}
		assert client.get("/users/me", headers=headers).status_code == 200
		r_out = client.get("/logout", params={"next": "/"}, headers=headers, follow_redirects=False)
		assert r_out.status_code == 303
		assert client.get("/users/me", headers=headers).status_code == 401


def test_projectnamevalid_no_auth_required_when_auth_disabled(client: TestClient) -> None:
	with patch.object(settings, "AUTH", False), patch("pygitweb.main.project_visible_in_list", return_value=False):
		r = client.get("/projectnamevalid", params={"name": "newproj"})
	assert r.status_code == 200


def test_projectnamevalid_401_without_credentials_when_auth_enabled(
	client: TestClient, local_auth_config: AuthConfig
) -> None:
	with patch.object(settings, "AUTH", True):
		r = client.get("/projectnamevalid", params={"name": "newproj"})
	assert r.status_code == 401


def test_projectnamevalid_ok_with_bearer_when_auth_enabled(client: TestClient, local_auth_config: AuthConfig) -> None:
	cfg = local_auth_config.model_copy(
		update={
			"oauth_permissions": PermissionsMap.model_validate({
				"*": [],
				PERMISSION_ADD_PROJECTS: ["admin"],
			}),
		},
	)
	with (
		patch.object(settings, "AUTH", True),
		patch("pygitweb.main.project_visible_in_list", return_value=False),
		patch("pygitweb.auth_config.auth_config", cfg),
		patch("pygitweb.auth.auth_config", cfg),
		patch("pygitweb.main.auth_config", cfg),
	):
		tok = client.post("/token", data={"username": "admin", "password": "secret"}).json()["access_token"]
		r = client.get(
			"/projectnamevalid",
			params={"name": "newproj"},
			headers={"Authorization": f"Bearer {tok}"},
		)
	assert r.status_code == 200


def test_auth_config_init_creates_file_only_when_auth_enabled(tmp_path: Path) -> None:
	cfg_path = tmp_path / "auth.json"
	s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path))
	assert not cfg_path.is_file()
	cfg = init_auth_config(s)
	assert cfg_path.is_file()
	assert cfg.local_users[0].user == "admin"
	assert len(cfg.salt) == 16
	assert cfg.local_users[0].pass_hash.startswith("scrypt$")
	assert cfg.local_users[0].password is None
	on_disk = json.loads(cfg_path.read_text(encoding="utf-8"))
	assert len(on_disk["salt"]) == 16
	user_entry = on_disk["local_users"][0]
	assert "pass" not in user_entry
	assert "pass_hash" in user_entry
	if sys.platform != "win32":
		assert oct(os.stat(cfg_path).st_mode & 0o777) == "0o600"


def test_auth_config_no_file_when_auth_disabled(tmp_path: Path) -> None:
	from pygitweb.auth_config import init_auth_config
	from pygitweb.config import Settings

	cfg_path = tmp_path / "auth.json"
	s = Settings(AUTH=False, AUTH_CONFIG=str(cfg_path))
	init_auth_config(s)
	assert not cfg_path.is_file()


def test_auth_config_does_not_regenerate_missing_password(tmp_path: Path) -> None:
	cfg_path = tmp_path / "auth.json"
	write_auth_config(
		cfg_path,
		AuthConfig(auth_mode="local", local_users=[LocalUser(user="admin", password="")]),
	)
	s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path))
	cfg = init_auth_config(s)
	assert cfg.local_users[0].password == ""
	assert cfg.local_users[0].pass_hash == ""


def test_auth_config_adds_salt_to_existing_file(tmp_path: Path) -> None:
	cfg_path = tmp_path / "auth.json"
	write_auth_config(
		cfg_path,
		AuthConfig(
			auth_mode="local",
			local_users=[LocalUser(user="admin", pass_hash=hash_password("secret"))],
		),
	)
	s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path))
	cfg = init_auth_config(s)
	assert len(cfg.salt) == 16
	assert verify_password("secret", cfg.local_users[0].pass_hash)


def test_auth_config_migrates_plaintext_pass_to_pass_hash(tmp_path: Path) -> None:
	cfg_path = tmp_path / "auth.json"
	cfg_path.write_text(
		'{"auth_mode": "local", "local_users": [{"user": "admin", "pass": "secret1234"}]}\n',
		encoding="utf-8",
	)
	if sys.platform != "win32":
		os.chmod(cfg_path, 0o600)
	s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path))
	cfg = init_auth_config(s)
	assert cfg.local_users[0].password is None
	assert verify_password("secret1234", cfg.local_users[0].pass_hash, config_salt=cfg.salt)
	on_disk = cfg_path.read_text(encoding="utf-8")
	root = json.loads(on_disk)
	assert len(root["salt"]) == 16
	user_entry = root["local_users"][0]
	assert "pass" not in user_entry
	assert "pass_hash" in user_entry


@pytest.mark.skipif(sys.platform == "win32", reason="Unix file modes")
def test_auth_config_rejects_insecure_permissions(tmp_path: Path) -> None:
	from pygitweb.auth_config import AuthConfigPermissionError, init_auth_config, write_auth_config
	from pygitweb.config import Settings

	cfg_path = tmp_path / "auth.json"
	write_auth_config(
		cfg_path,
		AuthConfig(auth_mode="local", local_users=[LocalUser(user="admin", password="secret")]),
	)
	os.chmod(cfg_path, 0o644)
	s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path))
	with pytest.raises(AuthConfigPermissionError, match="0600"):
		init_auth_config(s)


def test_local_login_stores_session_audit_fields(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		client.post(
			"/login",
			data={"username": "admin", "password": "secret", "next": "/"},
			follow_redirects=False,
		)
		tok = client.post("/token", data={"username": "admin", "password": "secret"}).json()["access_token"]
	rec = get_session(tok)
	assert rec is not None
	assert rec.username == "admin"
	assert rec.auth_method == "local"
	assert rec.subject is None
	assert rec.email is None


def test_oauth_allowed_emails_glob() -> None:
	cfg = AuthConfig(auth_mode="oauth", oauth_allowed_emails=["*@example.com", "guest@corp.org"])
	assert is_oauth_email_allowed(cfg, "alice@example.com")
	assert is_oauth_email_allowed(cfg, "Alice@Example.COM")
	assert is_oauth_email_allowed(cfg, "guest@corp.org")
	assert not is_oauth_email_allowed(cfg, "alice@other.org")
	assert not is_oauth_email_allowed(cfg, None)


def test_oauth_allowed_emails_exact_match() -> None:
	cfg = AuthConfig(auth_mode="oauth", oauth_allowed_emails=["only@example.com"])
	assert is_oauth_email_allowed(cfg, "only@example.com")
	assert not is_oauth_email_allowed(cfg, "other@example.com")


def test_anonymous_user_assumes_local_public_role_for_read_permissions() -> None:
	from pygitweb.auth import can_read_project

	cfg = AuthConfig(
		auth_mode="local",
		local_users=[
			LocalUser(
				user="public",
				password="secret",
				email="public@example.com",
				assume_anonymous=True,
			),
		],
		oauth_permissions=PermissionsMap.model_validate({
			"*": [],
			"pgw.read.secret": ["public@example.com"],
		}),
	)
	with (
		patch.object(settings, "AUTH", True),
		patch("pygitweb.auth.auth_config", cfg),
	):
		assert can_read_project("secret", None)
		assert not can_read_project("other", None)


_OAUTH_CFG = AuthConfig(
	auth_mode="oauth",
	oauth_provider="google",
	oauth_client_id="test-client-id",
	oauth_client_secret="test-client-secret",
	oauth_redirect_uri="http://testserver/auth/oauth/callback",
	oauth_allowed_emails=[],
	local_users=[],
)


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


def test_oauth_start_redirects_to_provider(client: TestClient, oauth_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		r = client.get("/auth/oauth/start", params={"next": "/boards"}, follow_redirects=False)
	assert r.status_code == 303
	assert "accounts.google.com" in r.headers["location"]
	assert OAUTH_STATE_COOKIE_NAME in r.cookies


def test_oauth_callback_creates_session(client: TestClient, oauth_auth_config: AuthConfig) -> None:
	sealed = _seal_oauth_state(
		{"state": "idp-state", "next": "/", "code_verifier": "verifier"},
		oauth_auth_config.oauth_client_secret,
	)
	token_payload = {"access_token": "provider-token"}
	profile = {"sub": "oauth-sub-1", "email": "user@example.com", "preferred_username": "oauthuser"}
	with (
		patch.object(settings, "AUTH", True),
		patch("pygitweb.auth_oauth._exchange_code", new_callable=AsyncMock, return_value=token_payload),
		patch("pygitweb.auth_oauth._fetch_profile", new_callable=AsyncMock, return_value=profile),
	):
		client.cookies.set(OAUTH_STATE_COOKIE_NAME, sealed)
		r = client.get(
			"/auth/oauth/callback",
			params={"code": "auth-code", "state": "idp-state"},
			follow_redirects=False,
		)
		assert r.status_code == 303
		rec = get_session(r.cookies["pygitweb_access_token"])
		assert rec is not None
		assert rec.username == "oauthuser"
		assert rec.subject == "oauthuser"
		assert rec.email == "user@example.com"
		assert rec.auth_method == "oauth"
		assert client.get("/users/me").json()["username"] == "oauthuser"
		status = client.get("/auth/status").json()
		assert status["username"] == "oauthuser"
		assert status["gravatar_url"] == gravatar_url("user@example.com")
		clear_client_cookies(client)


def test_auth_status_no_gravatar_for_local_login(client: TestClient, local_auth_config: AuthConfig) -> None:
	with patch.object(settings, "AUTH", True):
		client.post(
			"/login",
			data={"username": "admin", "password": "secret", "next": "/"},
			follow_redirects=False,
		)
		status = client.get("/auth/status").json()
	assert status["username"] == "admin"
	assert status["gravatar_url"] is None


def test_local_user_email_gravatar(client: TestClient) -> None:
	cfg = AuthConfig(
		auth_mode="local",
		local_users=[LocalUser(user="admin", password="secret", email="  admin@example.com  ")],
	)
	with (
		patch.object(settings, "AUTH", True),
		patch("pygitweb.auth_config.auth_config", cfg),
		patch("pygitweb.auth.auth_config", cfg),
		patch("pygitweb.main.auth_config", cfg),
	):
		client.post(
			"/login",
			data={"username": "admin", "password": "secret", "next": "/"},
			follow_redirects=False,
		)
		status = client.get("/auth/status").json()
	assert status["gravatar_url"] == gravatar_url("admin@example.com")
	rec = get_session(client.cookies[ACCESS_TOKEN_COOKIE_NAME])
	assert rec is not None
	assert rec.email == "admin@example.com"


def test_authenticate_local_user_disabled_in_oauth_only_mode() -> None:
	from pygitweb.auth import authenticate_local_user

	with patch("pygitweb.auth.auth_config", _OAUTH_CFG):
		assert authenticate_local_user("admin", "secret") is None


def test_authenticate_local_user_pass_hash() -> None:
	from pygitweb.auth import authenticate_local_user

	cfg = AuthConfig(
		auth_mode="local",
		salt="0123456789abcdef",
		local_users=[LocalUser(user="admin", pass_hash=hash_password("secret", config_salt="0123456789abcdef"))],
	)
	with patch("pygitweb.auth.auth_config", cfg):
		user = authenticate_local_user("admin", "secret")
	assert user is not None
	assert user.username == "admin"
	assert authenticate_local_user("admin", "wrong") is None


def test_exchange_code_success(oauth_auth_config: AuthConfig) -> None:
	resp = _DummyResponse({"access_token": "tok", "other": "x"})

	client = AsyncMock()
	client.post.return_value = resp
	client.__aenter__.return_value = client
	client.__aexit__.return_value = None

	async def runner() -> None:
		with patch("pygitweb.auth_oauth.httpx.AsyncClient", return_value=client):
			out = await _exchange_code(
				oauth_auth_config,
				code="auth-code",
				code_verifier="verifier",
			)
		assert out["access_token"] == "tok"

	asyncio.run(runner())


def test_exchange_code_rejects_non_mapping(oauth_auth_config: AuthConfig) -> None:
	resp = _DummyResponse(["not-a-dict"])

	client = AsyncMock()
	client.post.return_value = resp
	client.__aenter__.return_value = client
	client.__aexit__.return_value = None

	async def runner() -> None:
		with patch("pygitweb.auth_oauth.httpx.AsyncClient", return_value=client), pytest.raises(HTTPException):
			await _exchange_code(
				oauth_auth_config,
				code="auth-code",
				code_verifier="verifier",
			)

	asyncio.run(runner())


def test_fetch_google_profile_success(oauth_auth_config: AuthConfig) -> None:
	resp = _DummyResponse({"sub": "123", "email": "user@example.com"})

	client = AsyncMock()
	client.get.return_value = resp
	client.__aenter__.return_value = client
	client.__aexit__.return_value = None

	async def runner() -> None:
		with patch("pygitweb.auth_oauth.httpx.AsyncClient", return_value=client):
			profile = await _fetch_google_profile("access-token")
		assert profile["email"] == "user@example.com"

	asyncio.run(runner())


def test_fetch_google_profile_rejects_non_mapping(oauth_auth_config: AuthConfig) -> None:
	resp = _DummyResponse("not-a-dict")

	client = AsyncMock()
	client.get.return_value = resp
	client.__aenter__.return_value = client
	client.__aexit__.return_value = None

	async def runner() -> None:
		with patch("pygitweb.auth_oauth.httpx.AsyncClient", return_value=client), pytest.raises(HTTPException):
			await _fetch_google_profile("access-token")

	asyncio.run(runner())


def test_fetch_github_profile_uses_primary_verified_email() -> None:
	resp_user = _DummyResponse({"login": "octocat", "email": None})
	resp_emails = _DummyResponse([
		{"email": "secondary@example.com", "primary": False, "verified": True},
		{"email": "primary@example.com", "primary": True, "verified": True},
	])

	client = AsyncMock()
	client.get.side_effect = (resp_user, resp_emails)
	client.__aenter__.return_value = client
	client.__aexit__.return_value = None

	async def runner() -> None:
		with patch("pygitweb.auth_oauth.httpx.AsyncClient", return_value=client):
			profile = await _fetch_github_profile("access-token")
		assert profile["email"] == "primary@example.com"

	asyncio.run(runner())


def test_fetch_github_profile_rejects_non_mapping() -> None:
	resp_user = _DummyResponse("not-a-dict")

	client = AsyncMock()
	client.get.return_value = resp_user
	client.__aenter__.return_value = client
	client.__aexit__.return_value = None

	async def runner() -> None:
		with patch("pygitweb.auth_oauth.httpx.AsyncClient", return_value=client), pytest.raises(HTTPException):
			await _fetch_github_profile("access-token")

	asyncio.run(runner())


def test_fetch_profile_google(oauth_auth_config: AuthConfig) -> None:
	token_payload = {"access_token": "tok"}
	profile: dict[str, object] = {"sub": "123"}

	async def runner() -> None:
		with patch("pygitweb.auth_oauth._fetch_google_profile", new_callable=AsyncMock, return_value=profile):
			out = await _fetch_profile(oauth_auth_config, token_payload)
		assert out is profile

	asyncio.run(runner())


def test_fetch_profile_github() -> None:
	cfg = _OAUTH_CFG.model_copy(update={"oauth_provider": "github"})
	token_payload = {"access_token": "tok"}
	profile: dict[str, object] = {"id": "123"}

	async def runner() -> None:
		with patch("pygitweb.auth_oauth._fetch_github_profile", new_callable=AsyncMock, return_value=profile):
			out = await _fetch_profile(cfg, token_payload)
		assert out is profile

	asyncio.run(runner())


def test_fetch_profile_missing_access_token(oauth_auth_config: AuthConfig) -> None:
	async def runner() -> None:
		with pytest.raises(HTTPException):
			await _fetch_profile(oauth_auth_config, {})

	asyncio.run(runner())


def test_fetch_profile_unknown_provider() -> None:
	cfg = _OAUTH_CFG.model_copy(update={"oauth_provider": "other"})

	async def runner() -> None:
		with pytest.raises(HTTPException):
			await _fetch_profile(cfg, {"access_token": "tok"})

	asyncio.run(runner())