Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions src/note_mcp/auth/browser.py
Original file line number Diff line number Diff line change
Expand Up @@ -349,9 +349,9 @@ async def login_with_browser(
logger.info("Closing existing browser...")
await manager.close()

# Get fresh page
logger.info("Getting fresh page...")
page = await manager.get_page()
# Get fresh page in headful mode (user needs to see browser for manual login)
logger.info("Getting fresh page in headful mode...")
page = await manager.get_page(headless=False)

# Check for saved session and inject cookies if available
session_manager = SessionManager()
Expand Down Expand Up @@ -540,7 +540,7 @@ def is_logged_in(url: str) -> bool:
user_id = username
logger.info(f"Extracted username from URL: {username}")
except Exception as e:
logger.debug(f"Profile navigation method failed: {e}")
logger.warning(f"Profile navigation method failed: {e}")

# Method 3: Fallback to API if browser method failed
if not username:
Expand Down
51 changes: 36 additions & 15 deletions src/note_mcp/browser/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,13 @@

import asyncio
import atexit
import logging
from typing import TYPE_CHECKING, ClassVar

from playwright.async_api import Page, async_playwright

logger = logging.getLogger(__name__)

if TYPE_CHECKING:
from playwright.async_api import Browser, BrowserContext, Playwright

Expand Down Expand Up @@ -69,9 +72,13 @@ def _sync_cleanup(cls) -> None:
else:
# Run cleanup directly if loop is not running
loop.run_until_complete(cls._async_cleanup())
except RuntimeError:
# No event loop, create one for cleanup
asyncio.run(cls._async_cleanup())
except RuntimeError as e:
# Only handle "no event loop" errors, re-raise others
error_msg = str(e).lower()
if "no running event loop" in error_msg or "no current event loop" in error_msg:
asyncio.run(cls._async_cleanup())
else:
raise

@classmethod
async def _async_cleanup(cls) -> None:
Expand All @@ -87,33 +94,39 @@ async def _async_cleanup(cls) -> None:

@classmethod
async def _safe_close_browser(cls) -> None:
"""Safely close browser, ignoring errors."""
"""Safely close browser, logging errors at debug level."""
try:
if cls._browser is not None:
await cls._browser.close()
except Exception: # noqa: S110 - intentionally broad for cleanup
pass
except Exception as e: # noqa: BLE001 - intentionally broad for cleanup
logger.debug("Browser cleanup error (non-fatal): %s", e, exc_info=True)

@classmethod
async def _safe_stop_playwright(cls) -> None:
"""Safely stop playwright, ignoring errors."""
"""Safely stop playwright, logging errors at debug level."""
try:
if cls._playwright is not None:
await cls._playwright.stop()
except Exception: # noqa: S110 - intentionally broad for cleanup
pass
except Exception as e: # noqa: BLE001 - intentionally broad for cleanup
logger.debug("Playwright cleanup error (non-fatal): %s", e, exc_info=True)

async def _ensure_browser(self, headless: bool | None = None) -> None:
"""Ensure browser is started.

async def _ensure_browser(self) -> None:
"""Ensure browser is started."""
Args:
headless: If True, run in headless mode. If False, show browser window.
If None, use the default from config (NOTE_MCP_TEST_HEADLESS env var).
"""
if self._playwright is None:
self.__class__._playwright = await async_playwright().start()
playwright = self._playwright
assert playwright is not None

if self._browser is None:
from note_mcp.browser.config import get_headless_mode
if headless is None:
from note_mcp.browser.config import get_headless_mode

headless = get_headless_mode()
headless = get_headless_mode()
self.__class__._browser = await playwright.chromium.launch(headless=headless)
browser = self._browser
assert browser is not None
Expand All @@ -126,12 +139,20 @@ async def _ensure_browser(self) -> None:
if self._page is None or self._page.is_closed():
self.__class__._page = await context.new_page()

async def get_page(self) -> Page:
async def get_page(self, headless: bool | None = None) -> Page:
"""Get a browser page, creating if necessary.

Reuses existing page if available and not closed.
Uses lock for thread-safe access.

Args:
headless: If True, run in headless mode. If False, show browser window.
If None, use the default from config (NOTE_MCP_TEST_HEADLESS env var).

IMPORTANT: This parameter is only used when launching a NEW browser.
If a browser already exists, this parameter is SILENTLY IGNORED.
Call close() first if you need to change the headless mode.

Returns:
Playwright Page instance
"""
Expand All @@ -146,7 +167,7 @@ async def get_page(self) -> Page:
return self._page

# Create new browser/page if needed
await self._ensure_browser()
await self._ensure_browser(headless=headless)
assert self._page is not None
return self._page

Expand Down
85 changes: 85 additions & 0 deletions tests/integration/test_auth_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,91 @@
class TestLoginFlow:
"""Tests for browser-based login flow."""

@pytest.mark.asyncio
async def test_login_uses_headful_mode(self) -> None:
"""Test that login_with_browser always uses headful mode (headless=False).

This is critical because manual login requires the user to see and interact
with the browser window. Running in headless mode would block indefinitely
waiting for user input that can never happen.
"""
with patch("note_mcp.auth.browser.BrowserManager") as mock_manager_class:
mock_manager = MagicMock()
mock_manager_class.get_instance.return_value = mock_manager

mock_page = AsyncMock()
mock_page.goto = AsyncMock()
mock_page.wait_for_url = AsyncMock()
mock_page.url = "https://note.com/" # Already logged in
mock_page.context = AsyncMock()
mock_page.context.cookies = AsyncMock(
return_value=[
{"name": "_note_session_v5", "value": "session456"},
]
)
mock_page.context.add_cookies = AsyncMock()
mock_page.evaluate = AsyncMock(return_value={"id": "", "urlname": "testuser"})

mock_manager.close = AsyncMock()
mock_manager.get_page = AsyncMock(return_value=mock_page)

with patch("note_mcp.auth.browser.SessionManager") as mock_session_manager_class:
mock_session_manager = MagicMock()
mock_session_manager.load.return_value = None # No saved session
mock_session_manager_class.return_value = mock_session_manager

await login_with_browser(timeout=60)

# Verify get_page was called with headless=False
mock_manager.get_page.assert_called_once_with(headless=False)

@pytest.mark.asyncio
async def test_login_calls_close_before_get_page(self) -> None:
"""Test that login_with_browser calls close() before get_page().

This ensures the headless parameter is respected by forcing a fresh
browser launch, since get_page() ignores the parameter when a browser
already exists.
"""
call_order: list[str] = []

with patch("note_mcp.auth.browser.BrowserManager") as mock_manager_class:
mock_manager = MagicMock()
mock_manager_class.get_instance.return_value = mock_manager

mock_page = AsyncMock()
mock_page.goto = AsyncMock()
mock_page.wait_for_url = AsyncMock()
mock_page.url = "https://note.com/" # Already logged in
mock_page.context = AsyncMock()
mock_page.context.cookies = AsyncMock(
return_value=[
{"name": "_note_session_v5", "value": "session456"},
]
)
mock_page.context.add_cookies = AsyncMock()
mock_page.evaluate = AsyncMock(return_value={"id": "", "urlname": "testuser"})

async def track_close() -> None:
call_order.append("close")

async def track_get_page(headless: bool | None = None) -> AsyncMock:
call_order.append("get_page")
return mock_page

mock_manager.close = AsyncMock(side_effect=track_close)
mock_manager.get_page = AsyncMock(side_effect=track_get_page)

with patch("note_mcp.auth.browser.SessionManager") as mock_session_manager_class:
mock_session_manager = MagicMock()
mock_session_manager.load.return_value = None
mock_session_manager_class.return_value = mock_session_manager

await login_with_browser(timeout=60)

# Verify close() was called before get_page()
assert call_order == ["close", "get_page"]

@pytest.mark.asyncio
async def test_login_flow_extracts_cookies(self) -> None:
"""Test that login flow extracts cookies from browser."""
Expand Down
115 changes: 115 additions & 0 deletions tests/unit/test_browser_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,121 @@ async def test_close_handles_no_browser(self) -> None:
await manager.close()


class TestBrowserManagerHeadlessParameter:
"""Tests for BrowserManager headless parameter support."""

@pytest.fixture(autouse=True)
def reset_singleton(self) -> None:
"""Reset singleton instance before each test."""
BrowserManager._instance = None
BrowserManager._browser = None
BrowserManager._context = None
BrowserManager._page = None
BrowserManager._lock = None
BrowserManager._playwright = None

@pytest.mark.asyncio
async def test_get_page_accepts_headless_parameter(self) -> None:
"""Test that get_page accepts headless parameter."""
manager = BrowserManager.get_instance()

with patch("note_mcp.browser.manager.async_playwright") as mock_playwright:
mock_browser = AsyncMock()
mock_context = AsyncMock()
mock_page = AsyncMock()
mock_page.is_closed = MagicMock(return_value=False)

mock_pw = AsyncMock()
mock_pw.chromium.launch = AsyncMock(return_value=mock_browser)
mock_browser.new_context = AsyncMock(return_value=mock_context)
mock_context.new_page = AsyncMock(return_value=mock_page)

mock_playwright.return_value.start = AsyncMock(return_value=mock_pw)

page = await manager.get_page(headless=False)

assert page is mock_page
# Verify chromium.launch was called with headless=False
mock_pw.chromium.launch.assert_called_once_with(headless=False)

@pytest.mark.asyncio
async def test_get_page_headless_true_uses_headless_mode(self) -> None:
"""Test that get_page with headless=True launches in headless mode."""
manager = BrowserManager.get_instance()

with patch("note_mcp.browser.manager.async_playwright") as mock_playwright:
mock_browser = AsyncMock()
mock_context = AsyncMock()
mock_page = AsyncMock()
mock_page.is_closed = MagicMock(return_value=False)

mock_pw = AsyncMock()
mock_pw.chromium.launch = AsyncMock(return_value=mock_browser)
mock_browser.new_context = AsyncMock(return_value=mock_context)
mock_context.new_page = AsyncMock(return_value=mock_page)

mock_playwright.return_value.start = AsyncMock(return_value=mock_pw)

page = await manager.get_page(headless=True)

assert page is mock_page
mock_pw.chromium.launch.assert_called_once_with(headless=True)

@pytest.mark.asyncio
async def test_get_page_default_uses_config(self) -> None:
"""Test that get_page without headless parameter uses config default."""
manager = BrowserManager.get_instance()

with (
patch("note_mcp.browser.manager.async_playwright") as mock_playwright,
patch("note_mcp.browser.config.get_headless_mode", return_value=True) as mock_config,
):
mock_browser = AsyncMock()
mock_context = AsyncMock()
mock_page = AsyncMock()
mock_page.is_closed = MagicMock(return_value=False)

mock_pw = AsyncMock()
mock_pw.chromium.launch = AsyncMock(return_value=mock_browser)
mock_browser.new_context = AsyncMock(return_value=mock_context)
mock_context.new_page = AsyncMock(return_value=mock_page)

mock_playwright.return_value.start = AsyncMock(return_value=mock_pw)

page = await manager.get_page()

assert page is mock_page
mock_config.assert_called_once()
mock_pw.chromium.launch.assert_called_once_with(headless=True)

@pytest.mark.asyncio
async def test_get_page_ignores_headless_when_browser_exists(self) -> None:
"""Test that headless parameter is ignored when browser already exists.

This documents the expected behavior: once a browser is launched,
subsequent get_page() calls reuse it regardless of headless parameter.
Call close() first if you need to change the headless mode.
"""
manager = BrowserManager.get_instance()

# Set up existing browser and page
mock_browser = AsyncMock()
mock_page = AsyncMock()
mock_page.is_closed = MagicMock(return_value=False)

BrowserManager._browser = mock_browser
BrowserManager._page = mock_page
BrowserManager._lock = asyncio.Lock()

# Call get_page with headless=False, but browser already exists
page = await manager.get_page(headless=False)

# Should return existing page, not launch new browser
assert page is mock_page
# Browser launch should NOT have been called
mock_browser.new_context.assert_not_called()


class TestBrowserManagerLock:
"""Tests for BrowserManager locking behavior."""

Expand Down