import asyncio import json import pytest from unittest.mock import AsyncMock, MagicMock, patch from app.services.browser_session_service import BrowserSessionService, BrowserSession, BrowserTab, BrowserSessionError @pytest.fixture def service(): return BrowserSessionService() @pytest.fixture def session(service): fake_context = AsyncMock() fake_page = AsyncMock() # Playwright's page.on is synchronous fake_page.on = MagicMock() fake_page.is_closed = MagicMock(return_value=False) fake_page.url = "https://initial.test" fake_page.title = AsyncMock(return_value="Initial Tab") tab_id = "tab1" tab = BrowserTab(id=tab_id, page=fake_page, created_at=100.0) sess = BrowserSession( id="sess123", custom_page_id=1, profile_key="profile1", context=fake_context, tabs={tab_id: tab}, active_tab_id=tab_id, lock=asyncio.Lock(), ) service._sessions[sess.id] = sess return sess @pytest.mark.asyncio async def test_tab_capture(service, session): new_page = AsyncMock() new_page.on = MagicMock() new_page.is_closed = MagicMock(return_value=False) new_page.url = "https://new.test" new_page.title = AsyncMock(return_value="New Tab") new_page.bring_to_front = AsyncMock() # Simulate page capture service._handle_new_page(session, new_page) # Wait for the background registration task to finish await asyncio.sleep(0.1) assert len(session.tabs) == 2 assert session.active_tab_id != "tab1" new_tab_id = session.active_tab_id assert session.tabs[new_tab_id].page == new_page assert session.tab_revision == 1 # Wait a bit for the background task _init_new_tab to finish if possible, # though it's mocked anyway. await asyncio.sleep(0.1) @pytest.mark.asyncio async def test_activate_tab(service, session): new_page = AsyncMock() new_page.is_closed = MagicMock(return_value=False) new_page.bring_to_front = AsyncMock() tab2_id = "tab2" session.tabs[tab2_id] = BrowserTab(id=tab2_id, page=new_page, created_at=200.0) await service.activate_tab(session.id, tab2_id) assert session.active_tab_id == tab2_id assert new_page.bring_to_front.call_count == 1 assert session.tab_revision == 1 @pytest.mark.asyncio async def test_close_tab_safety(service, session): # Cannot close last tab with pytest.raises(ValueError, match="cannot close the last tab"): await service.close_tab(session.id, "tab1") @pytest.mark.asyncio async def test_close_active_tab_fallback(service, session): # Setup tab2 page2 = AsyncMock() page2.is_closed = MagicMock(return_value=False) page2.bring_to_front = AsyncMock() tab2_id = "tab2" session.tabs[tab2_id] = BrowserTab(id=tab2_id, page=page2, created_at=200.0) # Active is tab2 session.active_tab_id = tab2_id # Close active tab2 await service.close_tab(session.id, tab2_id) assert len(session.tabs) == 1 assert session.active_tab_id == "tab1" assert tab2_id not in session.tabs assert session.tabs["tab1"].page.bring_to_front.call_count == 1 @pytest.mark.asyncio async def test_session_state_includes_tabs(service, session): # Setup tab2 page2 = AsyncMock() page2.is_closed = MagicMock(return_value=False) page2.url = "https://tab2.test" page2.title = AsyncMock(return_value="Tab 2") tab2_id = "tab2" session.tabs[tab2_id] = BrowserTab(id=tab2_id, page=page2, created_at=200.0) state = await service.state(session.id) assert state["id"] == session.id assert state["active_tab_id"] == "tab1" assert len(state["tabs"]) == 2 # Tabs should be sorted by created_at assert state["tabs"][0]["id"] == "tab1" assert state["tabs"][1]["id"] == "tab2" assert state["tabs"][0]["title"] == "Initial Tab" assert state["tabs"][1]["url"] == "https://tab2.test" @pytest.mark.asyncio async def test_ensure_open_prunes_closed_tab(service, session): # Setup tab2 and make it active page2 = AsyncMock() page2.is_closed = MagicMock(return_value=True) # Page 2 is closed tab2_id = "tab2" session.tabs[tab2_id] = BrowserTab(id=tab2_id, page=page2, created_at=200.0) session.active_tab_id = tab2_id # Calling any interaction method should trigger _ensure_open # and fallback to tab1 await service.screenshot(session.id) assert session.active_tab_id == "tab1" assert tab2_id not in session.tabs assert session.tab_revision == 1 @pytest.mark.asyncio async def test_ensure_open_discards_session_if_all_tabs_closed(service, session): session.tabs["tab1"].page.is_closed = MagicMock(return_value=True) with pytest.raises(BrowserSessionError, match="all browser pages are closed"): await service.screenshot(session.id) assert session.id not in service._sessions @pytest.mark.asyncio async def test_tab_revision_bumps_on_events(service, session): # Setup listeners for the initial tab service._setup_tab_listeners(session, session.tabs["tab1"].page) # Extract the "load" listener callback calls = session.tabs["tab1"].page.on.call_args_list load_callback = next(c[0][1] for c in calls if c[0][0] == "load") initial_revision = session.tab_revision load_callback() assert session.tab_revision == initial_revision + 1 @pytest.mark.asyncio async def test_session_state_concurrency_with_popup(service, session): # Setup: page.title() will trigger a new page registration in background triggered = False async def mock_title(): nonlocal triggered # Only trigger popup once (state() calls title twice: once for tabs list, once for top-level) if not triggered: triggered = True # Simulate a popup arriving while title is being fetched new_page = AsyncMock() new_page.on = MagicMock() new_page.is_closed = MagicMock(return_value=False) new_page.url = "https://new.test" new_page.bring_to_front = AsyncMock() service._handle_new_page(session, new_page) # Yield to let the registration task start (it will block on the lock) await asyncio.sleep(0.01) return "Initial Tab" session.tabs["tab1"].page.title = mock_title # Call state() which takes the lock and calls _session_state (which calls mock_title) state = await service.state(session.id) assert len(state["tabs"]) == 1 # Still 1 because popup registration is waiting for lock # Yield to let registration task finish after state() released the lock await asyncio.sleep(0.1) assert len(session.tabs) == 2 assert session.tab_revision > 0