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

fix: isolate immutable session context tests (#714)

Browse Source
This commit is contained in:
sysadmin
2026-07-15 02:26:27 -04:00
parent 632c568865
commit 29ffe1407f
7 changed files with 303 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+35 -17
View File
@@ -26,6 +26,14 @@ def _reset_mutation_authority(monkeypatch):
Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
so durable load/save never touches host state even after env clears.
"""
import session_context_binding as session_ctx
# Each pytest item is an independent logical MCP session. Reset both before
# and after the item; the post-yield reset is in a finally block so an
# assertion, exception, or unittest teardown failure cannot pollute the
# next item. Production code has no automatic per-call reset path.
session_ctx._reset_session_context_for_testing()
for env_key in [
"GITEA_SESSION_PROFILE_LOCK",
"GITEA_ACTIVE_WORKTREE",
@@ -80,9 +88,15 @@ def _reset_mutation_authority(monkeypatch):
try:
import mcp_server
except Exception:
_state_tmp.cleanup()
yield
try:
yield
finally:
session_ctx._reset_session_context_for_testing()
_state_tmp.cleanup()
return
import gitea_config
monkeypatch.setattr(gitea_config, "_active_profile_override", None)
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
@@ -113,19 +127,23 @@ def _reset_mutation_authority(monkeypatch):
capability_stop_terminal.clear()
except Exception:
pass
yield
try:
import capability_stop_terminal
capability_stop_terminal.clear()
except Exception:
pass
try:
import review_workflow_load
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
except Exception:
pass
try:
mcp_server._REVIEW_DECISION_LOCK = None
except Exception:
pass
_state_tmp.cleanup()
yield
finally:
try:
import capability_stop_terminal
capability_stop_terminal.clear()
except Exception:
pass
try:
import review_workflow_load
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
except Exception:
pass
try:
mcp_server._REVIEW_DECISION_LOCK = None
except Exception:
pass
gitea_config._active_profile_override = None
session_ctx._reset_session_context_for_testing()
_state_tmp.cleanup()
tests/test_commit_payloads.py
+3
View File
@@ -227,6 +227,7 @@ class TestCommitPayloads(unittest.TestCase):
@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):
mock_api.return_value = {"login": "author-user"}
# Remove active lock file to ensure it fails on traversal/invalid locks
os.remove(self.lock_file_path)
@@ -248,6 +249,7 @@ class TestCommitPayloads(unittest.TestCase):
@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):
mock_api.return_value = {"login": "author-user"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
with self.assertRaises(ValueError) as ctx:
mcp_server.gitea_commit_files(
@@ -266,6 +268,7 @@ class TestCommitPayloads(unittest.TestCase):
@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):
mock_api.return_value = {"login": "author-user"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
with self.assertRaises(ValueError) as ctx:
mcp_server.gitea_commit_files(
tests/test_issue_714_session_context_immutability.py
+110 -10
View File
@@ -6,6 +6,7 @@ import json
import os
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -133,7 +134,6 @@ class TestIssue714SessionContextImmutability(unittest.TestCase):
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
session_ctx.clear_session_context()
self._env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": "mdcps-reviewer",
@@ -147,7 +147,6 @@ class TestIssue714SessionContextImmutability(unittest.TestCase):
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
session_ctx.clear_session_context()
self._dir.cleanup()
def _api_side_effect(self, method, url, header):
@@ -314,7 +313,6 @@ class TestIssue714SessionContextImmutability(unittest.TestCase):
}
with patch.dict(os.environ, env, clear=False):
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
session_ctx.clear_session_context()
gitea_config._active_profile_override = None
res = mcp_server.gitea_resolve_task_capability(
task="create_issue", remote="prgs"
@@ -331,7 +329,6 @@ class TestIssue714SessionContextImmutability(unittest.TestCase):
}
with patch.dict(os.environ, env, clear=False):
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
session_ctx.clear_session_context()
gitea_config._active_profile_override = None
res = mcp_server.gitea_resolve_task_capability(
task="comment_issue", remote="dadeschools"
@@ -342,12 +339,6 @@ class TestIssue714SessionContextImmutability(unittest.TestCase):
class TestSessionContextBindingUnit(unittest.TestCase):
def setUp(self):
session_ctx.clear_session_context()
def tearDown(self):
session_ctx.clear_session_context()
def test_profile_matches_remote_by_base_url(self):
remotes = {
"dadeschools": {"host": "gitea.dadeschools.net"},
@@ -385,6 +376,115 @@ class TestSessionContextBindingUnit(unittest.TestCase):
self.assertTrue(bad["block"])
self.assertTrue(any("drift" in r for r in bad["reasons"]))
def test_bound_context_survives_multiple_calls_and_exceptions(self):
original = session_ctx.bind_session_context(
profile_name="mdcps-reviewer",
remote="dadeschools",
host="gitea.dadeschools.net",
identity="913443",
source="test",
)
try:
for _ in range(3):
observed = session_ctx.seed_session_context_if_unbound(
profile_name="prgs-author",
remote="prgs",
host="gitea.prgs.cc",
identity="jcwalker3",
source="interleaved-test-call",
)
self.assertEqual(observed, original)
raise RuntimeError("simulated caller failure")
except RuntimeError:
pass
self.assertEqual(session_ctx.get_session_context(), original)
drift = session_ctx.assess_session_context(
profile_name="prgs-author",
remote="prgs",
host="gitea.prgs.cc",
identity="jcwalker3",
require_bound=True,
)
self.assertTrue(drift["block"])
def test_parallel_calls_cannot_overwrite_established_binding(self):
original = session_ctx.bind_session_context(
profile_name="mdcps-reviewer",
remote="dadeschools",
host="gitea.dadeschools.net",
identity="913443",
source="test",
)
barrier = threading.Barrier(9)
results = []
results_lock = threading.Lock()
def competing_seed(index: int) -> None:
barrier.wait()
result = session_ctx.seed_session_context_if_unbound(
profile_name=f"prgs-author-{index}",
remote="prgs",
host="gitea.prgs.cc",
identity=f"other-{index}",
source="parallel-test-call",
)
with results_lock:
results.append(result)
threads = [threading.Thread(target=competing_seed, args=(i,)) for i in range(8)]
for thread in threads:
thread.start()
barrier.wait()
for thread in threads:
thread.join(timeout=5)
self.assertFalse(any(thread.is_alive() for thread in threads))
self.assertEqual(results, [original] * 8)
self.assertEqual(session_ctx.get_session_context(), original)
def test_returned_snapshot_cannot_mutate_binding(self):
session_ctx.bind_session_context(
profile_name="mdcps-reviewer",
remote="dadeschools",
host="gitea.dadeschools.net",
identity="913443",
source="test",
)
snapshot = session_ctx.get_session_context()
snapshot["profile_name"] = "prgs-author"
self.assertEqual(
session_ctx.get_session_context()["profile_name"], "mdcps-reviewer"
)
class TestSessionContextTestBoundaryIsolation(unittest.TestCase):
def test_mdcps_binding_starts_clean_and_does_not_escape_test(self):
self.assertIsNone(session_ctx.get_session_context())
session_ctx.bind_session_context(
profile_name="mdcps-reviewer",
remote="dadeschools",
host="gitea.dadeschools.net",
identity="913443",
source="test-boundary",
)
def test_prgs_binding_starts_clean_and_does_not_escape_test(self):
self.assertIsNone(session_ctx.get_session_context())
session_ctx.bind_session_context(
profile_name="prgs-author",
remote="prgs",
host="gitea.prgs.cc",
identity="jcwalker3",
source="test-boundary",
)
def test_reset_helper_is_rejected_outside_pytest_boundary(self):
with patch.dict(os.environ, {}, clear=True):
with self.assertRaises(RuntimeError):
session_ctx._reset_session_context_for_testing()
if __name__ == "__main__":
unittest.main()
tests/test_operation_scoped_roles.py
-3
View File
@@ -16,7 +16,6 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
import gitea_config
import gitea_auth
import mcp_server
import session_context_binding as session_ctx
from reviewer_worktree import assess_author_worktree_continuity
CONFIG_TEST = {
@@ -68,7 +67,6 @@ class TestOperationScopedRoles(unittest.TestCase):
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
session_ctx.clear_session_context()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
self._write_config(CONFIG_TEST)
@@ -78,7 +76,6 @@ class TestOperationScopedRoles(unittest.TestCase):
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
session_ctx.clear_session_context()
self._dir.cleanup()
def _write_config(self, obj):
tests/test_resolve_task_capability.py
+2 -2
View File
@@ -188,8 +188,8 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertEqual(res["required_role_kind"], "author")
self.assertTrue(res["allowed_in_current_session"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
with patch.dict(os.environ, self._env("reviewer-profile")):
res = mcp_server.gitea_resolve_task_capability(