import unittest from unittest.mock import AsyncMock, MagicMock, patch import os import sys from pathlib import Path import base64 import httpx import json import binascii # Correction 4: Schema test isolation # Before importing server, install only a minimal test stub in sys.modules for # the unrelated runtime dependency `emma_adapter`. emma_adapter_stub = MagicMock() emma_adapter_stub.CanonicalEmma = MagicMock() sys.modules['emma_adapter'] = emma_adapter_stub REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) # Now that the path is set, we can import the modules import gitea_handler import server def make_stream_context(response_content, headers=None): async def aiter_bytes(): if isinstance(response_content, list): for item in response_content: yield item else: yield response_content response = MagicMock() response.aiter_bytes = aiter_bytes response.raise_for_status = MagicMock() response.headers = headers if headers is not None else {} context = MagicMock() context.__aenter__ = AsyncMock(return_value=response) context.__aexit__ = AsyncMock(return_value=False) return context def configure_stream(client, response_content, headers=None): client.stream = MagicMock( return_value=make_stream_context(response_content, headers=headers) ) def _get_mock_file_response(content=b"test content", size=None, content_b64=None): """Helper to create a default valid mock response for get_file tests.""" if content_b64 is None: encoded_content = base64.b64encode(content).decode('ascii') else: encoded_content = content_b64 if size is None: size = len(content) mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = {"size": size, "content": encoded_content} mock_response.raise_for_status = MagicMock() return mock_response class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): def test_validate_branch_name_valid(self): self.assertEqual(gitea_handler._validate_branch_name("main"), "main") self.assertEqual(gitea_handler._validate_branch_name("feat/opax-domain-decouple"), "feat/opax-domain-decouple") def test_validate_branch_name_invalid(self): with self.assertRaises(ValueError): gitea_handler._validate_branch_name("../../etc/passwd") with self.assertRaises(ValueError): gitea_handler._validate_branch_name("main@{bad}") with self.assertRaises(ValueError): gitea_handler._validate_branch_name("feat//bad") def test_validate_commit_sha_valid(self): self.assertEqual(gitea_handler._validate_commit_sha("f" * 40), "f" * 40) def test_validate_commit_sha_invalid(self): with self.assertRaises(ValueError): gitea_handler._validate_commit_sha("main") with self.assertRaises(ValueError): gitea_handler._validate_commit_sha("f" * 39) def test_validate_gitea_url_https_required(self): with self.assertRaises(ValueError): gitea_handler._validate_gitea_url("http://example.com") @patch("gitea_handler.httpx.AsyncClient") async def test_resolve_branch_to_commit_sha(self, mock_client): mock_response = unittest.mock.Mock() mock_response.status_code = 200 mock_response.json.return_value = {"commit": {"id": "f" * 40}} mock_client.return_value.__aenter__.return_value.get.return_value = mock_response sha = await gitea_handler.resolve_branch_to_commit_sha("main", "chris/OSVauco", "https://gitea.example.com") self.assertEqual(sha, "f" * 40) @patch("gitea_handler.httpx.AsyncClient") async def test_download_repo_archive_uses_sha(self, mock_client): stream_context = make_stream_context(b"archive-data") client = mock_client.return_value.__aenter__.return_value client.stream = MagicMock(return_value=stream_context) archive = await gitea_handler.download_repo_archive( "f" * 40, "chris/OSVauco", "https://gitea.example.com", ) self.assertEqual(archive, b"archive-data") client.stream.assert_called_once_with( "GET", ( f"https://gitea.example.com/api/v1/repos/" f"chris/OSVauco/archive/{'f' * 40}.tar.gz" ), headers=unittest.mock.ANY, ) @patch("gitea_handler.httpx.AsyncClient") async def test_download_repo_archive_size_limit(self, mock_client): response = MagicMock() response.headers = { "content-length": str( gitea_handler.MAX_SOURCE_ARCHIVE_BYTES + 1 ) } response.raise_for_status = MagicMock() stream_context = make_stream_context([]) # Empty stream client = mock_client.return_value.__aenter__.return_value client.stream = MagicMock(return_value=stream_context) # The response is used to check headers, not the stream content stream_context.__aenter__.return_value = response with self.assertRaisesRegex( ValueError, "Source archive exceeds allowed size", ): await gitea_handler.download_repo_archive( "f" * 40, "chris/OSVauco", "https://gitea.example.com", ) @patch("gitea_handler.httpx.AsyncClient") async def test_download_repo_archive_no_redirects(self, mock_client): stream_context = make_stream_context(b"") client = mock_client.return_value.__aenter__.return_value client.stream = MagicMock(return_value=stream_context) await gitea_handler.download_repo_archive( "f" * 40, "chris/OSVauco", "https://gitea.example.com", ) self.assertFalse( mock_client.call_args.kwargs["follow_redirects"] ) class TestGiteaCatalog(unittest.IsolatedAsyncioTestCase): def setUp(self): self.mock_env = patch.dict(os.environ, { "GITEA_URL": "https://gitea.example.com", "GITEA_TOKEN": "fake-token", }) self.mock_env.start() def tearDown(self): self.mock_env.stop() @patch("gitea_handler.httpx.AsyncClient") async def test_catalog_uses_fixed_namespace(self, mock_client): mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = [] mock_client.return_value.__aenter__.return_value.get.return_value = mock_response await gitea_handler.list_allowed_namespace_repositories() mock_client.return_value.__aenter__.return_value.get.assert_called_once() call_url = mock_client.return_value.__aenter__.return_value.get.call_args[0][0] self.assertIn("/api/v1/users/chris/repos", call_url) @patch("gitea_handler.httpx.AsyncClient") async def test_catalog_pagination_is_bounded(self, mock_client): async def get_response(url, headers): mock_response = MagicMock() mock_response.status_code = 200 if "page=1" in url: mock_response.json.return_value = [{"full_name": f"chris/repo{i}", "name": f"repo{i}"} for i in range(50)] elif "page=2" in url: mock_response.json.return_value = [{"full_name": f"chris/repo{i}", "name": f"repo{i}"} for i in range(50, 100)] else: # Should not be called for page 3 mock_response.json.return_value = [] return mock_response mock_client.return_value.__aenter__.return_value.get.side_effect = get_response result = await gitea_handler.list_allowed_namespace_repositories() self.assertEqual(mock_client.return_value.__aenter__.return_value.get.call_count, 2) self.assertEqual(len(result["repositories"]), 100) @patch("gitea_handler.httpx.AsyncClient") async def test_catalog_normalization_and_security(self, mock_client): gitea_response = [ {"full_name": "chris/repo1", "name": "repo1", "clone_url": "sensitive", "permissions": {}}, {"full_name": "other/repo2", "name": "repo2"}, {"full_name": "chris/repo3"}, {"full_name": "chris/repo1", "name": "repo1"}, ] mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = gitea_response mock_client.return_value.__aenter__.return_value.get.return_value = mock_response result = await gitea_handler.list_allowed_namespace_repositories() self.assertEqual(len(result["repositories"]), 1) repo = result["repositories"][0] self.assertEqual(repo["full_name"], "chris/repo1") self.assertNotIn("clone_url", repo) self.assertNotIn("permissions", repo) @patch("gitea_handler.httpx.AsyncClient") async def test_catalog_upstream_failure_500(self, mock_client): mock_client.return_value.__aenter__.return_value.get.side_effect = httpx.HTTPStatusError( "Internal Server Error", request=MagicMock(), response=MagicMock(status_code=500) ) with self.assertRaisesRegex(ValueError, "Repository catalog unavailable."): await gitea_handler.list_allowed_namespace_repositories() @patch("gitea_handler.httpx.AsyncClient") async def test_catalog_upstream_failure_timeout(self, mock_client): mock_client.return_value.__aenter__.return_value.get.side_effect = httpx.TimeoutException("Timeout") with self.assertRaisesRegex(ValueError, "Repository catalog unavailable."): await gitea_handler.list_allowed_namespace_repositories() @patch("gitea_handler.httpx.AsyncClient") async def test_catalog_malformed_json(self, mock_client): mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.side_effect = json.JSONDecodeError("err", "doc", 0) mock_client.return_value.__aenter__.return_value.get.return_value = mock_response with self.assertRaisesRegex(ValueError, "Repository catalog unavailable."): await gitea_handler.list_allowed_namespace_repositories() class TestGetFileHardening(unittest.IsolatedAsyncioTestCase): def setUp(self): self.mock_env_patcher = patch.dict(os.environ, { "GITEA_URL": "https://gitea.example.com", "GITEA_TOKEN": "fake-token", "GITEA_REPO": "chris/OSVauco", }) self.mock_env = self.mock_env_patcher.start() self.valid_sha = "f" * 40 self.params = {"path": "README.md", "ref": self.valid_sha} self.repo = "chris/OSVauco" def tearDown(self): self.mock_env_patcher.stop() @patch("gitea_handler.httpx.AsyncClient") async def test_get_file_repo_compatibility(self, mock_client): client = mock_client.return_value.__aenter__.return_value configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8")) # Allowed cases await gitea_handler.handle_get_file_content( {"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco" ) await gitea_handler.handle_get_file_content( {"path": "README.md", "ref": self.valid_sha, "repo": None}, "chris/OSVauco" ) await gitea_handler.handle_get_file_content( {"path": "README.md", "ref": self.valid_sha, "repo": "chris/OSVauco"}, "chris/OSVauco" ) self.assertEqual(client.stream.call_count, 3) # Rejected cases rejected_repos = { "empty_string": "", "wrong_type_false": False, "wrong_type_zero": 0, "mismatch": "other/repo", } for name, repo_val in rejected_repos.items(): with self.subTest(rejected_case=name): client.stream.reset_mock() with self.assertRaisesRegex(ValueError, "Repository file request is not allowed."): await gitea_handler.handle_get_file_content( {"path": "README.md", "ref": self.valid_sha, "repo": repo_val}, "chris/OSVauco" ) client.stream.assert_not_called() @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) async def test_get_file_ref_validation(self, mock_resolver): mock_resolver.side_effect = httpx.ConnectError("Network error") with self.assertRaisesRegex(ValueError, r"^Repository file is unavailable\.$"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": "main"}, "chris/OSVauco") mock_resolver.assert_awaited_once() self.assertEqual(mock_resolver.call_args.kwargs['branch_name'], "main") self.assertEqual(mock_resolver.call_args.kwargs['repo_id'], "chris/OSVauco") with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": None}, "chris/OSVauco") # Verify the resolver was not called a second time for the invalid ref self.assertEqual(mock_resolver.await_count, 1) async def test_gitea_repo_validation(self): with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "invalid-repo-id") with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "") async def test_get_file_path_not_allowed(self): for path in ["../secrets.txt", "/etc/passwd"]: with self.subTest(path=path): with self.assertRaisesRegex(ValueError, r"^PATH_NOT_ALLOWED$"): await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco") async def test_get_file_secret_path_denied(self): with self.assertRaisesRegex(ValueError, r"^SECRET_PATH_DENIED$"): await gitea_handler.handle_get_file_content({"path": "docs/.env", "ref": self.valid_sha}, "chris/OSVauco") @patch("gitea_handler.httpx.AsyncClient") async def test_get_file_valid_paths(self, mock_client): client = mock_client.return_value.__aenter__.return_value configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8")) for path in ["README.md", "docs/ARCHITECTURE.md", ".gemini/GEMINI.md"]: with self.subTest(path=path): await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco") self.assertEqual(client.stream.call_count, 3) @patch("gitea_handler.httpx.AsyncClient") async def test_get_file_size_limit(self, mock_client): client = mock_client.return_value.__aenter__.return_value # Declared size too large response_body = json.dumps({"size": gitea_handler.MAX_READ_FILE_BYTES + 1, "content": ""}).encode("utf-8") configure_stream(client, response_body) with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") # Transport size too large large_body = b'{' * (gitea_handler.MAX_GET_FILE_RESPONSE_BYTES + 1) configure_stream(client, large_body) with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") @patch("gitea_handler.httpx.AsyncClient") async def test_get_file_content_validation(self, mock_client): client = mock_client.return_value.__aenter__.return_value # Malformed Base64 configure_stream(client, json.dumps({'content': 'invalid-b64!', 'size': 12}).encode("utf-8")) with self.assertRaisesRegex(ValueError, "Repository file content is not readable text."): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") # Invalid UTF-8 invalid_utf8_b64 = base64.b64encode(b"\xff\xff\xff").decode('ascii') configure_stream(client, json.dumps({'content': invalid_utf8_b64, 'size': 3}).encode("utf-8")) with self.assertRaisesRegex(ValueError, "Repository file content is not readable text."): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") # NUL byte nul_byte_b64 = base64.b64encode(b"hello\0world").decode('ascii') configure_stream(client, json.dumps({'content': nul_byte_b64, 'size': 11}).encode("utf-8")) with self.assertRaisesRegex(ValueError, "Repository file content is not readable text."): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") @patch("gitea_handler.httpx.AsyncClient") async def test_content_length_validation(self, mock_client): client = mock_client.return_value.__aenter__.return_value test_cases = { "absent": None, "empty": "", "non-numeric": "not-a-number", "negative": "-100", } for name, length_val in test_cases.items(): with self.subTest(name=name): headers = {"content-length": length_val} if length_val is not None else {} configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8"), headers=headers ) # No ValueError should be raised, should proceed and succeed result = await gitea_handler.handle_get_file_content(self.params, self.repo) self.assertEqual(result["content"], "a") # Test valid but too large with self.subTest(name="valid_too_large"): headers = {"content-length": str(gitea_handler.MAX_GET_FILE_RESPONSE_BYTES + 1)} configure_stream(client, b"", headers=headers) with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): await gitea_handler.handle_get_file_content(self.params, self.repo) @patch("gitea_handler.httpx.AsyncClient") async def test_json_size_validation(self, mock_client): client = mock_client.return_value.__aenter__.return_value test_cases = { "True": True, "False": False, "negative": -1, "string": "123", } for name, size_val in test_cases.items(): with self.subTest(name=name): body = json.dumps({"content": "YQ==", "size": size_val}).encode("utf-8") configure_stream(client, body) # No error should be raised, should proceed and succeed result = await gitea_handler.handle_get_file_content(self.params, self.repo) self.assertEqual(result["content"], "a") # Test valid int over the limit with self.subTest(name="valid_too_large"): body = json.dumps({ "content": "", "size": gitea_handler.MAX_READ_FILE_BYTES + 1 }).encode("utf-8") configure_stream(client, body) with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): await gitea_handler.handle_get_file_content(self.params, self.repo) class TestMCPSchema(unittest.TestCase): def test_get_file_schema(self): _handler, _desc, schema = server.TOOLS["get_file"] self.assertEqual(schema["properties"].keys(), {"path", "ref", "repo"}) self.assertEqual(schema["required"], ["path", "ref"]) def test_list_gitea_repositories_schema(self): _handler, _desc, schema = server.TOOLS["list_gitea_repositories"] self.assertEqual(schema, {}) if __name__ == "__main__": unittest.main()