diff --git a/tests/server/login_flow/conftest.py b/tests/server/login_flow/conftest.py index d2c47344..d3abb125 100644 --- a/tests/server/login_flow/conftest.py +++ b/tests/server/login_flow/conftest.py @@ -544,3 +544,362 @@ async def nc_mcp_login_flow_client_no_custom_scopes( client_name="Login Flow MCP No Custom Scopes", ): yield session + + +# --------------------------------------------------------------------------- +# Multi-user Login Flow fixtures for permission / isolation tests +# --------------------------------------------------------------------------- + + +async def _get_login_flow_token_for_user( + browser, + login_flow_oauth_client_credentials, + auth_states: dict, + username: str, + password: str, +) -> str: + """Get an OAuth token for a specific user targeting the login-flow MCP server. + + Similar to the global ``_get_oauth_token_for_user`` but hard-wires the + resource / PRM discovery against port 8004. + """ + nextcloud_host = os.getenv("NEXTCLOUD_HOST") + if not nextcloud_host: + pytest.skip("Login Flow tests require NEXTCLOUD_HOST") + + client_id, client_secret, callback_url, token_endpoint, authorization_endpoint = ( + login_flow_oauth_client_credentials + ) + + # Discover resource identifier from the login-flow server + try: + resource_metadata = await get_mcp_server_resource_metadata( + LOGIN_FLOW_MCP_BASE_URL + ) + resource_id = resource_metadata.get("resource") + except Exception: + resource_id = None + + state = secrets.token_urlsafe(32) + + scopes_encoded = quote(DEFAULT_FULL_SCOPES, safe="") + auth_url = ( + f"{authorization_endpoint}?" + f"response_type=code&" + f"client_id={client_id}&" + f"redirect_uri={quote(callback_url, safe='')}&" + f"state={state}&" + f"scope={scopes_encoded}" + ) + if resource_id: + auth_url += f"&resource={quote(resource_id, safe='')}" + + context = await browser.new_context(ignore_https_errors=True) + page = await context.new_page() + + try: + await page.goto(auth_url, wait_until="networkidle", timeout=60000) + current_url = page.url + + # Login + if "/login" in current_url or "/index.php/login" in current_url: + await page.wait_for_selector('input[name="user"]', timeout=10000) + await page.fill('input[name="user"]', username) + await page.fill('input[name="password"]', password) + await page.click('button[type="submit"]') + await page.wait_for_load_state("networkidle", timeout=60000) + + # Wait for OIDC redirect chain to settle + settle_start = time.time() + while time.time() - settle_start < 15: + current_url = page.url + if "/consent" in current_url or "localhost:8081" in current_url: + break + await anyio.sleep(0.5) + + # Handle consent screen + if "/consent" in page.url: + await page.wait_for_load_state("networkidle", timeout=10000) + await _handle_oauth_consent_screen(page, username) + + # Wait for callback + start_time = time.time() + while state not in auth_states: + if time.time() - start_time > 30: + screenshot_path = f"/tmp/login_flow_oauth_timeout_{username}.png" + await page.screenshot(path=screenshot_path) + raise TimeoutError(f"Timeout waiting for OAuth callback for {username}") + await anyio.sleep(0.5) + + auth_code = auth_states[state] + finally: + await context.close() + + # Exchange code for token + async with httpx.AsyncClient(timeout=30.0) as http_client: + token_response = await http_client.post( + token_endpoint, + data={ + "grant_type": "authorization_code", + "code": auth_code, + "redirect_uri": callback_url, + "client_id": client_id, + "client_secret": client_secret, + }, + ) + token_response.raise_for_status() + return token_response.json()["access_token"] + + +@pytest.fixture(scope="session") +async def all_login_flow_user_tokens( + anyio_backend, + browser, + login_flow_oauth_client_credentials, + test_users_setup, + oauth_callback_server, +) -> dict[str, str]: + """Fetch OAuth tokens for all test users in parallel, targeting port 8004.""" + auth_states, _ = oauth_callback_server + + start_time = time.time() + logger.info("Fetching login-flow OAuth tokens for all users in parallel...") + + results: dict[str, str | Exception] = {} + + async def _fetch(username: str, config: dict, delay: float) -> None: + if delay > 0: + await anyio.sleep(delay) + try: + token = await _get_login_flow_token_for_user( + browser, + login_flow_oauth_client_credentials, + auth_states, + username, + config["password"], + ) + results[username] = token + except Exception as exc: + results[username] = exc + + user_list = list(test_users_setup.items()) + async with anyio.create_task_group() as tg: + for idx, (username, config) in enumerate(user_list): + tg.start_soon(_fetch, username, config, idx * 0.5) + + for username, result in results.items(): + if isinstance(result, Exception): + raise result + + elapsed = time.time() - start_time + logger.info( + f"Fetched {len(results)} login-flow tokens in {elapsed:.1f}s " + f"(~{elapsed / len(results):.1f}s per user)" + ) + return results # type: ignore[return-value] + + +async def _provision_login_flow_mcp_client( + token: str, + browser, + username: str, + password: str, +) -> AsyncGenerator[ClientSession, Any]: + """Connect to login-flow MCP server, complete Login Flow v2 provisioning, yield session.""" + login_url_holder: dict[str, str] = {} + + async def elicitation_callback( + context: Any, + params: ElicitRequestParams, + ) -> ElicitResult: + message = params.message + for line in message.split("\n"): + stripped = line.strip() + if stripped.startswith("http") and "/login/v2/" in stripped: + login_url_holder["url"] = stripped + break + + if "url" in login_url_holder: + await _complete_login_flow_v2_as_user( + browser, login_url_holder["url"], username, password + ) + + return ElicitResult(action="accept", content={"acknowledged": True}) + + async for session in create_mcp_client_session( + url=LOGIN_FLOW_MCP_URL, + token=token, + client_name=f"Login Flow MCP ({username})", + elicitation_callback=elicitation_callback, + ): + # Provision access + provision_result = await session.call_tool( + "nc_auth_provision_access", {"scopes": None} + ) + provision_data = json.loads(provision_result.content[0].text) + + if provision_data.get("status") == "login_required": + login_url = provision_data.get("login_url") + if login_url and "url" not in login_url_holder: + await _complete_login_flow_v2_as_user( + browser, login_url, username, password + ) + + # Poll for completion + for attempt in range(15): + status_result = await session.call_tool("nc_auth_check_status", {}) + status_data = json.loads(status_result.content[0].text) + if status_data.get("status") == "provisioned": + logger.info( + f"Login Flow v2 provisioned for {username}: " + f"{status_data.get('username')}" + ) + break + if status_data.get("status") in ("not_initiated", "error"): + raise RuntimeError( + f"Login Flow v2 failed for {username}: {status_data.get('message')}" + ) + await anyio.sleep(2) + else: + raise TimeoutError( + f"Login Flow v2 did not complete for {username} after 15 attempts" + ) + + yield session + + +async def _complete_login_flow_v2_as_user( + browser, login_url: str, username: str, password: str +) -> None: + """Complete Nextcloud Login Flow v2 in a browser as a specific user. + + Same steps as ``_complete_login_flow_v2`` but uses the given *username* and + *password* instead of reading from environment variables. + """ + login_url = _rewrite_login_flow_url(login_url) + + context = await browser.new_context(ignore_https_errors=True) + page = await context.new_page() + + try: + logger.info(f"Opening Login Flow v2 URL for {username}: {login_url[:80]}...") + await page.goto(login_url, wait_until="networkidle", timeout=60000) + + # Step 1: "Connect to your account" page + login_btn = page.get_by_role("button", name="Log in") + try: + await login_btn.wait_for(timeout=10000) + await login_btn.click() + await page.wait_for_load_state("networkidle", timeout=30000) + except Exception: + pass + + # Step 2: Login form + user_field = page.locator('input[name="user"]') + if await user_field.count() > 0: + await user_field.fill(username) + await page.locator('input[name="password"]').fill(password) + await page.get_by_role("button", name="Log in", exact=True).click() + await page.wait_for_load_state("networkidle", timeout=60000) + + # Step 3: "Account access" grant page + grant_btn = page.get_by_role("button", name="Grant access") + try: + await grant_btn.wait_for(timeout=15000) + await grant_btn.click() + except Exception: + pass + + # Step 4: Password confirmation dialog + confirm_password = page.get_by_role("dialog").get_by_role( + "textbox", name="Password" + ) + try: + await confirm_password.wait_for(timeout=10000) + await confirm_password.fill(password) + confirm_btn = page.get_by_role("dialog").get_by_role( + "button", name="Confirm" + ) + await confirm_btn.wait_for(timeout=5000) + await confirm_btn.click() + except Exception: + pass + + # Step 5: Wait for success + try: + await page.get_by_text("Account connected").wait_for(timeout=15000) + logger.info(f"Login Flow v2 completed for {username}") + except Exception: + await page.wait_for_load_state("networkidle", timeout=10000) + logger.info(f"Login Flow v2 done for {username}. URL: {page.url}") + + finally: + await context.close() + + +@pytest.fixture(scope="session") +async def alice_login_flow_mcp_client( + anyio_backend, + all_login_flow_user_tokens: dict[str, str], + test_users_setup, + browser, +) -> AsyncGenerator[ClientSession, Any]: + """MCP client authenticated and provisioned as alice (owner role).""" + async for session in _provision_login_flow_mcp_client( + token=all_login_flow_user_tokens["alice"], + browser=browser, + username="alice", + password=test_users_setup["alice"]["password"], + ): + yield session + + +@pytest.fixture(scope="session") +async def bob_login_flow_mcp_client( + anyio_backend, + all_login_flow_user_tokens: dict[str, str], + test_users_setup, + browser, +) -> AsyncGenerator[ClientSession, Any]: + """MCP client authenticated and provisioned as bob (viewer role).""" + async for session in _provision_login_flow_mcp_client( + token=all_login_flow_user_tokens["bob"], + browser=browser, + username="bob", + password=test_users_setup["bob"]["password"], + ): + yield session + + +@pytest.fixture(scope="session") +async def charlie_login_flow_mcp_client( + anyio_backend, + all_login_flow_user_tokens: dict[str, str], + test_users_setup, + browser, +) -> AsyncGenerator[ClientSession, Any]: + """MCP client authenticated and provisioned as charlie (editor role).""" + async for session in _provision_login_flow_mcp_client( + token=all_login_flow_user_tokens["charlie"], + browser=browser, + username="charlie", + password=test_users_setup["charlie"]["password"], + ): + yield session + + +@pytest.fixture(scope="session") +async def diana_login_flow_mcp_client( + anyio_backend, + all_login_flow_user_tokens: dict[str, str], + test_users_setup, + browser, +) -> AsyncGenerator[ClientSession, Any]: + """MCP client authenticated and provisioned as diana (no-access role).""" + async for session in _provision_login_flow_mcp_client( + token=all_login_flow_user_tokens["diana"], + browser=browser, + username="diana", + password=test_users_setup["diana"]["password"], + ): + yield session diff --git a/tests/server/login_flow/test_multi_user_permissions.py b/tests/server/login_flow/test_multi_user_permissions.py new file mode 100644 index 00000000..a0faee00 --- /dev/null +++ b/tests/server/login_flow/test_multi_user_permissions.py @@ -0,0 +1,540 @@ +"""Multi-user permission tests for Login Flow v2 deployment mode. + +Tests verify that Nextcloud's sharing / ACL enforcement works correctly +when resources are accessed through MCP tools by different users, each +authenticated via Login Flow v2. + +Ported from the removed ``tests/server/oauth/test_oauth_*_permissions.py`` +tests. The underlying assertions are deployment-mode-agnostic; only the +transport changed (OAuth MCP server -> Login Flow v2 MCP server). +""" + +import json +import logging + +import pytest +from mcp import ClientSession + +logger = logging.getLogger(__name__) + +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] + + +# --------------------------------------------------------------------------- +# WebDAV / Files +# --------------------------------------------------------------------------- + + +class TestFilePermissions: + """Test that MCP file tools respect Nextcloud sharing permissions.""" + + async def test_file_share_read_permissions( + self, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + diana_login_flow_mcp_client: ClientSession, + ): + """Alice shares a file with Bob (read-only). Bob can read it; + Diana (unshared) cannot.""" + file_path = "/alice_shared_file_read.txt" + file_content = "This file is shared with Bob for reading only." + + # Alice creates the file + result = await alice_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={"path": file_path, "content": file_content}, + ) + assert not result.isError, f"Alice failed to create file: {result.content}" + + share_id = None + try: + # Alice shares with Bob (read-only, permissions=1) + result = await alice_login_flow_mcp_client.call_tool( + "nc_share_create", + arguments={ + "path": file_path, + "share_with": "bob", + "share_type": 0, + "permissions": 1, + }, + ) + assert not result.isError, f"Share creation failed: {result.content}" + share_data = json.loads(result.content[0].text) + share_id = share_data["id"] + + # Bob reads the file + result = await bob_login_flow_mcp_client.call_tool( + "nc_webdav_read_file", arguments={"path": file_path} + ) + assert not result.isError, ( + f"Bob could not read shared file: {result.content}" + ) + response_data = json.loads(result.content[0].text) + assert file_content in response_data["content"] + + # Diana cannot read the file + result = await diana_login_flow_mcp_client.call_tool( + "nc_webdav_read_file", arguments={"path": file_path} + ) + assert result.isError, "Diana should not be able to read unshared file" + + finally: + if share_id: + await alice_login_flow_mcp_client.call_tool( + "nc_share_delete", arguments={"share_id": share_id} + ) + await alice_login_flow_mcp_client.call_tool( + "nc_webdav_delete_resource", arguments={"path": file_path} + ) + + async def test_file_share_write_permissions( + self, + alice_login_flow_mcp_client: ClientSession, + charlie_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + ): + """Alice shares a file with Charlie (edit) and Bob (read-only). + Charlie can overwrite; Bob cannot.""" + file_path = "/alice_shared_file_write.txt" + file_content = "This file is shared with Charlie for editing." + + result = await alice_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={"path": file_path, "content": file_content}, + ) + assert not result.isError + + charlie_share_id = None + bob_share_id = None + try: + # Share with Charlie (read+write, permissions=3) + result = await alice_login_flow_mcp_client.call_tool( + "nc_share_create", + arguments={ + "path": file_path, + "share_with": "charlie", + "share_type": 0, + "permissions": 3, + }, + ) + assert not result.isError + charlie_share_id = json.loads(result.content[0].text)["id"] + + # Share with Bob (read-only, permissions=1) + result = await alice_login_flow_mcp_client.call_tool( + "nc_share_create", + arguments={ + "path": file_path, + "share_with": "bob", + "share_type": 0, + "permissions": 1, + }, + ) + assert not result.isError + bob_share_id = json.loads(result.content[0].text)["id"] + + # Charlie can write + result = await charlie_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={ + "path": file_path, + "content": f"{file_content}\nCharlie added this line.", + }, + ) + assert not result.isError, ( + f"Charlie should be able to write: {result.content}" + ) + + # Bob cannot write + result = await bob_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={ + "path": file_path, + "content": "Bob tries to overwrite this.", + }, + ) + assert result.isError, "Bob should be denied write access (read-only)" + + finally: + for sid in (charlie_share_id, bob_share_id): + if sid: + await alice_login_flow_mcp_client.call_tool( + "nc_share_delete", arguments={"share_id": sid} + ) + await alice_login_flow_mcp_client.call_tool( + "nc_webdav_delete_resource", arguments={"path": file_path} + ) + + async def test_folder_share_permissions( + self, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + ): + """Alice shares a folder with Bob; Bob can list and read its contents.""" + folder_path = "/alice_shared_folder" + file_in_folder = f"{folder_path}/document.txt" + file_content = "Document in Alice's shared folder" + + result = await alice_login_flow_mcp_client.call_tool( + "nc_webdav_create_directory", arguments={"path": folder_path} + ) + assert not result.isError + + result = await alice_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={"path": file_in_folder, "content": file_content}, + ) + assert not result.isError + + share_id = None + try: + result = await alice_login_flow_mcp_client.call_tool( + "nc_share_create", + arguments={ + "path": folder_path, + "share_with": "bob", + "share_type": 0, + "permissions": 1, + }, + ) + assert not result.isError + share_id = json.loads(result.content[0].text)["id"] + + # Bob lists the shared folder + result = await bob_login_flow_mcp_client.call_tool( + "nc_webdav_list_directory", arguments={"path": folder_path} + ) + assert not result.isError, f"Bob should see shared folder: {result.content}" + response_data = json.loads(result.content[0].text) + file_names = [f["name"] for f in response_data.get("files", [])] + assert "document.txt" in file_names + + # Bob reads the file + result = await bob_login_flow_mcp_client.call_tool( + "nc_webdav_read_file", arguments={"path": file_in_folder} + ) + assert not result.isError + assert file_content in json.loads(result.content[0].text)["content"] + + finally: + if share_id: + await alice_login_flow_mcp_client.call_tool( + "nc_share_delete", arguments={"share_id": share_id} + ) + await alice_login_flow_mcp_client.call_tool( + "nc_webdav_delete_resource", arguments={"path": folder_path} + ) + + async def test_user_isolation_files( + self, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + ): + """Users can only see their own files when nothing is shared.""" + alice_file = "/alice_private_file.txt" + bob_file = "/bob_private_file.txt" + + # Each user creates their own file + result = await alice_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={"path": alice_file, "content": "Alice's private file"}, + ) + assert not result.isError + + result = await bob_login_flow_mcp_client.call_tool( + "nc_webdav_write_file", + arguments={"path": bob_file, "content": "Bob's private file"}, + ) + assert not result.isError + + try: + # Bob lists root — should NOT see Alice's file + result = await bob_login_flow_mcp_client.call_tool( + "nc_webdav_list_directory", arguments={"path": "/"} + ) + assert not result.isError + bob_visible = [ + f["name"] for f in json.loads(result.content[0].text).get("files", []) + ] + assert "alice_private_file.txt" not in bob_visible, ( + "Bob should not see Alice's private file" + ) + + # Alice lists root — should NOT see Bob's file + result = await alice_login_flow_mcp_client.call_tool( + "nc_webdav_list_directory", arguments={"path": "/"} + ) + assert not result.isError + alice_visible = [ + f["name"] for f in json.loads(result.content[0].text).get("files", []) + ] + assert "bob_private_file.txt" not in alice_visible, ( + "Alice should not see Bob's private file" + ) + + finally: + await alice_login_flow_mcp_client.call_tool( + "nc_webdav_delete_resource", arguments={"path": alice_file} + ) + await bob_login_flow_mcp_client.call_tool( + "nc_webdav_delete_resource", arguments={"path": bob_file} + ) + + +# --------------------------------------------------------------------------- +# Deck +# --------------------------------------------------------------------------- + + +class TestDeckPermissions: + """Test that MCP Deck tools respect board ACL permissions.""" + + async def _add_board_acl( + self, nc_client, board_id: int, user: str, permission_type: int = 0 + ) -> int: + """Add ACL entry. permission_type: 0=view, 1=edit, 2=manage.""" + acl = await nc_client.deck.add_acl_rule( + board_id=board_id, + type=0, + participant=user, + permission_edit=permission_type >= 1, + permission_share=permission_type >= 2, + permission_manage=permission_type >= 2, + ) + return acl.id + + async def test_deck_board_view_permissions( + self, + nc_client, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + diana_login_flow_mcp_client: ClientSession, + ): + """Admin creates a board, adds Bob (view). Bob can see it; Diana cannot.""" + board = await nc_client.deck.create_board("Shared Board - View Test", "FF0000") + board_id = board.id + bob_acl_id = None + + try: + bob_acl_id = await self._add_board_acl(nc_client, board_id, "bob", 0) + + # Bob can see the board + result = await bob_login_flow_mcp_client.call_tool( + "deck_get_boards", arguments={} + ) + assert not result.isError + board_ids = [ + b["id"] for b in json.loads(result.content[0].text).get("boards", []) + ] + assert board_id in board_ids, "Bob should see shared board" + + # Diana cannot see the board + result = await diana_login_flow_mcp_client.call_tool( + "deck_get_boards", arguments={} + ) + assert not result.isError + board_ids = [ + b["id"] for b in json.loads(result.content[0].text).get("boards", []) + ] + assert board_id not in board_ids, "Diana should not see board without ACL" + + finally: + if bob_acl_id: + await nc_client.deck.delete_acl_rule(board_id, bob_acl_id) + await nc_client.deck.delete_board(board_id) + + async def test_deck_board_edit_permissions( + self, + nc_client, + charlie_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + ): + """Charlie (edit) can create cards; Bob (view-only) cannot.""" + board = await nc_client.deck.create_board("Shared Board - Edit Test", "00FF00") + board_id = board.id + stack = await nc_client.deck.create_stack(board_id, "Test Stack", 1) + stack_id = stack.id + charlie_acl_id = None + bob_acl_id = None + + try: + charlie_acl_id = await self._add_board_acl( + nc_client, board_id, "charlie", 1 + ) + bob_acl_id = await self._add_board_acl(nc_client, board_id, "bob", 0) + + # Charlie creates a card + result = await charlie_login_flow_mcp_client.call_tool( + "deck_create_card", + arguments={ + "board_id": board_id, + "stack_id": stack_id, + "title": "Charlie's Card", + "description": "Created by Charlie with edit permission", + }, + ) + assert not result.isError, f"Charlie should create cards: {result.content}" + card_id = json.loads(result.content[0].text).get("id") + if card_id: + await nc_client.deck.delete_card(board_id, stack_id, card_id) + + # Bob cannot create a card + result = await bob_login_flow_mcp_client.call_tool( + "deck_create_card", + arguments={ + "board_id": board_id, + "stack_id": stack_id, + "title": "Bob's Card", + "description": "Bob trying to create a card", + }, + ) + assert result.isError, "Bob should be denied card creation (view-only)" + + finally: + for acl_id in (charlie_acl_id, bob_acl_id): + if acl_id: + await nc_client.deck.delete_acl_rule(board_id, acl_id) + await nc_client.deck.delete_board(board_id) + + async def test_deck_user_isolation( + self, + nc_client, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + ): + """Users can only see their own boards when nothing is shared.""" + alice_board = await nc_client.deck.create_board( + "Alice's Private Board", "FF00FF" + ) + bob_board = await nc_client.deck.create_board("Bob's Private Board", "00FFFF") + + try: + # Alice should NOT see Bob's board + result = await alice_login_flow_mcp_client.call_tool( + "deck_get_boards", arguments={} + ) + assert not result.isError + board_ids = [ + b["id"] for b in json.loads(result.content[0].text).get("boards", []) + ] + assert bob_board.id not in board_ids, ( + "Alice should not see Bob's private board" + ) + + # Bob should NOT see Alice's board + result = await bob_login_flow_mcp_client.call_tool( + "deck_get_boards", arguments={} + ) + assert not result.isError + board_ids = [ + b["id"] for b in json.loads(result.content[0].text).get("boards", []) + ] + assert alice_board.id not in board_ids, ( + "Bob should not see Alice's private board" + ) + + finally: + await nc_client.deck.delete_board(alice_board.id) + await nc_client.deck.delete_board(bob_board.id) + + +# --------------------------------------------------------------------------- +# Notes +# --------------------------------------------------------------------------- + + +class TestNotesPermissions: + """Test that MCP Notes tools respect user isolation. + + Nextcloud Notes are inherently single-user (no sharing API). These tests + verify that notes created by one user are invisible to others. + """ + + async def test_user_isolation_notes( + self, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + ): + """Notes created by Alice are invisible to Bob and vice versa.""" + # Alice creates a note + result = await alice_login_flow_mcp_client.call_tool( + "nc_notes_create_note", + arguments={ + "title": "Alice's Private Note", + "content": "This is Alice's private content.", + "category": "PermTest", + }, + ) + assert not result.isError + alice_note_id = json.loads(result.content[0].text)["id"] + + # Bob creates a note + result = await bob_login_flow_mcp_client.call_tool( + "nc_notes_create_note", + arguments={ + "title": "Bob's Private Note", + "content": "This is Bob's private content.", + "category": "PermTest", + }, + ) + assert not result.isError + bob_note_id = json.loads(result.content[0].text)["id"] + + try: + # Alice searches — should NOT see Bob's note + result = await alice_login_flow_mcp_client.call_tool( + "nc_notes_search_notes", arguments={"query": "PermTest"} + ) + assert not result.isError + alice_visible_ids = [ + n["id"] for n in json.loads(result.content[0].text).get("results", []) + ] + assert bob_note_id not in alice_visible_ids, ( + "Alice should not see Bob's private note" + ) + + # Bob searches — should NOT see Alice's note + result = await bob_login_flow_mcp_client.call_tool( + "nc_notes_search_notes", arguments={"query": "PermTest"} + ) + assert not result.isError + bob_visible_ids = [ + n["id"] for n in json.loads(result.content[0].text).get("results", []) + ] + assert alice_note_id not in bob_visible_ids, ( + "Bob should not see Alice's private note" + ) + + finally: + await alice_login_flow_mcp_client.call_tool( + "nc_notes_delete_note", arguments={"note_id": alice_note_id} + ) + await bob_login_flow_mcp_client.call_tool( + "nc_notes_delete_note", arguments={"note_id": bob_note_id} + ) + + +# --------------------------------------------------------------------------- +# Smoke: all multi-user clients initialised +# --------------------------------------------------------------------------- + + +class TestMultiUserSmoke: + """Quick check that all multi-user MCP clients are functional.""" + + async def test_all_clients_can_list_tools( + self, + alice_login_flow_mcp_client: ClientSession, + bob_login_flow_mcp_client: ClientSession, + charlie_login_flow_mcp_client: ClientSession, + diana_login_flow_mcp_client: ClientSession, + ): + for name, client in [ + ("alice", alice_login_flow_mcp_client), + ("bob", bob_login_flow_mcp_client), + ("charlie", charlie_login_flow_mcp_client), + ("diana", diana_login_flow_mcp_client), + ]: + tools = await client.list_tools() + assert len(tools.tools) > 0, f"{name} MCP client has no tools" + logger.info(f"{name} MCP client working ({len(tools.tools)} tools)")