Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into docs/issue-260-forbid-commit-fallbacks

Browse Source
This commit is contained in:
sysadmin
2026-07-07 00:59:30 -05:00
parent 2f4672218c 042783076f
commit d0e250e860
10 changed files with 1014 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_temp_artifacts.py
+87
View File
@@ -0,0 +1,87 @@
"""Tests for agent temp artifact detection and preflight warnings (#261)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import agent_temp_artifacts
import mcp_server
class TestAgentTempArtifactPatterns(unittest.TestCase):
def test_detects_root_level_encode_emit_inline(self):
porcelain = (
"?? _encode_commit_payload.py\n"
"?? _emit_payload.py\n"
"?? _inline_b64.py\n"
)
self.assertEqual(
agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(porcelain),
["_emit_payload.py", "_encode_commit_payload.py", "_inline_b64.py"],
)
def test_ignores_nested_and_unrelated_untracked(self):
porcelain = (
"?? scripts/_encode_x.py\n"
"?? README-draft.md\n"
" M task_capability_map.py\n"
)
self.assertEqual(
agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(porcelain),
[],
)
class TestPreflightWarnings(unittest.TestCase):
def setUp(self):
self._snapshot = (
mcp_server._preflight_whoami_called,
mcp_server._preflight_capability_called,
)
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
def tearDown(self):
mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
self._snapshot
)
@patch(
"mcp_server._get_workspace_porcelain",
return_value="?? _encode_commit_payload.py\n",
)
def test_assess_preflight_surfaces_warning_not_block(self, _porcelain):
status = mcp_server.assess_preflight_status()
self.assertTrue(status["preflight_ready"])
self.assertEqual(status["preflight_block_reasons"], [])
self.assertEqual(len(status["preflight_warnings"]), 1)
self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
class TestIssueLockArtifactWarning(unittest.TestCase):
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
@patch("issue_lock_worktree.read_worktree_git_state")
def test_lock_success_includes_artifact_warning(self, mock_state, _lock_file, *_mocks):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
}
result = mcp_server.gitea_lock_issue(
issue_number=261,
branch_name="docs/issue-261-agent-artifact-cleanup",
remote="prgs",
)
self.assertTrue(result["success"])
self.assertIn("warnings", result)
self.assertIn("_emit_payload.py", result["warnings"][0])
if __name__ == "__main__":
unittest.main()
tests/test_commit_files_capability.py
+199
View File
@@ -0,0 +1,199 @@
"""Tests for commit_files task capability resolver mapping (#262)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import task_capability_map
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"commit-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.repo.commit"],
"forbidden_operations": ["gitea.pr.create"],
"execution_profile": "commit-author",
},
"pr-only-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "pr-author",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.pr.create"],
"forbidden_operations": ["gitea.repo.commit"],
"execution_profile": "pr-only-author",
},
"reviewer-profile": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.merge"],
"forbidden_operations": [
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push",
],
"execution_profile": "reviewer-profile",
},
},
"rules": {"allow_runtime_switching": False},
}
class CommitFilesCapabilityBase(unittest.TestCase):
def setUp(self):
self._preflight_snapshot = (
mcp_server._preflight_whoami_called,
mcp_server._preflight_capability_called,
)
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
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()
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))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
self._preflight_snapshot
)
self._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_TOKEN_REVIEWER": "reviewer-pass",
}
class TestCommitFilesResolver(CommitFilesCapabilityBase):
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_commit_files_resolves_to_repo_commit(self, _auth, _api):
with patch.dict(os.environ, self._env("commit-author"), clear=True):
for task in ("commit_files", "gitea_commit_files"):
with self.subTest(task=task):
res = mcp_server.gitea_resolve_task_capability(
task=task, remote="prgs")
self.assertEqual(res["required_operation_permission"],
"gitea.repo.commit")
self.assertEqual(res["required_role_kind"], "author")
self.assertTrue(res["allowed_in_current_session"])
@patch("mcp_server.api_request", return_value={"login": "pr-author"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_create_pr_does_not_satisfy_commit_files(self, _auth, _api):
with patch.dict(os.environ, self._env("pr-only-author"), clear=True):
commit_res = mcp_server.gitea_resolve_task_capability(
task="commit_files", remote="prgs")
pr_res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs")
self.assertFalse(commit_res["allowed_in_current_session"])
self.assertTrue(pr_res["allowed_in_current_session"])
self.assertEqual(commit_res["required_operation_permission"],
"gitea.repo.commit")
self.assertEqual(pr_res["required_operation_permission"],
"gitea.pr.create")
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_denied_commit_files(self, _auth, _api):
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
res = mcp_server.gitea_resolve_task_capability(
task="commit_files", remote="prgs")
self.assertFalse(res["allowed_in_current_session"])
self.assertEqual(res["required_operation_permission"],
"gitea.repo.commit")
self.assertTrue(res["different_mcp_namespace_required"])
class TestCommitFilesToolGate(CommitFilesCapabilityBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
mock_api.return_value = {
"commit": {"sha": "abc"},
"branch": {"name": "feat/x"},
}
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with patch.dict(os.environ, self._env("commit-author"), clear=True):
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
message="test",
new_branch="feat/x",
remote="prgs",
)
self.assertTrue(res["success"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_reviewer_blocked_with_permission_report(self, _auth, _role):
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
message="test",
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res.get("performed", True))
self.assertIn("permission_report", res)
self.assertEqual(res["permission_report"]["missing_permission"],
"gitea.repo.commit")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
def test_preflight_blocks_before_api(self, _role):
env = {**self._env("commit-author"), "GITEA_TEST_FORCE_DIRTY": "1"}
with patch.dict(os.environ, env, clear=True):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
message="test",
remote="prgs",
)
self.assertIn("gitea_whoami", str(ctx.exception))
class TestCommitFilesMapAlignment(unittest.TestCase):
def test_tool_gate_matches_resolver(self):
self.assertEqual(
task_capability_map.tool_required_permission("gitea_commit_files"),
task_capability_map.required_permission("commit_files"),
)
if __name__ == "__main__":
unittest.main()
tests/test_commit_files_gate.py
+251
View File
@@ -0,0 +1,251 @@
"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification.
Covers Issue #262 requirements.
"""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import task_capability_map
import gitea_config
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.issue.create", "gitea.repo.commit"
],
"forbidden_operations": [],
"execution_profile": "full-author",
},
"reviewer-no-commit": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": [
"gitea.read", "gitea.pr.review", "gitea.pr.approve"
],
"forbidden_operations": [
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
],
"execution_profile": "reviewer-no-commit",
},
},
"rules": {"allow_runtime_switching": False},
}
class TestCommitFilesGate(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()
gitea_config._active_profile_override = None
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))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._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_TOKEN_REVIEWER": "reviewer-pass",
}
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
mock_api.return_value = {
"commit": {"sha": "abc123commit"},
"branch": {"name": "some-branch"},
}
with patch.dict(os.environ, self._env("full-author"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="commit_files", remote="prgs"
)
self.assertTrue(resolve["allowed_in_current_session"])
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertTrue(res["success"])
self.assertEqual(res["commit"], "abc123commit")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_denied_reviewer_blocked(self, _auth, mock_api, _role):
mock_api.return_value = {"login": "reviewer-user"}
with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="commit_files", remote="prgs"
)
self.assertFalse(resolve["allowed_in_current_session"])
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res.get("performed", True))
self.assertIn("permission_report", res)
self.assertEqual(
res["permission_report"]["missing_permission"], "gitea.repo.commit"
)
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
self.assertFalse(post_calls)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_unknown_profile_fails_closed(self, _auth, mock_api, _role):
mock_api.return_value = {"login": "reviewer-user"}
with patch.dict(os.environ, self._env("non-existent"), clear=True):
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res.get("performed", True))
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
self.assertFalse(post_calls)
class TestPreflightCommitFilesGate(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()
self.orig_whoami_called = mcp_server._preflight_whoami_called
self.orig_capability_called = mcp_server._preflight_capability_called
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
self.orig_capability_violation = mcp_server._preflight_capability_violation
self.orig_resolved_role = mcp_server._preflight_resolved_role
self.orig_process_start = mcp_server._process_start_porcelain
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
self.orig_capability_files = mcp_server._preflight_capability_violation_files
self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
mcp_server._preflight_whoami_violation = False
mcp_server._preflight_capability_violation = False
mcp_server._preflight_resolved_role = None
mcp_server._process_start_porcelain = ""
mcp_server._preflight_whoami_baseline_porcelain = None
mcp_server._preflight_capability_baseline_porcelain = None
mcp_server._preflight_whoami_violation_files = []
mcp_server._preflight_capability_violation_files = []
mcp_server._preflight_reviewer_violation_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))
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
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
mcp_server._preflight_capability_violation = self.orig_capability_violation
mcp_server._preflight_resolved_role = self.orig_resolved_role
mcp_server._process_start_porcelain = self.orig_process_start
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
mcp_server._preflight_capability_violation_files = self.orig_capability_files
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
self._dir.cleanup()
os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
os.environ.pop("GITEA_TEST_PORCELAIN", None)
def _env(self, profile: str) -> dict:
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
}
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_preflight_not_called_blocks_commit(self, _auth, mock_api):
mock_api.return_value = {"login": "author-user"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
os.environ["GITEA_TEST_PORCELAIN"] = ""
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api):
mock_api.return_value = {"login": "author-user"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
tests/test_commit_payloads.py
+251
View File
@@ -0,0 +1,251 @@
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))
tests/test_mcp_server.py
+12 -6
View File
@@ -1071,9 +1071,11 @@ class TestGetFile(unittest.TestCase):
# ---------------------------------------------------------------------------
class TestCommitFiles(unittest.TestCase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_commit_files_success(self, _auth, mock_api):
def test_commit_files_success(self, _auth, mock_api, _role):
mock_api.return_value = {
"commit": {"sha": "commit-sha-123"},
"branch": {"name": "test-branch"}
@@ -1081,11 +1083,15 @@ class TestCommitFiles(unittest.TestCase):
files = [
{"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
]
result = gitea_commit_files(
files=files,
message="Initial commit",
new_branch="test-branch"
)
env = {
"GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
}
with patch.dict(os.environ, env, clear=True):
result = gitea_commit_files(
files=files,
message="Initial commit",
new_branch="test-branch"
)
self.assertTrue(result["success"])
self.assertEqual(result["commit"], "commit-sha-123")
self.assertEqual(result["branch"], "test-branch")