import json import os import tempfile import unittest from unittest.mock import MagicMock, patch import gitea_config import mcp_server import task_capability_map CONFIG = { "version": 2, "contexts": { "ctx": { "enabled": True, "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, } }, "profiles": { "full-author": { "enabled": True, "context": "ctx", "role": "author", "username": "author-user", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": [ "gitea.read", "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push", ], "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", ], "execution_profile": "full-author", }, }, } class TestCommitPayloads(unittest.TestCase): def setUp(self): self._remotes = patch.dict(mcp_server.REMOTES, { "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}, }) self._remotes.start() mcp_server._IDENTITY_CACHE.clear() # Setup temp directories and lock files self._dir = tempfile.TemporaryDirectory() self.config_path = os.path.join(self._dir.name, "profiles.json") with open(self.config_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(CONFIG)) self.locked_worktree_dir = tempfile.TemporaryDirectory() self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name) self.lock_file_path = "/tmp/gitea_issue_lock.json" self.lock_data = { "issue_number": 263, "branch_name": "feat/issue-263-native-commit-payloads", "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", "worktree_path": self.locked_worktree_path, } with open(self.lock_file_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(self.lock_data)) # Reset preflight status to bypass/pass verification in tests self.orig_whoami_called = mcp_server._preflight_whoami_called self.orig_capability_called = mcp_server._preflight_capability_called mcp_server._preflight_whoami_called = True mcp_server._preflight_capability_called = True def tearDown(self): self._remotes.stop() mcp_server._IDENTITY_CACHE.clear() mcp_server._preflight_whoami_called = self.orig_whoami_called mcp_server._preflight_capability_called = self.orig_capability_called self._dir.cleanup() self.locked_worktree_dir.cleanup() if os.path.exists(self.lock_file_path): os.remove(self.lock_file_path) def _env(self, profile: str) -> dict: return { "GITEA_MCP_CONFIG": self.config_path, "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", "GITEA_TEST_PORCELAIN": "", } @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_commit_files_content_plain(self, _auth, mock_api): mock_api.return_value = { "commit": {"sha": "sha-123"}, "branch": {"name": "feat/issue-263-native-commit-payloads"} } with patch.dict(os.environ, self._env("full-author"), clear=True): res = mcp_server.gitea_commit_files( files=[ { "operation": "create", "path": "hello.txt", "content_plain": "Hello World!", } ], message="Add hello.txt", remote="prgs", ) print("DEBUG RES IS:", res) self.assertTrue(res["success"]) self.assertEqual(res["commit"], "sha-123") self.assertEqual(res["content_source_proof"][0]["source"], "inline_plain") # Verify posted base64 content post_args = mock_api.call_args[0] payload = post_args[3] self.assertEqual(payload["files"][0]["content"], "SGVsbG8gV29ybGQh") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_commit_files_workspace_path(self, _auth, mock_api): mock_api.return_value = { "commit": {"sha": "sha-456"}, "branch": {"name": "feat/issue-263-native-commit-payloads"} } # Create a workspace file file_relative = "src/code.py" file_abs = os.path.join(self.locked_worktree_path, file_relative) os.makedirs(os.path.dirname(file_abs), exist_ok=True) with open(file_abs, "wb") as fh: fh.write(b"print('hello')") with patch.dict(os.environ, self._env("full-author"), clear=True): res = mcp_server.gitea_commit_files( files=[ { "operation": "create", "path": "src/code.py", "workspace_path": file_relative, } ], message="Add code.py", remote="prgs", ) self.assertTrue(res["success"]) self.assertEqual(res["content_source_proof"][0]["source"], "workspace_path") post_args = mock_api.call_args[0] payload = post_args[3] # "print('hello')" in base64 is cHJpbnQoJ2hlbGxvJyk= self.assertEqual(payload["files"][0]["content"], "cHJpbnQoJ2hlbGxvJyk=") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_commit_files_local_path(self, _auth, mock_api): mock_api.return_value = { "commit": {"sha": "sha-789"}, "branch": {"name": "feat/issue-263-native-commit-payloads"} } file_abs = os.path.join(self.locked_worktree_path, "local.bin") with open(file_abs, "wb") as fh: fh.write(b"\x00\x01\x02\x03\xff") with patch.dict(os.environ, self._env("full-author"), clear=True): res = mcp_server.gitea_commit_files( files=[ { "operation": "create", "path": "local.bin", "local_path": file_abs, } ], message="Add local.bin", remote="prgs", ) self.assertTrue(res["success"]) self.assertEqual(res["content_source_proof"][0]["source"], "local_path") post_args = mock_api.call_args[0] payload = post_args[3] # b"\x00\x01\x02\x03\xff" in base64 is AAECA/8= self.assertEqual(payload["files"][0]["content"], "AAECA/8=") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_commit_files_traversal_blocked(self, _auth, mock_api): # Remove active lock file to ensure it fails on traversal/invalid locks os.remove(self.lock_file_path) with patch.dict(os.environ, self._env("full-author"), clear=True): with self.assertRaises(RuntimeError) as ctx: mcp_server.gitea_commit_files( files=[ { "operation": "create", "path": "hack.txt", "workspace_path": "any.txt", } ], message="Add hack", remote="prgs", ) self.assertIn("Issue lock is missing", str(ctx.exception)) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_commit_files_outside_scope_blocked(self, _auth, mock_api): with patch.dict(os.environ, self._env("full-author"), clear=True): with self.assertRaises(ValueError) as ctx: mcp_server.gitea_commit_files( files=[ { "operation": "create", "path": "hack.txt", "workspace_path": "../outside.txt", } ], message="Add hack", remote="prgs", ) self.assertIn("falls outside of locked worktree", str(ctx.exception)) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_commit_files_multiple_sources_blocked(self, _auth, mock_api): with patch.dict(os.environ, self._env("full-author"), clear=True): with self.assertRaises(ValueError) as ctx: mcp_server.gitea_commit_files( files=[ { "operation": "create", "path": "hello.txt", "content_plain": "Hello!", "content": "SGVsbG8=", } ], message="Add hello", remote="prgs", ) self.assertIn("Multiple content sources specified", str(ctx.exception))