import unittest
from pathlib import Path
import sys
from unittest.mock import AsyncMock, MagicMock, patch
import os
import base64
import json
import httpx

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))

from gitea_handler import handle_list_repo_files, _validate_read_branch_ref, handle_get_file_content


class TestValidateReadBranchRef(unittest.TestCase):
    def test_accepted_refs(self):
        accepted = [
            "main",
            "feat/opax-domain-decouple",
            "docs/mcp-diagnostics",
            "fix/example",
            "chore/example",
            "release/1.2.3",
        ]
        for ref in accepted:
            with self.subTest(ref=ref):
                self.assertEqual(_validate_read_branch_ref(ref), ref)

    def test_rejected_refs(self):
        rejected = [
            "",
            " main",
            "main ",
            "feature branch",
            "main\tbranch",
            "main\nbranch",
            "main\x00branch",
            "main\x7fbranch",
            "HEAD",
            "head",
            "refs/heads/main",
            "/main",
            "main/",
            "main//next",
            "main/./next",
            "main/../next",
            "main\\next",
            "main?x=1",
            "main#fragment",
            "https://example.invalid",
            "user@host",
            "feature*",
            "feature~1",
            "feature^",
            "feature{a}",
            "feature;cmd",
            "feature|cmd",
            "feature%2Fbranch",
            "feature'quote",
            'feature"quote',
            "feature[abc]",
        ]
        for ref in rejected:
            with self.subTest(ref=ref):
                with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
                    _validate_read_branch_ref(ref)

class TestGiteaHandler(unittest.TestCase):

    def setUp(self):
        self.original_gitea_url = os.environ.get("GITEA_URL")
        self.original_gitea_token = os.environ.get("GITEA_TOKEN")
        os.environ["GITEA_URL"] = "https://gitea.example.com"
        os.environ["GITEA_TOKEN"] = "test-token"

    def tearDown(self):
        if self.original_gitea_url is not None:
            os.environ["GITEA_URL"] = self.original_gitea_url
        else:
            del os.environ["GITEA_URL"]
        if self.original_gitea_token is not None:
            os.environ["GITEA_TOKEN"] = self.original_gitea_token
        else:
            del os.environ["GITEA_TOKEN"]

    def _prepare_mock_client(self, payload, is_file=False):
        response = MagicMock()
        response.raise_for_status = MagicMock()
        if is_file:
            response.json.return_value = {"name": "file.txt", "path": "file.txt", "type": "file"}
        else:
            response.json.return_value = payload
        
        client = MagicMock()
        client.get = AsyncMock(return_value=response)

        async_client = MagicMock()
        async_client.__aenter__ = AsyncMock(return_value=client)
        async_client.__aexit__ = AsyncMock(return_value=None)

        return async_client, client

    @patch("httpx.AsyncClient")
    def test_list_repo_files_root(self, mock_async_client_constructor):
        payload = [
            {"name": "README.md", "path": "README.md", "type": "file", "sha": "sha123", "size": 1024},
            {"name": "docs", "path": "docs", "type": "dir"}
        ]
        mock_async_client, _ = self._prepare_mock_client(payload)
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            result = await handle_list_repo_files({"path": None}, "test/repo")
            self.assertEqual(result["path"], "")
            self.assertEqual(len(result["files"]), 2)
            self.assertEqual(result["files"][0]["name"], "README.md")
            self.assertIn("sha", result["files"][0])
            self.assertIn("size", result["files"][0])
            self.assertEqual(result["files"][1]["name"], "docs")
            self.assertNotIn("sha", result["files"][1])
            self.assertNotIn("size", result["files"][1])

        import asyncio
        asyncio.run(run_test())
    
    @patch("httpx.AsyncClient")
    def test_list_repo_files_empty_path(self, mock_async_client_constructor):
        payload = []
        mock_async_client, _ = self._prepare_mock_client(payload)
        mock_async_client_constructor.return_value = mock_async_client
        async def run_test():
            result = await handle_list_repo_files({"path": ""}, "test/repo")
            self.assertEqual(result["path"], "")
        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_nested_path(self, mock_async_client_constructor):
        payload = []
        mock_async_client, mock_client = self._prepare_mock_client(payload)
        mock_async_client_constructor.return_value = mock_async_client
        async def run_test():
            await handle_list_repo_files({"path": "docs/security"}, "test/repo")
            mock_client.get.assert_called_once()
            called_url = mock_client.get.call_args[0][0]
            self.assertIn("docs/security", called_url)

        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_path_is_file(self, mock_async_client_constructor):
        mock_async_client, _ = self._prepare_mock_client(None, is_file=True)
        mock_async_client_constructor.return_value = mock_async_client
        async def run_test():
            with self.assertRaisesRegex(ValueError, "Path is a file, not a directory. Use get_file_content instead."):
                await handle_list_repo_files({"path": "file.txt"}, "test/repo")
        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_root_none(self, mock_async_client_constructor):
        """Verify path=None retains root-listing behavior and params."""
        mock_async_client, mock_client = self._prepare_mock_client([])
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            await handle_list_repo_files({"path": None}, "test/repo")
            mock_client.get.assert_called_once()
            called_url = mock_client.get.call_args[0][0]
            self.assertTrue(called_url.endswith("/contents/"))
            self.assertEqual(mock_client.get.call_args[1].get("params"), {"ref": "main"})

        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_root_empty(self, mock_async_client_constructor):
        """Verify path="" retains root-listing behavior and params."""
        mock_async_client, mock_client = self._prepare_mock_client([])
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            await handle_list_repo_files({"path": ""}, "test/repo")
            mock_client.get.assert_called_once()
            called_url = mock_client.get.call_args[0][0]
            self.assertTrue(called_url.endswith("/contents/"))
            self.assertEqual(mock_client.get.call_args[1].get("params"), {"ref": "main"})

        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_dot_path_denied(self, mock_async_client_constructor):
        """Verify path='.' is denied before making an HTTP request."""
        mock_async_client, mock_client = self._prepare_mock_client([])
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaises(ValueError) as cm:
                await handle_list_repo_files({"path": "."}, "test/repo")
            self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
            mock_client.get.assert_not_called()

        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_safe_nested_path(self, mock_async_client_constructor):
        """Verify a safe nested path reaches the client and preserves params."""
        mock_async_client, mock_client = self._prepare_mock_client([])
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            await handle_list_repo_files({"path": "docs/sub"}, "test/repo")
            mock_client.get.assert_called_once()
            called_url = mock_client.get.call_args[0][0]
            self.assertTrue(called_url.endswith("/contents/docs/sub"))
            self.assertEqual(mock_client.get.call_args[1].get("params"), {"ref": "main"})

        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_traversal_path_denied(self, mock_async_client_constructor):
        """Verify a traversal path is denied before any HTTP request."""
        mock_async_client, mock_client = self._prepare_mock_client([])
        mock_async_client_constructor.return_value = mock_async_client
        async def run_test():
            with self.assertRaises(ValueError) as cm:
                await handle_list_repo_files({"path": "a/../b"}, "test/repo")
            self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
            mock_client.get.assert_not_called()
        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_secret_request_path_denied(self, mock_async_client_constructor):
        """Verify a secret requested path is denied before any HTTP request."""
        mock_async_client, mock_client = self._prepare_mock_client([])
        mock_async_client_constructor.return_value = mock_async_client
        async def run_test():
            with self.assertRaises(ValueError) as cm:
                await handle_list_repo_files({"path": ".env"}, "test/repo")
            self.assertEqual(str(cm.exception), "SECRET_PATH_DENIED")
            mock_client.get.assert_not_called()
        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_hardened_output_filtering(self, mock_async_client_constructor):
        """Verify secret paths are omitted from the returned list."""
        gitea_response_payload = [
            {"name": "README.md", "path": "README.md", "type": "file", "sha": "safe_sha", "size": 100},
            {"name": ".env", "path": ".env", "type": "file"},
            {"name": "config", "path": ".git/config", "type": "file"},
            {"name": "app.txt", "path": "secrets/app.txt", "type": "file"},
            {"name": "secret.pem", "path": "keys/secret.pem", "type": "file"},
        ]
        mock_async_client, _ = self._prepare_mock_client(gitea_response_payload)
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            result = await handle_list_repo_files({"path": ""}, "test/repo")
            self.assertEqual(result["files"], [{"name": "README.md", "path": "README.md", "type": "file", "sha": "safe_sha", "size": 100}])
            returned_paths = {f["path"] for f in result["files"]}
            self.assertNotIn(".env", returned_paths)
            self.assertNotIn(".git/config", returned_paths)
            self.assertNotIn("secrets/app.txt", returned_paths)
            self.assertNotIn("keys/secret.pem", returned_paths)

        import asyncio
        asyncio.run(run_test())

    def test_hardened_unexpected_error_reraised(self):
        """Verify other ValueErrors are not suppressed during filtering."""
        import gitea_handler
        real_validator = gitea_handler._validate_safe_path

        def validator_side_effect(path, *args, **kwargs):
            if path == "trigger.txt":
                raise ValueError("UNEXPECTED_ERROR")
            return real_validator(path, *args, **kwargs)

        async def run_test():
            with patch("gitea_handler._validate_safe_path") as mock_validator:
                mock_validator.side_effect = validator_side_effect

                with patch("httpx.AsyncClient") as mock_async_client_constructor:
                    mock_async_client, mock_client = self._prepare_mock_client(
                        [
                            {
                                "name": "safe.txt",
                                "path": "safe.txt",
                                "type": "file",
                                "sha": "safe_sha",
                                "size": 1,
                            },
                            {
                                "name": "trigger.txt",
                                "path": "trigger.txt",
                                "type": "file",
                                "sha": "trigger_sha",
                                "size": 1,
                            },
                        ]
                    )
                    mock_async_client_constructor.return_value = mock_async_client

                    with self.assertRaises(ValueError) as cm:
                        await handle_list_repo_files({"path": ""}, "test/repo")

                    self.assertEqual(str(cm.exception), "UNEXPECTED_ERROR")
                    mock_client.get.assert_called_once()

        import asyncio
        asyncio.run(run_test())

    def test_invalid_paths(self):
        invalid_paths = [
            "../secrets",
            "docs/../secrets",
            "docs\\secrets",
            "/etc/passwd",
            "https://example.com",
            "docs//security",
            "docs/./security",
        ]
        async def run_test():
            for path in invalid_paths:
                with self.subTest(path=path):
                    # Use patch here to ensure no real network call is made for paths
                    # that might otherwise be valid URLs.
                    with patch("httpx.AsyncClient") as mock_client:
                        mock_async_client, mocked_http_client = self._prepare_mock_client([])
                        mock_client.return_value = mock_async_client
                        with self.assertRaises(ValueError) as cm:
                            await handle_list_repo_files({"path": path}, "test/repo")
                        self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
                        # Ensure no network call was attempted for invalid paths
                        mocked_http_client.get.assert_not_called()
        import asyncio
        asyncio.run(run_test())

    def test_invalid_input_types(self):
        async def run_test():
            with self.assertRaises(ValueError) as cm:
                await handle_list_repo_files({"path": 123}, "test/repo")
            self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
        import asyncio
        asyncio.run(run_test())

    def test_unsupported_fields(self):
        unsupported = [
            {"repo": "x"},
            {"ref": "y"},
            {"branch": "z"},
            {"extra": "key"},
        ]
        async def run_test():
            for p in unsupported:
                with self.subTest(p=p):
                    with self.assertRaisesRegex(ValueError, "Unsupported list_repo_files input field."):
                        await handle_list_repo_files(p, "test/repo")
        import asyncio
        asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_uses_trusted_repo(self, mock_async_client_constructor):
        payload = []
        mock_async_client, mock_client = self._prepare_mock_client(payload)
        mock_async_client_constructor.return_value = mock_async_client
        async def run_test():
            await handle_list_repo_files({"path": "docs", "repo": "user/bad-repo"}, "test/repo")
            # The handler should raise an error due to the unsupported 'repo' field.
        import asyncio
        with self.assertRaisesRegex(ValueError, "Unsupported list_repo_files input field."):
            asyncio.run(run_test())

    @patch("httpx.AsyncClient")
    def test_http_error_propagation(self, mock_async_client_constructor):
        response = MagicMock()
        response.raise_for_status.side_effect = httpx.HTTPStatusError("Error", request=MagicMock(), response=MagicMock())
        client = MagicMock()
        client.get = AsyncMock(return_value=response)
        async_client = MagicMock()
        async_client.__aenter__ = AsyncMock(return_value=client)
        async_client.__aexit__ = AsyncMock(return_value=None)
        mock_async_client_constructor.return_value = async_client

        async def run_test():
            with self.assertRaises(httpx.HTTPStatusError):
                await handle_list_repo_files({"path": ""}, "test/repo")
        
        import asyncio
        asyncio.run(run_test())


class TestHandleGetFileContent(unittest.TestCase):
    def setUp(self):
        self.original_gitea_url = os.environ.get("GITEA_URL")
        self.original_gitea_token = os.environ.get("GITEA_TOKEN")
        os.environ["GITEA_URL"] = "https://gitea.example.com"
        os.environ["GITEA_TOKEN"] = "test-token"
        self.server_repo_id = "chris/OSVauco"

    def tearDown(self):
        if self.original_gitea_url is None:
            os.environ.pop("GITEA_URL", None)
        else:
            os.environ["GITEA_URL"] = self.original_gitea_url
        if self.original_gitea_token is None:
            os.environ.pop("GITEA_TOKEN", None)
        else:
            os.environ["GITEA_TOKEN"] = self.original_gitea_token

    def _mock_contents_client(self, text):
        body = json.dumps({
            "content": base64.b64encode(text.encode("utf-8")).decode("ascii"),
            "encoding": "base64",
        }).encode("utf-8")

        async def chunks():
            yield body

        response = MagicMock()
        response.headers = {"content-length": str(len(body))}
        response.aiter_bytes = chunks
        stream = MagicMock()
        stream.__aenter__.return_value = response
        client = MagicMock()
        client.stream.return_value = stream
        async_client = MagicMock()
        async_client.__aenter__.return_value = client
        return async_client, client

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_direct_lowercase_sha_success(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        sha = "a" * 40
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            return await handle_get_file_content(
                {
                    "repo": self.server_repo_id,
                    "path": "README.md",
                    "ref": sha,
                },
                self.server_repo_id,
            )

        result = asyncio.run(run_test())

        mock_resolver.assert_not_called()
        client.stream.assert_called_once()
        request_url = client.stream.call_args.args[1]
        self.assertIn(f"ref={sha}", request_url)
        self.assertEqual(result["requested_ref"], sha)
        self.assertEqual(result["resolved_commit_sha"], sha)
        self.assertEqual(result["content"], "file content")
        self.assertEqual(result["encoding"], "utf-8")

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_direct_uppercase_sha_normalizes_for_request(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        requested_sha = "A" * 40
        resolved_sha = requested_sha.lower()
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            return await handle_get_file_content(
                {
                    "repo": self.server_repo_id,
                    "path": "README.md",
                    "ref": requested_sha,
                },
                self.server_repo_id,
            )

        result = asyncio.run(run_test())

        mock_resolver.assert_not_called()
        request_url = client.stream.call_args.args[1]
        self.assertIn(f"ref={resolved_sha}", request_url)
        self.assertNotIn(f"ref={requested_sha}", request_url)
        self.assertEqual(result["requested_ref"], requested_sha)
        self.assertEqual(result["resolved_commit_sha"], resolved_sha)

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_branch_resolves_to_sha_before_contents_request(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        resolved_sha = "b" * 40
        mock_resolver.return_value = resolved_sha
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            return await handle_get_file_content(
                {
                    "repo": self.server_repo_id,
                    "path": "README.md",
                    "ref": "main",
                },
                self.server_repo_id,
            )

        result = asyncio.run(run_test())

        mock_resolver.assert_awaited_once_with(
            branch_name="main",
            repo_id=self.server_repo_id,
            gitea_url="https://gitea.example.com",
        )
        request_url = client.stream.call_args.args[1]
        self.assertIn(f"ref={resolved_sha}", request_url)
        self.assertNotIn("ref=main", request_url)
        self.assertEqual(result["repo"], self.server_repo_id)
        self.assertEqual(result["path"], "README.md")
        self.assertEqual(result["requested_ref"], "main")
        self.assertEqual(result["resolved_commit_sha"], resolved_sha)
        self.assertEqual(result["content"], "file content")
        self.assertEqual(result["encoding"], "utf-8")

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_unresolved_branch_raises_sanitized_error(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        mock_resolver.side_effect = httpx.HTTPStatusError(
            "Not Found",
            request=MagicMock(),
            response=MagicMock(status_code=404),
        )
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaisesRegex(
                ValueError, r"^Unknown or inaccessible branch reference\.$"
            ):
                await handle_get_file_content(
                    {
                        "repo": self.server_repo_id,
                        "path": "README.md",
                        "ref": "missing-branch",
                    },
                    self.server_repo_id,
                )

        asyncio.run(run_test())

        client.stream.assert_not_called()

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_resolver_non_404_http_error_is_sanitized(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        mock_resolver.side_effect = httpx.HTTPStatusError(
            "Server Error",
            request=MagicMock(),
            response=MagicMock(status_code=500),
        )
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaisesRegex(
                ValueError, r"^Repository file is unavailable\.$"
            ):
                await handle_get_file_content(
                    {
                        "repo": self.server_repo_id,
                        "path": "README.md",
                        "ref": "main",
                    },
                    self.server_repo_id,
                )

        asyncio.run(run_test())

        client.stream.assert_not_called()

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_resolver_unexpected_exception_is_sanitized(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        mock_resolver.side_effect = RuntimeError(
            "upstream internal details must not reach the caller"
        )
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaisesRegex(
                ValueError, r"^Repository file is unavailable\.$"
            ):
                await handle_get_file_content(
                    {
                        "repo": self.server_repo_id,
                        "path": "README.md",
                        "ref": "main",
                    },
                    self.server_repo_id,
                )

        asyncio.run(run_test())

        client.stream.assert_not_called()

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_resolver_returns_malformed_sha_is_rejected(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        mock_resolver.return_value = "not-a-40-character-commit-sha"
        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaisesRegex(ValueError, r"^Invalid commit SHA$"):
                await handle_get_file_content(
                    {
                        "repo": self.server_repo_id,
                        "path": "README.md",
                        "ref": "main",
                    },
                    self.server_repo_id,
                )

        asyncio.run(run_test())

        client.stream.assert_not_called()

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_invalid_branch_ref_is_rejected(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        mock_async_client, client = self._mock_contents_client("file content")
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
                await handle_get_file_content(
                    {
                        "repo": self.server_repo_id,
                        "path": "README.md",
                        "ref": "main/../secret",
                    },
                    self.server_repo_id,
                )

        asyncio.run(run_test())

        mock_resolver.assert_not_called()
        client.stream.assert_not_called()

    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
    @patch("httpx.AsyncClient")
    def test_contents_api_404_is_sanitized(
        self, mock_async_client_constructor, mock_resolver
    ):
        import asyncio

        resolved_sha = "c" * 40
        mock_resolver.return_value = resolved_sha
        mock_async_client, client = self._mock_contents_client("file content")

        response_mock = client.stream.return_value.__aenter__.return_value
        response_mock.raise_for_status.side_effect = httpx.HTTPStatusError(
            "Not Found", request=MagicMock(), response=MagicMock(status_code=404)
        )
        mock_async_client_constructor.return_value = mock_async_client

        async def run_test():
            with self.assertRaisesRegex(
                ValueError, r"^Repository file is unavailable\.$"
            ):
                await handle_get_file_content(
                    {
                        "repo": self.server_repo_id,
                        "path": "README.md",
                        "ref": "main",
                    },
                    self.server_repo_id,
                )

        asyncio.run(run_test())

        mock_resolver.assert_awaited_once_with(
            branch_name="main",
            repo_id=self.server_repo_id,
            gitea_url="https://gitea.example.com",
        )
        client.stream.assert_called_once()


if __name__ == '__main__':
    unittest.main()