FossilRepo

fossilrepo / tests / test_mcp_server.py
Source Blame History 710 lines
c588255… ragelink 1 """Tests for MCP server tool definitions and handlers.
c588255… ragelink 2
c588255… ragelink 3 Covers:
c588255… ragelink 4 - Tool registry: all 17 tools registered with correct schemas
c588255… ragelink 5 - Tool dispatch: execute_tool routes to correct handler
c588255… ragelink 6 - Read handlers: list_projects, get_project, browse_code, read_file,
c588255… ragelink 7 get_timeline, get_checkin, search_code, list_tickets, get_ticket,
c588255… ragelink 8 list_wiki_pages, get_wiki_page, list_branches, get_file_blame,
c588255… ragelink 9 get_file_history, sql_query
c588255… ragelink 10 - Write handlers: create_ticket, update_ticket
c588255… ragelink 11 - Error handling: unknown tool, missing project, exceptions
c588255… ragelink 12 """
c588255… ragelink 13
c588255… ragelink 14 from datetime import UTC, datetime
c588255… ragelink 15 from unittest.mock import MagicMock, patch
c588255… ragelink 16
c588255… ragelink 17 import pytest
c588255… ragelink 18
c588255… ragelink 19 from fossil.models import FossilRepository
c588255… ragelink 20 from fossil.reader import (
c588255… ragelink 21 CheckinDetail,
c588255… ragelink 22 FileEntry,
c588255… ragelink 23 RepoMetadata,
c588255… ragelink 24 TicketEntry,
c588255… ragelink 25 TimelineEntry,
c588255… ragelink 26 WikiPage,
c588255… ragelink 27 )
c588255… ragelink 28 from mcp_server.tools import TOOLS, execute_tool
c588255… ragelink 29
c588255… ragelink 30 # Patch targets -- tools.py does deferred imports inside handler functions,
c588255… ragelink 31 # so we patch at the source module rather than at the consumer.
c588255… ragelink 32 _READER = "fossil.reader.FossilReader"
c588255… ragelink 33 _CLI = "fossil.cli.FossilCLI"
c588255… ragelink 34
c588255… ragelink 35 # ---------------------------------------------------------------------------
c588255… ragelink 36 # Fixtures
c588255… ragelink 37 # ---------------------------------------------------------------------------
c588255… ragelink 38
c588255… ragelink 39
c588255… ragelink 40 @pytest.fixture
c588255… ragelink 41 def fossil_repo_obj(sample_project):
c588255… ragelink 42 """Return the auto-created FossilRepository for sample_project."""
c588255… ragelink 43 return FossilRepository.objects.get(project=sample_project, deleted_at__isnull=True)
c588255… ragelink 44
c588255… ragelink 45
c588255… ragelink 46 def _mock_reader():
c588255… ragelink 47 """Return a context-manager mock for FossilReader."""
c588255… ragelink 48 reader = MagicMock()
c588255… ragelink 49 reader.__enter__ = MagicMock(return_value=reader)
c588255… ragelink 50 reader.__exit__ = MagicMock(return_value=False)
c588255… ragelink 51 return reader
c588255… ragelink 52
c588255… ragelink 53
c588255… ragelink 54 # ---------------------------------------------------------------------------
c588255… ragelink 55 # Tool registry tests
c588255… ragelink 56 # ---------------------------------------------------------------------------
c588255… ragelink 57
c588255… ragelink 58
c588255… ragelink 59 class TestToolRegistry:
c588255… ragelink 60 def test_all_17_tools_registered(self):
c588255… ragelink 61 assert len(TOOLS) == 17
c588255… ragelink 62
c588255… ragelink 63 def test_tool_names_are_unique(self):
c588255… ragelink 64 names = [t.name for t in TOOLS]
c588255… ragelink 65 assert len(names) == len(set(names))
c588255… ragelink 66
c588255… ragelink 67 def test_every_tool_has_input_schema(self):
c588255… ragelink 68 for tool in TOOLS:
c588255… ragelink 69 assert tool.inputSchema is not None
c588255… ragelink 70 assert tool.inputSchema.get("type") == "object"
c588255… ragelink 71
c588255… ragelink 72 def test_every_tool_has_description(self):
c588255… ragelink 73 for tool in TOOLS:
c588255… ragelink 74 assert tool.description
c588255… ragelink 75 assert len(tool.description) > 10
c588255… ragelink 76
c588255… ragelink 77 def test_expected_tools_present(self):
c588255… ragelink 78 names = {t.name for t in TOOLS}
c588255… ragelink 79 expected = {
c588255… ragelink 80 "list_projects",
c588255… ragelink 81 "get_project",
c588255… ragelink 82 "browse_code",
c588255… ragelink 83 "read_file",
c588255… ragelink 84 "get_timeline",
c588255… ragelink 85 "get_checkin",
c588255… ragelink 86 "search_code",
c588255… ragelink 87 "list_tickets",
c588255… ragelink 88 "get_ticket",
c588255… ragelink 89 "create_ticket",
c588255… ragelink 90 "update_ticket",
c588255… ragelink 91 "list_wiki_pages",
c588255… ragelink 92 "get_wiki_page",
c588255… ragelink 93 "list_branches",
c588255… ragelink 94 "get_file_blame",
c588255… ragelink 95 "get_file_history",
c588255… ragelink 96 "sql_query",
c588255… ragelink 97 }
c588255… ragelink 98 assert names == expected
c588255… ragelink 99
c588255… ragelink 100 def test_slug_required_for_project_scoped_tools(self):
c588255… ragelink 101 """All tools except list_projects require a slug parameter."""
c588255… ragelink 102 for tool in TOOLS:
c588255… ragelink 103 if tool.name == "list_projects":
c588255… ragelink 104 assert "slug" not in tool.inputSchema.get("required", [])
c588255… ragelink 105 else:
c588255… ragelink 106 assert "slug" in tool.inputSchema.get("required", []), f"{tool.name} should require slug"
c588255… ragelink 107
c588255… ragelink 108
c588255… ragelink 109 # ---------------------------------------------------------------------------
c588255… ragelink 110 # Dispatch tests
c588255… ragelink 111 # ---------------------------------------------------------------------------
c588255… ragelink 112
c588255… ragelink 113
c588255… ragelink 114 class TestDispatch:
c588255… ragelink 115 def test_unknown_tool_returns_error(self):
c588255… ragelink 116 result = execute_tool("nonexistent_tool", {})
c588255… ragelink 117 assert "error" in result
c588255… ragelink 118 assert "Unknown tool" in result["error"]
c588255… ragelink 119
c588255… ragelink 120 @pytest.mark.django_db
c588255… ragelink 121 def test_missing_project_returns_error(self):
c588255… ragelink 122 result = execute_tool("get_project", {"slug": "does-not-exist"})
c588255… ragelink 123 assert "error" in result
c588255… ragelink 124
c588255… ragelink 125 @pytest.mark.django_db
c588255… ragelink 126 def test_exception_in_handler_returns_error(self, sample_project):
c588255… ragelink 127 with patch("mcp_server.tools._get_repo", side_effect=RuntimeError("boom")):
c588255… ragelink 128 result = execute_tool("get_project", {"slug": sample_project.slug})
c588255… ragelink 129 assert "error" in result
c588255… ragelink 130 assert "boom" in result["error"]
c588255… ragelink 131
c588255… ragelink 132
c588255… ragelink 133 # ---------------------------------------------------------------------------
c588255… ragelink 134 # list_projects
c588255… ragelink 135 # ---------------------------------------------------------------------------
c588255… ragelink 136
c588255… ragelink 137
c588255… ragelink 138 @pytest.mark.django_db
c588255… ragelink 139 class TestListProjects:
c588255… ragelink 140 def test_returns_all_active_projects(self, sample_project):
c588255… ragelink 141 result = execute_tool("list_projects", {})
c588255… ragelink 142 assert "projects" in result
c588255… ragelink 143 slugs = [p["slug"] for p in result["projects"]]
c588255… ragelink 144 assert sample_project.slug in slugs
c588255… ragelink 145
c588255… ragelink 146 def test_excludes_deleted_projects(self, sample_project, admin_user):
c588255… ragelink 147 sample_project.soft_delete(user=admin_user)
c588255… ragelink 148 result = execute_tool("list_projects", {})
c588255… ragelink 149 slugs = [p["slug"] for p in result["projects"]]
c588255… ragelink 150 assert sample_project.slug not in slugs
c588255… ragelink 151
c588255… ragelink 152
c588255… ragelink 153 # ---------------------------------------------------------------------------
c588255… ragelink 154 # get_project
c588255… ragelink 155 # ---------------------------------------------------------------------------
c588255… ragelink 156
c588255… ragelink 157
c588255… ragelink 158 @pytest.mark.django_db
c588255… ragelink 159 class TestGetProject:
c588255… ragelink 160 @patch(_READER)
c588255… ragelink 161 def test_returns_project_details(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 162 reader = _mock_reader()
c588255… ragelink 163 reader.get_metadata.return_value = RepoMetadata(
c588255… ragelink 164 project_name="Test",
c588255… ragelink 165 checkin_count=10,
c588255… ragelink 166 ticket_count=3,
c588255… ragelink 167 wiki_page_count=2,
c588255… ragelink 168 )
c588255… ragelink 169 mock_reader_cls.return_value = reader
c588255… ragelink 170
c588255… ragelink 171 with patch.object(type(fossil_repo_obj), "exists_on_disk", new_callable=lambda: property(lambda s: True)):
c588255… ragelink 172 result = execute_tool("get_project", {"slug": sample_project.slug})
c588255… ragelink 173
c588255… ragelink 174 assert result["name"] == sample_project.name
c588255… ragelink 175 assert result["slug"] == sample_project.slug
c588255… ragelink 176 assert result["visibility"] == sample_project.visibility
c588255… ragelink 177
c588255… ragelink 178 def test_nonexistent_slug_returns_error(self):
c588255… ragelink 179 result = execute_tool("get_project", {"slug": "no-such-project"})
c588255… ragelink 180 assert "error" in result
c588255… ragelink 181
c588255… ragelink 182
c588255… ragelink 183 # ---------------------------------------------------------------------------
c588255… ragelink 184 # browse_code
c588255… ragelink 185 # ---------------------------------------------------------------------------
c588255… ragelink 186
c588255… ragelink 187
c588255… ragelink 188 @pytest.mark.django_db
c588255… ragelink 189 class TestBrowseCode:
c588255… ragelink 190 @patch(_READER)
c588255… ragelink 191 def test_lists_files_at_root(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 192 reader = _mock_reader()
c588255… ragelink 193 reader.get_latest_checkin_uuid.return_value = "abc123"
c588255… ragelink 194 reader.get_files_at_checkin.return_value = [
c588255… ragelink 195 FileEntry(name="README.md", uuid="f1", size=100),
c588255… ragelink 196 FileEntry(name="src/main.py", uuid="f2", size=200),
c588255… ragelink 197 ]
c588255… ragelink 198 mock_reader_cls.return_value = reader
c588255… ragelink 199
c588255… ragelink 200 result = execute_tool("browse_code", {"slug": sample_project.slug})
c588255… ragelink 201 assert len(result["files"]) == 2
c588255… ragelink 202 assert result["checkin"] == "abc123"
c588255… ragelink 203
c588255… ragelink 204 @patch(_READER)
c588255… ragelink 205 def test_filters_by_path(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 206 reader = _mock_reader()
c588255… ragelink 207 reader.get_latest_checkin_uuid.return_value = "abc123"
c588255… ragelink 208 reader.get_files_at_checkin.return_value = [
c588255… ragelink 209 FileEntry(name="README.md", uuid="f1", size=100),
c588255… ragelink 210 FileEntry(name="src/main.py", uuid="f2", size=200),
c588255… ragelink 211 FileEntry(name="src/utils.py", uuid="f3", size=150),
c588255… ragelink 212 ]
c588255… ragelink 213 mock_reader_cls.return_value = reader
c588255… ragelink 214
c588255… ragelink 215 result = execute_tool("browse_code", {"slug": sample_project.slug, "path": "src"})
c588255… ragelink 216 assert len(result["files"]) == 2
c588255… ragelink 217 assert all(f["name"].startswith("src/") for f in result["files"])
c588255… ragelink 218
c588255… ragelink 219 @patch(_READER)
c588255… ragelink 220 def test_empty_repo_returns_error(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 221 reader = _mock_reader()
c588255… ragelink 222 reader.get_latest_checkin_uuid.return_value = None
c588255… ragelink 223 mock_reader_cls.return_value = reader
c588255… ragelink 224
c588255… ragelink 225 result = execute_tool("browse_code", {"slug": sample_project.slug})
c588255… ragelink 226 assert "error" in result
c588255… ragelink 227
c588255… ragelink 228
c588255… ragelink 229 # ---------------------------------------------------------------------------
c588255… ragelink 230 # read_file
c588255… ragelink 231 # ---------------------------------------------------------------------------
c588255… ragelink 232
c588255… ragelink 233
c588255… ragelink 234 @pytest.mark.django_db
c588255… ragelink 235 class TestReadFile:
c588255… ragelink 236 @patch(_READER)
c588255… ragelink 237 def test_reads_text_file(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 238 reader = _mock_reader()
c588255… ragelink 239 reader.get_latest_checkin_uuid.return_value = "abc123"
c588255… ragelink 240 reader.get_files_at_checkin.return_value = [
c588255… ragelink 241 FileEntry(name="README.md", uuid="f1", size=100),
c588255… ragelink 242 ]
c588255… ragelink 243 reader.get_file_content.return_value = b"# Hello World"
c588255… ragelink 244 mock_reader_cls.return_value = reader
c588255… ragelink 245
c588255… ragelink 246 result = execute_tool("read_file", {"slug": sample_project.slug, "filepath": "README.md"})
c588255… ragelink 247 assert result["filepath"] == "README.md"
c588255… ragelink 248 assert result["content"] == "# Hello World"
c588255… ragelink 249
c588255… ragelink 250 @patch(_READER)
c588255… ragelink 251 def test_binary_file_returns_metadata(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 252 reader = _mock_reader()
c588255… ragelink 253 reader.get_latest_checkin_uuid.return_value = "abc123"
c588255… ragelink 254 reader.get_files_at_checkin.return_value = [
c588255… ragelink 255 FileEntry(name="image.png", uuid="f1", size=5000),
c588255… ragelink 256 ]
c588255… ragelink 257 reader.get_file_content.return_value = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00"
c588255… ragelink 258 mock_reader_cls.return_value = reader
c588255… ragelink 259
c588255… ragelink 260 result = execute_tool("read_file", {"slug": sample_project.slug, "filepath": "image.png"})
c588255… ragelink 261 assert result["binary"] is True
c588255… ragelink 262 assert result["size"] > 0
c588255… ragelink 263
c588255… ragelink 264 @patch(_READER)
c588255… ragelink 265 def test_file_not_found(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 266 reader = _mock_reader()
c588255… ragelink 267 reader.get_latest_checkin_uuid.return_value = "abc123"
c588255… ragelink 268 reader.get_files_at_checkin.return_value = []
c588255… ragelink 269 mock_reader_cls.return_value = reader
c588255… ragelink 270
c588255… ragelink 271 result = execute_tool("read_file", {"slug": sample_project.slug, "filepath": "nope.txt"})
c588255… ragelink 272 assert "error" in result
c588255… ragelink 273 assert "not found" in result["error"].lower()
c588255… ragelink 274
c588255… ragelink 275
c588255… ragelink 276 # ---------------------------------------------------------------------------
c588255… ragelink 277 # get_timeline
c588255… ragelink 278 # ---------------------------------------------------------------------------
c588255… ragelink 279
c588255… ragelink 280
c588255… ragelink 281 @pytest.mark.django_db
c588255… ragelink 282 class TestGetTimeline:
c588255… ragelink 283 @patch(_READER)
c588255… ragelink 284 def test_returns_checkins(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 285 reader = _mock_reader()
c588255… ragelink 286 reader.get_timeline.return_value = [
c588255… ragelink 287 TimelineEntry(
c588255… ragelink 288 rid=1,
c588255… ragelink 289 uuid="abc123",
c588255… ragelink 290 event_type="ci",
c588255… ragelink 291 timestamp=datetime(2025, 1, 15, 10, 0, 0, tzinfo=UTC),
c588255… ragelink 292 user="alice",
c588255… ragelink 293 comment="Initial commit",
c588255… ragelink 294 branch="trunk",
c588255… ragelink 295 ),
c588255… ragelink 296 ]
c588255… ragelink 297 mock_reader_cls.return_value = reader
c588255… ragelink 298
c588255… ragelink 299 result = execute_tool("get_timeline", {"slug": sample_project.slug})
c588255… ragelink 300 assert len(result["checkins"]) == 1
c588255… ragelink 301 assert result["checkins"][0]["uuid"] == "abc123"
c588255… ragelink 302 assert result["checkins"][0]["user"] == "alice"
c588255… ragelink 303
c588255… ragelink 304 @patch(_READER)
c588255… ragelink 305 def test_branch_filter(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 306 reader = _mock_reader()
c588255… ragelink 307 reader.get_timeline.return_value = [
c588255… ragelink 308 TimelineEntry(
c588255… ragelink 309 rid=1,
c588255… ragelink 310 uuid="a1",
c588255… ragelink 311 event_type="ci",
c588255… ragelink 312 timestamp=datetime(2025, 1, 15, tzinfo=UTC),
c588255… ragelink 313 user="alice",
c588255… ragelink 314 comment="on trunk",
c588255… ragelink 315 branch="trunk",
c588255… ragelink 316 ),
c588255… ragelink 317 TimelineEntry(
c588255… ragelink 318 rid=2,
c588255… ragelink 319 uuid="b2",
c588255… ragelink 320 event_type="ci",
c588255… ragelink 321 timestamp=datetime(2025, 1, 14, tzinfo=UTC),
c588255… ragelink 322 user="bob",
c588255… ragelink 323 comment="on feature",
c588255… ragelink 324 branch="feature-x",
c588255… ragelink 325 ),
c588255… ragelink 326 ]
c588255… ragelink 327 mock_reader_cls.return_value = reader
c588255… ragelink 328
c588255… ragelink 329 result = execute_tool("get_timeline", {"slug": sample_project.slug, "branch": "trunk"})
c588255… ragelink 330 assert len(result["checkins"]) == 1
c588255… ragelink 331 assert result["checkins"][0]["branch"] == "trunk"
c588255… ragelink 332
c588255… ragelink 333
c588255… ragelink 334 # ---------------------------------------------------------------------------
c588255… ragelink 335 # get_checkin
c588255… ragelink 336 # ---------------------------------------------------------------------------
c588255… ragelink 337
c588255… ragelink 338
c588255… ragelink 339 @pytest.mark.django_db
c588255… ragelink 340 class TestGetCheckin:
c588255… ragelink 341 @patch(_READER)
c588255… ragelink 342 def test_returns_checkin_detail(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 343 reader = _mock_reader()
c588255… ragelink 344 reader.get_checkin_detail.return_value = CheckinDetail(
c588255… ragelink 345 uuid="abc123full",
c588255… ragelink 346 timestamp=datetime(2025, 1, 15, 10, 0, 0, tzinfo=UTC),
c588255… ragelink 347 user="alice",
c588255… ragelink 348 comment="Initial commit",
c588255… ragelink 349 branch="trunk",
c588255… ragelink 350 parent_uuid="parent000",
c588255… ragelink 351 files_changed=[{"name": "README.md", "change_type": "added", "uuid": "f1", "prev_uuid": ""}],
c588255… ragelink 352 )
c588255… ragelink 353 mock_reader_cls.return_value = reader
c588255… ragelink 354
c588255… ragelink 355 result = execute_tool("get_checkin", {"slug": sample_project.slug, "uuid": "abc123"})
c588255… ragelink 356 assert result["uuid"] == "abc123full"
c588255… ragelink 357 assert len(result["files_changed"]) == 1
c588255… ragelink 358
c588255… ragelink 359 @patch(_READER)
c588255… ragelink 360 def test_checkin_not_found(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 361 reader = _mock_reader()
c588255… ragelink 362 reader.get_checkin_detail.return_value = None
c588255… ragelink 363 mock_reader_cls.return_value = reader
c588255… ragelink 364
c588255… ragelink 365 result = execute_tool("get_checkin", {"slug": sample_project.slug, "uuid": "nonexistent"})
c588255… ragelink 366 assert "error" in result
c588255… ragelink 367
c588255… ragelink 368
c588255… ragelink 369 # ---------------------------------------------------------------------------
c588255… ragelink 370 # search_code
c588255… ragelink 371 # ---------------------------------------------------------------------------
c588255… ragelink 372
c588255… ragelink 373
c588255… ragelink 374 @pytest.mark.django_db
c588255… ragelink 375 class TestSearchCode:
c588255… ragelink 376 @patch(_READER)
c588255… ragelink 377 def test_returns_search_results(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 378 reader = _mock_reader()
c588255… ragelink 379 reader.search.return_value = {
c588255… ragelink 380 "checkins": [{"uuid": "abc", "timestamp": datetime(2025, 1, 15, tzinfo=UTC), "user": "alice", "comment": "fix bug"}],
c588255… ragelink 381 "tickets": [{"uuid": "tkt1", "title": "Bug report", "status": "Open", "created": datetime(2025, 1, 10, tzinfo=UTC)}],
c588255… ragelink 382 "wiki": [{"name": "Debugging"}],
c588255… ragelink 383 }
c588255… ragelink 384 mock_reader_cls.return_value = reader
c588255… ragelink 385
c588255… ragelink 386 result = execute_tool("search_code", {"slug": sample_project.slug, "query": "bug"})
c588255… ragelink 387 assert len(result["checkins"]) == 1
c588255… ragelink 388 assert len(result["tickets"]) == 1
c588255… ragelink 389 assert len(result["wiki"]) == 1
c588255… ragelink 390 # Timestamps should be serialized to strings
c588255… ragelink 391 assert isinstance(result["checkins"][0]["timestamp"], str)
c588255… ragelink 392
c588255… ragelink 393
c588255… ragelink 394 # ---------------------------------------------------------------------------
c588255… ragelink 395 # list_tickets / get_ticket
c588255… ragelink 396 # ---------------------------------------------------------------------------
c588255… ragelink 397
c588255… ragelink 398
c588255… ragelink 399 @pytest.mark.django_db
c588255… ragelink 400 class TestTickets:
c588255… ragelink 401 @patch(_READER)
c588255… ragelink 402 def test_list_tickets(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 403 reader = _mock_reader()
c588255… ragelink 404 reader.get_tickets.return_value = [
c588255… ragelink 405 TicketEntry(
c588255… ragelink 406 uuid="tkt-001",
c588255… ragelink 407 title="Fix bug",
c588255… ragelink 408 status="Open",
c588255… ragelink 409 type="Code_Defect",
c588255… ragelink 410 created=datetime(2025, 1, 10, tzinfo=UTC),
c588255… ragelink 411 owner="alice",
c588255… ragelink 412 priority="High",
c588255… ragelink 413 ),
c588255… ragelink 414 ]
c588255… ragelink 415 mock_reader_cls.return_value = reader
c588255… ragelink 416
c588255… ragelink 417 result = execute_tool("list_tickets", {"slug": sample_project.slug})
c588255… ragelink 418 assert len(result["tickets"]) == 1
c588255… ragelink 419 assert result["tickets"][0]["uuid"] == "tkt-001"
c588255… ragelink 420
c588255… ragelink 421 @patch(_READER)
c588255… ragelink 422 def test_get_ticket_detail(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 423 reader = _mock_reader()
c588255… ragelink 424 reader.get_ticket_detail.return_value = TicketEntry(
c588255… ragelink 425 uuid="tkt-001",
c588255… ragelink 426 title="Fix bug",
c588255… ragelink 427 status="Open",
c588255… ragelink 428 type="Code_Defect",
c588255… ragelink 429 created=datetime(2025, 1, 10, tzinfo=UTC),
c588255… ragelink 430 owner="alice",
c588255… ragelink 431 body="Detailed description",
c588255… ragelink 432 priority="High",
c588255… ragelink 433 severity="Critical",
c588255… ragelink 434 )
c588255… ragelink 435 reader.get_ticket_comments.return_value = [
c588255… ragelink 436 {"timestamp": datetime(2025, 1, 11, tzinfo=UTC), "user": "bob", "comment": "Reproduced", "mimetype": "text/plain"},
c588255… ragelink 437 ]
c588255… ragelink 438 mock_reader_cls.return_value = reader
c588255… ragelink 439
c588255… ragelink 440 result = execute_tool("get_ticket", {"slug": sample_project.slug, "uuid": "tkt-001"})
c588255… ragelink 441 assert result["title"] == "Fix bug"
c588255… ragelink 442 assert result["body"] == "Detailed description"
c588255… ragelink 443 assert len(result["comments"]) == 1
c588255… ragelink 444
c588255… ragelink 445 @patch(_READER)
c588255… ragelink 446 def test_ticket_not_found(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 447 reader = _mock_reader()
c588255… ragelink 448 reader.get_ticket_detail.return_value = None
c588255… ragelink 449 mock_reader_cls.return_value = reader
c588255… ragelink 450
c588255… ragelink 451 result = execute_tool("get_ticket", {"slug": sample_project.slug, "uuid": "nonexistent"})
c588255… ragelink 452 assert "error" in result
c588255… ragelink 453
c588255… ragelink 454
c588255… ragelink 455 # ---------------------------------------------------------------------------
c588255… ragelink 456 # create_ticket / update_ticket
c588255… ragelink 457 # ---------------------------------------------------------------------------
c588255… ragelink 458
c588255… ragelink 459
c588255… ragelink 460 @pytest.mark.django_db
c588255… ragelink 461 class TestWriteTickets:
c588255… ragelink 462 @patch(_CLI)
c588255… ragelink 463 def test_create_ticket(self, mock_cli_cls, sample_project, fossil_repo_obj):
c588255… ragelink 464 cli = MagicMock()
c588255… ragelink 465 cli.ticket_add.return_value = True
c588255… ragelink 466 mock_cli_cls.return_value = cli
c588255… ragelink 467
c588255… ragelink 468 result = execute_tool(
c588255… ragelink 469 "create_ticket",
c588255… ragelink 470 {
c588255… ragelink 471 "slug": sample_project.slug,
c588255… ragelink 472 "title": "New bug",
c588255… ragelink 473 "body": "Something is broken",
c588255… ragelink 474 },
c588255… ragelink 475 )
c588255… ragelink 476 assert result["success"] is True
c588255… ragelink 477 assert result["title"] == "New bug"
c588255… ragelink 478 cli.ticket_add.assert_called_once()
c588255… ragelink 479 call_args = cli.ticket_add.call_args
c588255… ragelink 480 fields = call_args[0][1]
c588255… ragelink 481 assert fields["title"] == "New bug"
c588255… ragelink 482 assert fields["comment"] == "Something is broken"
c588255… ragelink 483 assert fields["status"] == "Open"
c588255… ragelink 484
c588255… ragelink 485 @patch(_CLI)
c588255… ragelink 486 def test_create_ticket_failure(self, mock_cli_cls, sample_project, fossil_repo_obj):
c588255… ragelink 487 cli = MagicMock()
c588255… ragelink 488 cli.ticket_add.return_value = False
c588255… ragelink 489 mock_cli_cls.return_value = cli
c588255… ragelink 490
c588255… ragelink 491 result = execute_tool(
c588255… ragelink 492 "create_ticket",
c588255… ragelink 493 {
c588255… ragelink 494 "slug": sample_project.slug,
c588255… ragelink 495 "title": "Failing",
c588255… ragelink 496 "body": "Will fail",
c588255… ragelink 497 },
c588255… ragelink 498 )
c588255… ragelink 499 assert "error" in result
c588255… ragelink 500
c588255… ragelink 501 @patch(_CLI)
c588255… ragelink 502 def test_update_ticket_status(self, mock_cli_cls, sample_project, fossil_repo_obj):
c588255… ragelink 503 cli = MagicMock()
c588255… ragelink 504 cli.ticket_change.return_value = True
c588255… ragelink 505 mock_cli_cls.return_value = cli
c588255… ragelink 506
c588255… ragelink 507 result = execute_tool(
c588255… ragelink 508 "update_ticket",
c588255… ragelink 509 {
c588255… ragelink 510 "slug": sample_project.slug,
c588255… ragelink 511 "uuid": "tkt-001",
c588255… ragelink 512 "status": "Closed",
c588255… ragelink 513 },
c588255… ragelink 514 )
c588255… ragelink 515 assert result["success"] is True
c588255… ragelink 516 call_args = cli.ticket_change.call_args
c588255… ragelink 517 assert call_args[0][1] == "tkt-001"
c588255… ragelink 518 assert call_args[0][2]["status"] == "Closed"
c588255… ragelink 519
c588255… ragelink 520 @patch(_CLI)
c588255… ragelink 521 def test_update_ticket_comment(self, mock_cli_cls, sample_project, fossil_repo_obj):
c588255… ragelink 522 cli = MagicMock()
c588255… ragelink 523 cli.ticket_change.return_value = True
c588255… ragelink 524 mock_cli_cls.return_value = cli
c588255… ragelink 525
c588255… ragelink 526 result = execute_tool(
c588255… ragelink 527 "update_ticket",
c588255… ragelink 528 {
c588255… ragelink 529 "slug": sample_project.slug,
c588255… ragelink 530 "uuid": "tkt-001",
c588255… ragelink 531 "comment": "Fixed in latest push",
c588255… ragelink 532 },
c588255… ragelink 533 )
c588255… ragelink 534 assert result["success"] is True
c588255… ragelink 535 call_args = cli.ticket_change.call_args
c588255… ragelink 536 assert call_args[0][2]["icomment"] == "Fixed in latest push"
c588255… ragelink 537
c588255… ragelink 538 @patch(_CLI)
c588255… ragelink 539 def test_update_ticket_no_fields(self, mock_cli_cls, sample_project, fossil_repo_obj):
c588255… ragelink 540 cli = MagicMock()
c588255… ragelink 541 mock_cli_cls.return_value = cli
c588255… ragelink 542
c588255… ragelink 543 result = execute_tool(
c588255… ragelink 544 "update_ticket",
c588255… ragelink 545 {
c588255… ragelink 546 "slug": sample_project.slug,
c588255… ragelink 547 "uuid": "tkt-001",
c588255… ragelink 548 },
c588255… ragelink 549 )
c588255… ragelink 550 assert "error" in result
c588255… ragelink 551 assert "No fields" in result["error"]
c588255… ragelink 552
c588255… ragelink 553
c588255… ragelink 554 # ---------------------------------------------------------------------------
c588255… ragelink 555 # wiki handlers
c588255… ragelink 556 # ---------------------------------------------------------------------------
c588255… ragelink 557
c588255… ragelink 558
c588255… ragelink 559 @pytest.mark.django_db
c588255… ragelink 560 class TestWiki:
c588255… ragelink 561 @patch(_READER)
c588255… ragelink 562 def test_list_wiki_pages(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 563 reader = _mock_reader()
c588255… ragelink 564 reader.get_wiki_pages.return_value = [
c588255… ragelink 565 WikiPage(name="Home", content="", last_modified=datetime(2025, 1, 12, tzinfo=UTC), user="alice"),
c588255… ragelink 566 WikiPage(name="FAQ", content="", last_modified=datetime(2025, 1, 13, tzinfo=UTC), user="bob"),
c588255… ragelink 567 ]
c588255… ragelink 568 mock_reader_cls.return_value = reader
c588255… ragelink 569
c588255… ragelink 570 result = execute_tool("list_wiki_pages", {"slug": sample_project.slug})
c588255… ragelink 571 assert len(result["pages"]) == 2
c588255… ragelink 572 assert result["pages"][0]["name"] == "Home"
c588255… ragelink 573
c588255… ragelink 574 @patch(_READER)
c588255… ragelink 575 def test_get_wiki_page(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 576 reader = _mock_reader()
c588255… ragelink 577 reader.get_wiki_page.return_value = WikiPage(
c588255… ragelink 578 name="Home",
c588255… ragelink 579 content="# Welcome\nThis is home.",
c588255… ragelink 580 last_modified=datetime(2025, 1, 12, tzinfo=UTC),
c588255… ragelink 581 user="alice",
c588255… ragelink 582 )
c588255… ragelink 583 mock_reader_cls.return_value = reader
c588255… ragelink 584
c588255… ragelink 585 result = execute_tool("get_wiki_page", {"slug": sample_project.slug, "page_name": "Home"})
c588255… ragelink 586 assert result["name"] == "Home"
c588255… ragelink 587 assert "Welcome" in result["content"]
c588255… ragelink 588
c588255… ragelink 589 @patch(_READER)
c588255… ragelink 590 def test_wiki_page_not_found(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 591 reader = _mock_reader()
c588255… ragelink 592 reader.get_wiki_page.return_value = None
c588255… ragelink 593 mock_reader_cls.return_value = reader
c588255… ragelink 594
c588255… ragelink 595 result = execute_tool("get_wiki_page", {"slug": sample_project.slug, "page_name": "Missing"})
c588255… ragelink 596 assert "error" in result
c588255… ragelink 597
c588255… ragelink 598
c588255… ragelink 599 # ---------------------------------------------------------------------------
c588255… ragelink 600 # branches, blame, file history
c588255… ragelink 601 # ---------------------------------------------------------------------------
c588255… ragelink 602
c588255… ragelink 603
c588255… ragelink 604 @pytest.mark.django_db
c588255… ragelink 605 class TestBranchesAndHistory:
c588255… ragelink 606 @patch(_READER)
c588255… ragelink 607 def test_list_branches(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 608 reader = _mock_reader()
c588255… ragelink 609 reader.get_branches.return_value = [
c588255… ragelink 610 {
c588255… ragelink 611 "name": "trunk",
c588255… ragelink 612 "last_checkin": datetime(2025, 1, 15, tzinfo=UTC),
c588255… ragelink 613 "last_user": "alice",
c588255… ragelink 614 "checkin_count": 30,
c588255… ragelink 615 "last_uuid": "abc123",
c588255… ragelink 616 },
c588255… ragelink 617 ]
c588255… ragelink 618 mock_reader_cls.return_value = reader
c588255… ragelink 619
c588255… ragelink 620 result = execute_tool("list_branches", {"slug": sample_project.slug})
c588255… ragelink 621 assert len(result["branches"]) == 1
c588255… ragelink 622 assert result["branches"][0]["name"] == "trunk"
c588255… ragelink 623
c588255… ragelink 624 @patch(_CLI)
c588255… ragelink 625 def test_get_file_blame(self, mock_cli_cls, sample_project, fossil_repo_obj):
c588255… ragelink 626 cli = MagicMock()
c588255… ragelink 627 cli.blame.return_value = [
c588255… ragelink 628 {"uuid": "aaa", "date": "2025-01-15", "user": "alice", "text": "line 1"},
c588255… ragelink 629 {"uuid": "bbb", "date": "2025-01-14", "user": "bob", "text": "line 2"},
c588255… ragelink 630 ]
c588255… ragelink 631 mock_cli_cls.return_value = cli
c588255… ragelink 632
c588255… ragelink 633 result = execute_tool("get_file_blame", {"slug": sample_project.slug, "filepath": "main.py"})
c588255… ragelink 634 assert result["filepath"] == "main.py"
c588255… ragelink 635 assert len(result["lines"]) == 2
c588255… ragelink 636 assert result["total"] == 2
c588255… ragelink 637
c588255… ragelink 638 @patch(_READER)
c588255… ragelink 639 def test_get_file_history(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 640 reader = _mock_reader()
c588255… ragelink 641 reader.get_file_history.return_value = [
c588255… ragelink 642 {"uuid": "c1", "timestamp": datetime(2025, 1, 15, tzinfo=UTC), "user": "alice", "comment": "Update"},
c588255… ragelink 643 {"uuid": "c2", "timestamp": datetime(2025, 1, 14, tzinfo=UTC), "user": "bob", "comment": "Create"},
c588255… ragelink 644 ]
c588255… ragelink 645 mock_reader_cls.return_value = reader
c588255… ragelink 646
c588255… ragelink 647 result = execute_tool("get_file_history", {"slug": sample_project.slug, "filepath": "main.py"})
c588255… ragelink 648 assert result["filepath"] == "main.py"
c588255… ragelink 649 assert len(result["history"]) == 2
c588255… ragelink 650 # Timestamps should be serialized
c588255… ragelink 651 assert isinstance(result["history"][0]["timestamp"], str)
c588255… ragelink 652
c588255… ragelink 653
c588255… ragelink 654 # ---------------------------------------------------------------------------
c588255… ragelink 655 # sql_query
c588255… ragelink 656 # ---------------------------------------------------------------------------
c588255… ragelink 657
c588255… ragelink 658
c588255… ragelink 659 @pytest.mark.django_db
c588255… ragelink 660 class TestSqlQuery:
c588255… ragelink 661 def test_rejects_non_select(self, sample_project, fossil_repo_obj):
c588255… ragelink 662 result = execute_tool("sql_query", {"slug": sample_project.slug, "sql": "DELETE FROM ticket"})
c588255… ragelink 663 assert "error" in result
c588255… ragelink 664 assert "SELECT" in result["error"]
c588255… ragelink 665
c588255… ragelink 666 def test_rejects_empty_query(self, sample_project, fossil_repo_obj):
c588255… ragelink 667 result = execute_tool("sql_query", {"slug": sample_project.slug, "sql": ""})
c588255… ragelink 668 assert "error" in result
c588255… ragelink 669
c588255… ragelink 670 def test_rejects_drop(self, sample_project, fossil_repo_obj):
c588255… ragelink 671 result = execute_tool("sql_query", {"slug": sample_project.slug, "sql": "SELECT 1; DROP TABLE ticket"})
c588255… ragelink 672 assert "error" in result
c588255… ragelink 673
c588255… ragelink 674 @patch(_READER)
c588255… ragelink 675 def test_valid_select(self, mock_reader_cls, sample_project, fossil_repo_obj):
c588255… ragelink 676 reader = _mock_reader()
c588255… ragelink 677 mock_cursor = MagicMock()
c588255… ragelink 678 mock_cursor.description = [("tkt_uuid",), ("title",)]
c588255… ragelink 679 mock_cursor.fetchmany.return_value = [("uuid-1", "Bug one"), ("uuid-2", "Bug two")]
c588255… ragelink 680 reader.conn.cursor.return_value = mock_cursor
c588255… ragelink 681 mock_reader_cls.return_value = reader
c588255… ragelink 682
c588255… ragelink 683 result = execute_tool("sql_query", {"slug": sample_project.slug, "sql": "SELECT tkt_uuid, title FROM ticket"})
c588255… ragelink 684 assert result["columns"] == ["tkt_uuid", "title"]
c588255… ragelink 685 assert len(result["rows"]) == 2
c588255… ragelink 686 assert result["count"] == 2
c588255… ragelink 687
c588255… ragelink 688
c588255… ragelink 689 # ---------------------------------------------------------------------------
c588255… ragelink 690 # Server module smoke test
c588255… ragelink 691 # ---------------------------------------------------------------------------
c588255… ragelink 692
c588255… ragelink 693
c588255… ragelink 694 class TestServerModule:
c588255… ragelink 695 def test_server_instance_exists(self):
c588255… ragelink 696 from mcp_server.server import server
c588255… ragelink 697
c588255… ragelink 698 assert server.name == "fossilrepo"
c588255… ragelink 699
c588255… ragelink 700 def test_main_is_coroutine(self):
c588255… ragelink 701 import inspect
c588255… ragelink 702
c588255… ragelink 703 from mcp_server.server import main
c588255… ragelink 704
c588255… ragelink 705 assert inspect.iscoroutinefunction(main)
c588255… ragelink 706
c588255… ragelink 707 def test_entry_point_function_exists(self):
c588255… ragelink 708 from mcp_server.__main__ import run
c588255… ragelink 709
c588255… ragelink 710 assert callable(run)

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button