Share review-workflow-load and review-decision-lock state across MCP daemon process pools via a user-private durable store keyed by profile identity (not host-global /tmp), with TTL and fail-closed profile checks.
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
"""Shared pytest fixtures for the Gitea-Tools test suite."""
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_mutation_authority(monkeypatch):
|
|
"""Isolate the in-process mutation authority between tests (#199).
|
|
|
|
The mutation-authority gate stays LIVE in every test — this fixture only
|
|
clears the per-process record and the session profile lock so one test's
|
|
seeded authority (or an intentionally mismatched one) cannot leak into
|
|
the next test. It must never replace verify_mutation_authority with a
|
|
no-op: individual tests that need a specific authority state set it up
|
|
explicitly.
|
|
"""
|
|
monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
|
|
# Isolate durable session-state files so tests never share host cache (#559).
|
|
import tempfile
|
|
|
|
_state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-")
|
|
monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name)
|
|
try:
|
|
import mcp_server
|
|
except Exception:
|
|
_state_tmp.cleanup()
|
|
yield
|
|
return
|
|
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
|
|
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
|
|
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
|
|
try:
|
|
import review_workflow_load
|
|
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
|
|
except Exception:
|
|
pass
|
|
try:
|
|
import capability_stop_terminal
|
|
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()
|