test_branching.py (6166B)
1 """T074 — Tests for branching endpoints (fork + list branches).""" 2 3 from __future__ import annotations 4 from unittest.mock import AsyncMock, MagicMock, patch 5 import pytest 6 from fastapi import FastAPI 7 from fastapi.testclient import TestClient 8 9 10 # ── helpers ────────────────────────────────────────────────────────────── 11 12 def _make_hermes_fork_response(session_id: str = "fork-1", parent_id: str = "s1") -> dict: 13 """Build a mock fork response in Hermes API format.""" 14 return { 15 "object": "hermes.session", 16 "session": { 17 "id": session_id, 18 "parent_session_id": parent_id, 19 "title": f"Forked from {parent_id}", 20 "source": "api_server", 21 }, 22 } 23 24 25 def _mock_client(**overrides: AsyncMock) -> MagicMock: 26 """Build a mock HermesClient with fork methods.""" 27 client = MagicMock() 28 client.fork_session = AsyncMock(return_value=_make_hermes_fork_response()) 29 for name, mock in overrides.items(): 30 setattr(client, name, mock) 31 return client 32 33 def _mock_db(rows: list[tuple] | None = None) -> MagicMock: 34 """Build a mock DB that returns given rows for branch queries.""" 35 db = MagicMock() 36 db.fetch_all = AsyncMock(return_value=rows or []) 37 db.execute = AsyncMock() 38 return db 39 40 def _test_app(db: MagicMock | None = None) -> FastAPI: 41 """Build a minimal app with sessions router, no auth middleware.""" 42 from app.routes.sessions import router as sessions_router 43 app = FastAPI() 44 app.include_router(sessions_router) 45 if db is not None: 46 app.state.db = db 47 return app 48 49 50 # ── tests ──────────────────────────────────────────────────────────────── 51 52 53 class TestForkSession: 54 """POST /api/sessions/{id}/fork creates a forked session.""" 55 56 def test_forks_session(self) -> None: 57 """Fork proxy returns the normalized session from Hermes.""" 58 mock_client = _mock_client() 59 app = _test_app() 60 with patch( 61 "app.routes.sessions.get_hermes_client", return_value=mock_client, 62 ): 63 with TestClient(app) as tc: 64 resp = tc.post("/api/sessions/s1/fork") 65 assert resp.status_code == 200 66 data = resp.json() 67 assert data["id"] == "fork-1" 68 assert data["name"] == "Forked from s1" 69 assert data["parent_session_id"] == "s1" 70 mock_client.fork_session.assert_awaited_once_with("s1") 71 72 def test_fork_stores_branch_in_db(self) -> None: 73 """Fork inserts a row into the branches table.""" 74 db = _mock_db() 75 mock_client = _mock_client() 76 app = _test_app(db=db) 77 with patch( 78 "app.routes.sessions.get_hermes_client", return_value=mock_client, 79 ): 80 with TestClient(app) as tc: 81 resp = tc.post("/api/sessions/s1/fork") 82 assert resp.status_code == 200 83 data = resp.json() 84 assert data["id"] == "fork-1" 85 db.execute.assert_awaited_once() 86 call_args = db.execute.call_args 87 sql = call_args[0][0] 88 assert "branches" in sql 89 assert "INSERT" in sql 90 params = call_args[0][1] 91 assert params[0] == "fork-1" 92 assert params[1] == "s1" 93 94 def test_fork_hermes_error_returns_502(self) -> None: 95 """Hermes failure during fork returns 502.""" 96 mock_client = _mock_client( 97 fork_session=AsyncMock(side_effect=Exception("Hermes down")), 98 ) 99 app = _test_app() 100 with patch( 101 "app.routes.sessions.get_hermes_client", return_value=mock_client, 102 ): 103 with TestClient(app) as tc: 104 resp = tc.post("/api/sessions/s1/fork") 105 assert resp.status_code == 502 106 body = resp.json() 107 assert "detail" in body 108 assert "Hermes" in body["detail"] 109 assert "Hermes down" in body["detail"] 110 111 112 class TestListBranches: 113 """GET /api/sessions/{id}/branches returns child sessions.""" 114 115 def test_returns_branches(self) -> None: 116 """Branches query returns children from the DB.""" 117 db = _mock_db(rows=[ 118 ("fork-1", "s1", 3), 119 ("fork-2", "s1", 5), 120 ]) 121 mock_client = _mock_client() 122 app = _test_app(db=db) 123 with patch( 124 "app.routes.sessions.get_hermes_client", return_value=mock_client, 125 ): 126 with TestClient(app) as tc: 127 resp = tc.get("/api/sessions/s1/branches") 128 assert resp.status_code == 200 129 body = resp.json() 130 assert "branches" in body 131 branches = body["branches"] 132 assert len(branches) == 2 133 assert branches[0]["session_id"] == "fork-1" 134 assert branches[0]["parent_session_id"] == "s1" 135 assert branches[0]["fork_message_index"] == 3 136 assert branches[1]["session_id"] == "fork-2" 137 assert branches[1]["parent_session_id"] == "s1" 138 assert branches[1]["fork_message_index"] == 5 139 140 def test_returns_empty_when_no_branches(self) -> None: 141 """Empty DB returns empty branches list.""" 142 db = _mock_db(rows=[]) 143 mock_client = _mock_client() 144 app = _test_app(db=db) 145 with patch( 146 "app.routes.sessions.get_hermes_client", return_value=mock_client, 147 ): 148 with TestClient(app) as tc: 149 resp = tc.get("/api/sessions/s1/branches") 150 assert resp.status_code == 200 151 assert resp.json() == {"branches": []} 152 153 def test_no_db_returns_empty_branches(self) -> None: 154 """Without a DB, branches returns empty list.""" 155 mock_client = _mock_client() 156 app = _test_app() 157 with patch( 158 "app.routes.sessions.get_hermes_client", return_value=mock_client, 159 ): 160 with TestClient(app) as tc: 161 resp = tc.get("/api/sessions/s1/branches") 162 assert resp.status_code == 200 163 assert resp.json() == {"branches": []}