liumangmang
148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
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
|