"""Tests for tree forking and lineage tracking.""" import pytest from httpx import AsyncClient pytestmark = pytest.mark.asyncio class TestTreeForking: """Test tree fork creation, lineage, and update detection.""" async def test_fork_tree(self, client: AsyncClient, auth_headers, test_tree): """Fork a tree and verify fork metadata.""" response = await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "Customizing for our network"}, headers=auth_headers ) assert response.status_code == 201 fork = response.json() assert fork["name"] == f"Fork of {test_tree['name']}" assert fork["tree_structure"] == test_tree["tree_structure"] assert fork["is_public"] is False assert fork["version"] == 1 # Verify fork_info assert fork["fork_info"] is not None assert fork["fork_info"]["parent_tree_id"] == test_tree["id"] assert fork["fork_info"]["root_tree_id"] == test_tree["id"] assert fork["fork_info"]["fork_reason"] == "Customizing for our network" assert fork["fork_info"]["fork_depth"] == 1 assert fork["fork_info"]["has_parent_updates"] is False async def test_fork_with_custom_name(self, client: AsyncClient, auth_headers, test_tree): """Fork a tree with a custom name.""" response = await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"name": "My Custom Fork", "fork_reason": "Testing"}, headers=auth_headers ) assert response.status_code == 201 assert response.json()["name"] == "My Custom Fork" async def test_fork_of_fork_lineage(self, client: AsyncClient, auth_headers, test_tree): """Fork a fork and verify lineage tracking (root_tree_id, fork_depth).""" # Create first fork resp1 = await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "Junior's customization"}, headers=auth_headers ) assert resp1.status_code == 201 fork1 = resp1.json() # Fork the fork resp2 = await client.post( f"/api/v1/trees/{fork1['id']}/fork", json={"fork_reason": "Senior's refinement"}, headers=auth_headers ) assert resp2.status_code == 201 fork2 = resp2.json() # Verify lineage assert fork2["fork_info"]["parent_tree_id"] == fork1["id"] assert fork2["fork_info"]["root_tree_id"] == test_tree["id"] # Points to original assert fork2["fork_info"]["fork_depth"] == 2 assert fork2["fork_info"]["fork_reason"] == "Senior's refinement" async def test_fork_nonexistent_tree(self, client: AsyncClient, auth_headers): """Fork a nonexistent tree returns 404.""" response = await client.post( "/api/v1/trees/00000000-0000-0000-0000-000000000000/fork", json={"fork_reason": "test"}, headers=auth_headers ) assert response.status_code == 404 async def test_list_forks(self, client: AsyncClient, auth_headers, test_tree): """List forks of a tree.""" # Create two forks await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "Fork 1"}, headers=auth_headers ) await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "Fork 2"}, headers=auth_headers ) response = await client.get( f"/api/v1/trees/{test_tree['id']}/forks", headers=auth_headers ) assert response.status_code == 200 forks = response.json() assert len(forks) == 2 async def test_lineage_chain(self, client: AsyncClient, auth_headers, test_tree): """Get lineage from fork back to root.""" # Create chain: root → fork1 → fork2 resp1 = await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "Level 1"}, headers=auth_headers ) fork1 = resp1.json() resp2 = await client.post( f"/api/v1/trees/{fork1['id']}/fork", json={"fork_reason": "Level 2"}, headers=auth_headers ) fork2 = resp2.json() # Get lineage from fork2 response = await client.get( f"/api/v1/trees/{fork2['id']}/lineage", headers=auth_headers ) assert response.status_code == 200 lineage = response.json() # Should be [fork2, fork1, root] assert len(lineage) == 3 assert lineage[0]["id"] == fork2["id"] assert lineage[1]["id"] == fork1["id"] assert lineage[2]["id"] == test_tree["id"] async def test_lineage_of_root_tree(self, client: AsyncClient, auth_headers, test_tree): """Root tree lineage is just itself.""" response = await client.get( f"/api/v1/trees/{test_tree['id']}/lineage", headers=auth_headers ) assert response.status_code == 200 lineage = response.json() assert len(lineage) == 1 assert lineage[0]["id"] == test_tree["id"] async def test_fork_preserves_tree_structure(self, client: AsyncClient, auth_headers, test_tree): """Fork copies the complete tree_structure JSONB.""" response = await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "Copy check"}, headers=auth_headers ) fork = response.json() # Get full fork detail detail = await client.get( f"/api/v1/trees/{fork['id']}", headers=auth_headers ) assert detail.status_code == 200 assert detail.json()["tree_structure"] == test_tree["tree_structure"] async def test_fork_requires_auth(self, client: AsyncClient, test_tree): """Fork without auth returns 401.""" response = await client.post( f"/api/v1/trees/{test_tree['id']}/fork", json={"fork_reason": "No auth"} ) assert response.status_code == 401