nearxos
1d71b3f822
Surface Cursor run progress to stderr and response payloads, enforce CURSOR_TIMEOUT_SECONDS, validate git repos before delegate, and fix resume session cwd persistence. Includes docs, tests, and packaging. Co-authored-by: Cursor <cursoragent@cursor.com>
77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from cursor_mcp.errors import CursorMCPError
|
|
from cursor_mcp.repo import validate_git_repo
|
|
from cursor_mcp.sessions import get_session, save_session
|
|
|
|
|
|
class RepoValidationTests(unittest.TestCase):
|
|
def test_requires_git_repo_by_default(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
workdir = Path(tmp)
|
|
with self.assertRaises(CursorMCPError) as ctx:
|
|
validate_git_repo(workdir, required=True)
|
|
self.assertIn("not a git repository", str(ctx.exception))
|
|
|
|
def test_accepts_git_repo(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
workdir = Path(tmp)
|
|
subprocess.run(["git", "init"], cwd=workdir, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "commit", "--allow-empty", "-m", "init"],
|
|
cwd=workdir,
|
|
check=True,
|
|
capture_output=True,
|
|
env={
|
|
**os.environ,
|
|
"GIT_AUTHOR_NAME": "test",
|
|
"GIT_AUTHOR_EMAIL": "test@example.com",
|
|
"GIT_COMMITTER_NAME": "test",
|
|
"GIT_COMMITTER_EMAIL": "test@example.com",
|
|
},
|
|
)
|
|
info = validate_git_repo(workdir, required=True)
|
|
self.assertIsNotNone(info)
|
|
assert info is not None
|
|
self.assertIn(info["branch"], {"main", "master"})
|
|
self.assertFalse(info["dirty"])
|
|
|
|
def test_optional_validation(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
self.assertIsNone(validate_git_repo(Path(tmp), required=False))
|
|
|
|
|
|
class SessionTests(unittest.TestCase):
|
|
def test_get_session_returns_saved_cwd(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "sessions.json"
|
|
from cursor_mcp import config
|
|
|
|
original = config.settings.sessions_path
|
|
config.settings.sessions_path = str(path)
|
|
try:
|
|
save_session(
|
|
agent_id="agent-123",
|
|
run_id="run-1",
|
|
cwd="/tmp/project-a",
|
|
task_summary="first task",
|
|
status="finished",
|
|
result_preview="done",
|
|
)
|
|
entry = get_session("agent-123")
|
|
self.assertIsNotNone(entry)
|
|
assert entry is not None
|
|
self.assertEqual(entry["cwd"], "/tmp/project-a")
|
|
finally:
|
|
config.settings.sessions_path = original
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|