Gitea-Tools/tests/test_mcp_session_state.py

"""Tests for durable MCP session state shared across daemon processes (#559)."""

from __future__ import annotations

import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

sys_path_root = str(Path(__file__).resolve().parent.parent)
import sys

if sys_path_root not in sys.path:
    sys.path.insert(0, sys_path_root)

import mcp_session_state
import review_workflow_load
import mcp_server


class TestMcpSessionStateStore(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self.state_dir = self._tmpdir.name
        self._env = patch.dict(
            os.environ,
            {
                mcp_session_state.STATE_DIR_ENV: self.state_dir,
                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
                "GITEA_MCP_PROFILE": "prgs-reviewer",
            },
            clear=False,
        )
        self._env.start()

    def tearDown(self):
        self._env.stop()
        self._tmpdir.cleanup()

    def test_round_trip_same_identity(self):
        saved = mcp_session_state.save_state(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            payload={"loaded": True, "workflow_hash": "abc123def456"},
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertIsNotNone(saved)
        self.assertEqual(saved["workflow_hash"], "abc123def456")
        self.assertIn("recorded_at", saved)

        loaded = mcp_session_state.load_state(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertIsNotNone(loaded)
        self.assertEqual(loaded["workflow_hash"], "abc123def456")
        self.assertEqual(loaded["profile_identity"], "prgs-reviewer")

    def test_profile_mismatch_returns_none(self):
        mcp_session_state.save_state(
            kind=mcp_session_state.KIND_DECISION_LOCK,
            payload={"final_review_decision_ready": True, "remote": "prgs"},
            remote="prgs",
        )
        with patch.dict(
            os.environ,
            {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
            clear=False,
        ):
            loaded = mcp_session_state.load_state(
                kind=mcp_session_state.KIND_DECISION_LOCK,
                remote="prgs",
            )
        self.assertIsNone(loaded)

    def test_clear_removes_file(self):
        mcp_session_state.save_state(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            payload={"loaded": True},
            remote="prgs",
        )
        path = mcp_session_state.state_file_path(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            remote="prgs",
            profile_identity="prgs-reviewer",
            state_dir=self.state_dir,
        )
        self.assertTrue(os.path.exists(path))
        mcp_session_state.clear_state(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            remote="prgs",
            profile_identity="prgs-reviewer",
        )
        self.assertFalse(os.path.exists(path))

    def test_files_are_private_mode(self):
        mcp_session_state.save_state(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            payload={"loaded": True},
            remote="prgs",
        )
        path = mcp_session_state.state_file_path(
            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
            remote="prgs",
            profile_identity="prgs-reviewer",
            state_dir=self.state_dir,
        )
        mode = os.stat(path).st_mode & 0o777
        self.assertEqual(mode, 0o600)


class TestWorkflowLoadCrossProcess(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self._env = patch.dict(
            os.environ,
            {
                mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
                "GITEA_MCP_PROFILE": "prgs-reviewer",
                "GITEA_MCP_REMOTE": "prgs",
            },
            clear=False,
        )
        self._env.start()
        review_workflow_load.clear_review_workflow_load()

    def tearDown(self):
        review_workflow_load.clear_review_workflow_load()
        self._env.stop()
        self._tmpdir.cleanup()

    def test_different_pid_same_profile_accepted(self):
        root = str(Path(__file__).resolve().parent.parent)
        recorded = review_workflow_load.record_review_workflow_load(root)
        self.assertTrue(recorded["loaded"])

        # Simulate another daemon process: clear memory, keep durable file,
        # and change apparent PID identity only (profile remains the same).
        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
        status = review_workflow_load.workflow_load_status(root)
        self.assertTrue(status["workflow_load_proof_present"])
        self.assertTrue(
            status["workflow_load_valid"],
            msg=status.get("reasons"),
        )

    def test_profile_mismatch_blocks(self):
        root = str(Path(__file__).resolve().parent.parent)
        review_workflow_load.record_review_workflow_load(root)
        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
        with patch.dict(
            os.environ,
            {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
            clear=False,
        ):
            # Durable load uses active identity; mismatched profile key misses.
            status = review_workflow_load.workflow_load_status(root)
        self.assertFalse(status["workflow_load_proof_present"])


class TestDecisionLockCrossProcess(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self._env = patch.dict(
            os.environ,
            {
                mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
            },
            clear=False,
        )
        self._env.start()
        mcp_server._save_review_decision_lock(None)
        review_workflow_load.clear_review_workflow_load()

    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        review_workflow_load.clear_review_workflow_load()
        self._env.stop()
        self._tmpdir.cleanup()

    def test_decision_lock_survives_memory_clear(self):
        with patch.object(mcp_server, "get_profile", return_value={
            "profile_name": "prgs-reviewer",
        }):
            mcp_server.init_review_decision_lock("prgs", "review_pr", force=True)
            lock = mcp_server._load_review_decision_lock()
            self.assertIsNotNone(lock)
            self.assertEqual(lock["remote"], "prgs")
            self.assertFalse(lock["final_review_decision_ready"])

            # New process: memory empty, durable state remains.
            mcp_server._REVIEW_DECISION_LOCK = None
            restored = mcp_server._load_review_decision_lock()
            self.assertIsNotNone(restored)
            self.assertEqual(restored["remote"], "prgs")
            reasons = mcp_server._review_decision_session_reasons(restored)
            self.assertEqual(reasons, [])


if __name__ == "__main__":
    unittest.main()