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)) repo_root = os.path.realpath( os.path.join(os.path.dirname(__file__), os.pardir) ) branches_root = os.path.join(repo_root, "branches") os.makedirs(branches_root, exist_ok=True) self.locked_worktree_dir = tempfile.TemporaryDirectory( dir=branches_root, prefix="test-commit-payloads-", ) self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name) import issue_lock_store import issue_lock_provenance self._lock_dir = tempfile.TemporaryDirectory() os.environ["GITEA_ISSUE_LOCK_DIR"] = self._lock_dir.name work_lease = { "operation_type": "author_issue_work", "issue_number": 263, "branch": "feat/issue-263-native-commit-payloads", "claimant": {"username": "test-user", "profile": "test-author"}, "expires_at": "2999-01-01T00:00:00Z", } 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, "work_lease": work_lease, "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( tool="gitea_lock_issue", claimant=work_lease.get("claimant"), ), } self.lock_file_path = issue_lock_store.bind_session_lock(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 self._dup_fetcher_patcher = patch( "mcp_server.issue_duplicate_context_fetcher", return_value=([], [], {"status": "not_claimed"}), ) self._dup_fetcher_patcher.start() def tearDown(self): self._dup_fetcher_patcher.stop() 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() self._lock_dir.cleanup() 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": "", "GITEA_AUTHOR_WORKTREE": self.locked_worktree_path, "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name, } @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))