admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into docs/adr-024-dynaconf-config-management

Browse Source
This commit is contained in:
Chris Coutinho
2026-04-07 14:17:57 +02:00
parent b8b1616897 6f16ece92b
commit 5b093e49b1
48 changed files with 733 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+33 -33
View File
@@ -29,43 +29,43 @@ logger = logging.getLogger(__name__)
# Default scopes for OAuth testing - all app-specific read/write scopes
DEFAULT_FULL_SCOPES = (
"openid profile email "
"notes:read notes:write "
"calendar:read calendar:write "
"todo:read todo:write "
"contacts:read contacts:write "
"cookbook:read cookbook:write "
"deck:read deck:write "
"tables:read tables:write "
"files:read files:write "
"sharing:read sharing:write"
"notes.read notes.write "
"calendar.read calendar.write "
"todo.read todo.write "
"contacts.read contacts.write "
"cookbook.read cookbook.write "
"deck.read deck.write "
"tables.read tables.write "
"files.read files.write "
"sharing.read sharing.write"
)
# Read-only scopes (all read scopes across apps) - should match DEFAULT_FULL_SCOPES read portion
DEFAULT_READ_SCOPES = (
"openid profile email "
"notes:read "
"calendar:read "
"todo:read "
"contacts:read "
"cookbook:read "
"deck:read "
"tables:read "
"files:read "
"sharing:read"
"notes.read "
"calendar.read "
"todo.read "
"contacts.read "
"cookbook.read "
"deck.read "
"tables.read "
"files.read "
"sharing.read"
)
# Write-only scopes (all write scopes across apps) - should match DEFAULT_FULL_SCOPES write portion
DEFAULT_WRITE_SCOPES = (
"openid profile email "
"notes:write "
"calendar:write "
"todo:write "
"contacts:write "
"cookbook:write "
"deck:write "
"tables:write "
"files:write "
"sharing:write"
"notes.write "
"calendar.write "
"todo.write "
"contacts.write "
"cookbook.write "
"deck.write "
"tables.write "
"files.write "
"sharing.write"
)
@@ -545,7 +545,7 @@ async def nc_mcp_oauth_client_no_custom_scopes(
Connects to the OAuth-enabled MCP server on port 8001.
This client has only OIDC default scopes (openid, profile, email) without
application-specific scopes (notes:read, notes:write, etc.).
application-specific scopes (notes.read, notes.write, etc.).
Expected behavior: Should see 0 tools (all tools require custom scopes).
@@ -1671,7 +1671,7 @@ async def no_custom_scopes_oauth_client_credentials(
Fixture for OAuth client with NO custom scopes (only OIDC defaults).
Tests the security behavior when a user grants only the default OIDC scopes
(openid, profile, email) but declines custom application scopes (notes:read, notes:write, etc.).
(openid, profile, email) but declines custom application scopes (notes.read, notes.write, etc.).
The client is automatically deleted from Nextcloud after the test session completes.
@@ -1808,7 +1808,7 @@ async def playwright_oauth_token(
f"client_id={client_id}&"
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
f"scope=openid%20profile%20email%20notes.read%20notes.write%20calendar.read%20calendar.write%20contacts.read%20contacts.write%20cookbook.read%20cookbook.write%20deck.read%20deck.write%20tables.read%20tables.write%20files.read%20files.write%20sharing.read%20sharing.write"
)
# Add resource parameter (RFC 8707) if available
@@ -2060,7 +2060,7 @@ async def _get_oauth_token_with_scopes(
browser: Playwright browser instance
shared_oauth_client_credentials: Tuple of OAuth client credentials
oauth_callback_server: OAuth callback server fixture
scopes: Space-separated list of scopes (e.g., "openid profile email notes:read")
scopes: Space-separated list of scopes (e.g., "openid profile email notes.read")
resource: Optional resource parameter (RFC 8707) for token audience
mcp_server_base_url: Base URL of the MCP server for resource metadata discovery
@@ -2521,7 +2521,7 @@ async def _get_oauth_token_for_user(
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"resource={quote(mcp_server_resource, safe='')}&" # Resource URI from PRM
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
f"scope=openid%20profile%20email%20notes.read%20notes.write%20calendar.read%20calendar.write%20contacts.read%20contacts.write%20cookbook.read%20cookbook.write%20deck.read%20deck.write%20tables.read%20tables.write%20files.read%20files.write%20sharing.read%20sharing.write"
)
logger.info(f"Performing browser OAuth flow for {username}...")
@@ -3038,7 +3038,7 @@ async def configure_astrolabe_for_mcp_server(nc_client):
"--resource_url",
mcp_server_public_url,
"--allowed_scopes",
"openid profile email offline_access notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write",
"openid profile email offline_access notes.read notes.write calendar.read calendar.write contacts.read contacts.write cookbook.read cookbook.write deck.read deck.write tables.read tables.write files.read files.write",
],
check=True,
capture_output=True,
tests/server/login_flow/test_dcr_lifecycle.py
+4 -4
View File
@@ -37,7 +37,7 @@ async def get_oauth_token_with_client(
authorization_endpoint: str,
callback_url: str,
auth_states: dict,
scopes: str = "openid profile email notes:read notes:write",
scopes: str = "openid profile email notes.read notes.write",
) -> str:
"""
Helper to obtain OAuth access token using existing client credentials.
@@ -187,7 +187,7 @@ async def test_dcr_register_and_delete_lifecycle(
"token_endpoint_auth_method": "client_secret_post",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"scope": "openid profile email notes:read",
"scope": "openid profile email notes.read",
"token_type": "Bearer",
}
@@ -209,7 +209,7 @@ async def test_dcr_register_and_delete_lifecycle(
registration_endpoint=registration_endpoint,
client_name="DCR Lifecycle Test Client 2",
redirect_uris=[callback_url],
scopes="openid profile email notes:read",
scopes="openid profile email notes.read",
token_type="Bearer",
)
@@ -235,7 +235,7 @@ async def test_dcr_register_and_delete_lifecycle(
authorization_endpoint=authorization_endpoint,
callback_url=callback_url,
auth_states=auth_states,
scopes="openid profile email notes:read",
scopes="openid profile email notes.read",
)
assert access_token, "Failed to obtain access token"
tests/server/login_flow/test_dcr_token_type.py
+5 -5
View File
@@ -93,7 +93,7 @@ async def get_oauth_token_with_client(
authorization_endpoint: str,
callback_url: str,
auth_states: dict,
scopes: str = "openid profile email notes:read notes:write",
scopes: str = "openid profile email notes.read notes.write",
) -> str:
"""
Helper to obtain OAuth access token using existing client credentials.
@@ -241,7 +241,7 @@ async def test_dcr_respects_jwt_token_type(
registration_endpoint=registration_endpoint,
client_name="DCR Test - JWT Token Type",
redirect_uris=[callback_url],
scopes="openid profile email notes:read notes:write",
scopes="openid profile email notes.read notes.write",
token_type="jwt",
)
@@ -276,8 +276,8 @@ async def test_dcr_respects_jwt_token_type(
# Verify scope claim exists (critical for MCP tool filtering)
assert "scope" in payload, "JWT payload missing 'scope' claim"
scopes = payload["scope"].split()
assert "notes:read" in scopes, "JWT scope claim missing notes:read"
assert "notes:write" in scopes, "JWT scope claim missing notes:write"
assert "notes.read" in scopes, "JWT scope claim missing notes.read"
assert "notes.write" in scopes, "JWT scope claim missing notes.write"
logger.info(
f"✅ DCR with token_type=jwt works correctly! "
@@ -325,7 +325,7 @@ async def test_dcr_respects_bearer_token_type(
registration_endpoint=registration_endpoint,
client_name="DCR Test - Opaque Token Type",
redirect_uris=[callback_url],
scopes="openid profile email notes:read notes:write",
scopes="openid profile email notes.read notes.write",
token_type="opaque",
)
tests/server/login_flow/test_scope_authorization.py
+62 -62
View File
@@ -31,8 +31,8 @@ async def test_prm_endpoint():
prm_data = response.json()
assert prm_data["resource"] == "http://localhost:8004/mcp"
assert "notes:read" in prm_data["scopes_supported"]
assert "notes:write" in prm_data["scopes_supported"]
assert "notes.read" in prm_data["scopes_supported"]
assert "notes.write" in prm_data["scopes_supported"]
assert "http://localhost:8004" in prm_data["authorization_servers"]
assert "header" in prm_data["bearer_methods_supported"]
assert "RS256" in prm_data["resource_signing_alg_values_supported"]
@@ -67,7 +67,7 @@ async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read
logger = logging.getLogger(__name__)
# Connect with token that has only "notes:read" scope
# Connect with token that has only "notes.read" scope
result = await nc_mcp_login_flow_client_read_only.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -76,13 +76,13 @@ async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read
logger.info(f"Read-only token sees {len(tool_names)} tools")
# Verify read tools are present (only for apps with :read scopes)
# Read-only token has: notes:read, calendar:read, contacts:read,
# cookbook:read, deck:read, tables:read, files:read, sharing:read
# Read-only token has: notes.read, calendar.read, contacts.read,
# cookbook.read, deck.read, tables.read, files.read, sharing.read
expected_read_tools = [
"nc_notes_get_note", # notes:read
"nc_notes_search_notes", # notes:read
"nc_calendar_list_calendars", # calendar:read
"nc_calendar_get_event", # calendar:read
"nc_notes_get_note", # notes.read
"nc_notes_search_notes", # notes.read
"nc_calendar_list_calendars", # calendar.read
"nc_calendar_get_event", # calendar.read
]
for tool in expected_read_tools:
@@ -90,12 +90,12 @@ async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read
# Verify write tools are NOT present (filtered out)
write_tools_should_be_filtered = [
"nc_notes_create_note", # notes:write
"nc_notes_update_note", # notes:write
"nc_notes_delete_note", # notes:write
"nc_calendar_create_event", # calendar:write
"nc_calendar_update_event", # calendar:write
"nc_calendar_delete_event", # calendar:write
"nc_notes_create_note", # notes.write
"nc_notes_update_note", # notes.write
"nc_notes_delete_note", # notes.write
"nc_calendar_create_event", # calendar.write
"nc_calendar_update_event", # calendar.write
"nc_calendar_delete_event", # calendar.write
]
for tool in write_tools_should_be_filtered:
@@ -116,7 +116,7 @@ async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_writ
logger = logging.getLogger(__name__)
# Connect with token that has only "notes:write" scope
# Connect with token that has only "notes.write" scope
result = await nc_mcp_login_flow_client_write_only.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -125,15 +125,15 @@ async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_writ
logger.info(f"Write-only token sees {len(tool_names)} tools")
# Verify write tools are present
# Write-only token has: notes:write, calendar:write, contacts:write,
# cookbook:write, deck:write, tables:write, files:write, sharing:write
# Write-only token has: notes.write, calendar.write, contacts.write,
# cookbook.write, deck.write, tables.write, files.write, sharing.write
expected_write_tools = [
"nc_notes_create_note", # notes:write
"nc_notes_update_note", # notes:write
"nc_notes_delete_note", # notes:write
"nc_calendar_create_event", # calendar:write
"nc_calendar_update_event", # calendar:write
"nc_calendar_delete_event", # calendar:write
"nc_notes_create_note", # notes.write
"nc_notes_update_note", # notes.write
"nc_notes_delete_note", # notes.write
"nc_calendar_create_event", # calendar.write
"nc_calendar_update_event", # calendar.write
"nc_calendar_delete_event", # calendar.write
]
for tool in expected_write_tools:
@@ -141,10 +141,10 @@ async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_writ
# Verify read-only tools are NOT present (write-only scope)
read_tools_should_be_filtered = [
"nc_notes_get_note", # notes:read
"nc_notes_search_notes", # notes:read
"nc_calendar_list_calendars", # calendar:read
"nc_calendar_get_event", # calendar:read
"nc_notes_get_note", # notes.read
"nc_notes_search_notes", # notes.read
"nc_calendar_list_calendars", # calendar.read
"nc_calendar_get_event", # calendar.read
]
for tool in read_tools_should_be_filtered:
@@ -165,7 +165,7 @@ async def test_full_access_token_shows_all_tools(nc_mcp_login_flow_client_full_a
logger = logging.getLogger(__name__)
# Connect with token that has both "notes:read" and "notes:write" scopes
# Connect with token that has both "notes.read" and "notes.write" scopes
result = await nc_mcp_login_flow_client_full_access.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -177,14 +177,14 @@ async def test_full_access_token_shows_all_tools(nc_mcp_login_flow_client_full_a
# Verify both read and write tools are present
# Full access has all *read and *write scopes
expected_read_tools = [
"nc_notes_get_note", # notes:read
"nc_notes_search_notes", # notes:read
"nc_calendar_list_calendars", # calendar:read
"nc_notes_get_note", # notes.read
"nc_notes_search_notes", # notes.read
"nc_calendar_list_calendars", # calendar.read
]
expected_write_tools = [
"nc_notes_create_note", # notes:write
"nc_calendar_create_event", # calendar:write
"nc_notes_create_note", # notes.write
"nc_calendar_create_event", # calendar.write
]
for tool in expected_read_tools:
@@ -217,17 +217,17 @@ async def test_scope_helper_functions():
pass
# Add scope metadata
mock_read_tool._required_scopes = ["notes:read"] # type: ignore
mock_write_tool._required_scopes = ["notes:write"] # type: ignore
mock_read_tool._required_scopes = ["notes.read"] # type: ignore
mock_write_tool._required_scopes = ["notes.write"] # type: ignore
# Test get_required_scopes
assert get_required_scopes(mock_read_tool) == ["notes:read"]
assert get_required_scopes(mock_write_tool) == ["notes:write"]
assert get_required_scopes(mock_read_tool) == ["notes.read"]
assert get_required_scopes(mock_write_tool) == ["notes.write"]
assert get_required_scopes(mock_no_scope_tool) == []
# Test has_required_scopes
read_only_scopes = {"notes:read"}
full_scopes = {"notes:read", "notes:write"}
read_only_scopes = {"notes.read"}
full_scopes = {"notes.read", "notes.write"}
no_scopes = set()
# User with only read scope
@@ -251,13 +251,13 @@ async def test_scope_decorator_stores_metadata():
"""Test that @require_scopes decorator properly stores metadata."""
from nextcloud_mcp_server.auth import require_scopes
@require_scopes("notes:read", "notes:write")
@require_scopes("notes.read", "notes.write")
async def test_function():
pass
# Check that metadata was stored
assert hasattr(test_function, "_required_scopes")
assert test_function._required_scopes == ["notes:read", "notes:write"]
assert test_function._required_scopes == ["notes.read", "notes.write"]
@pytest.mark.integration
@@ -308,28 +308,28 @@ async def test_scope_classification():
from scripts.add_scope_decorators_simple import classify_function
# Test read operations
assert classify_function("nc_notes_get_note") == "notes:read"
assert classify_function("nc_notes_search_notes") == "notes:read"
assert classify_function("nc_calendar_list_events") == "calendar:read"
assert classify_function("nc_webdav_read_file") == "files:read"
assert classify_function("nc_calendar_find_availability") == "calendar:read"
assert classify_function("nc_calendar_get_upcoming_events") == "notes:read"
assert classify_function("nc_notes_get_note") == "notes.read"
assert classify_function("nc_notes_search_notes") == "notes.read"
assert classify_function("nc_calendar_list_events") == "calendar.read"
assert classify_function("nc_webdav_read_file") == "files.read"
assert classify_function("nc_calendar_find_availability") == "calendar.read"
assert classify_function("nc_calendar_get_upcoming_events") == "notes.read"
# Test write operations
assert classify_function("nc_notes_create_note") == "notes:write"
assert classify_function("nc_notes_update_note") == "notes:write"
assert classify_function("nc_notes_delete_note") == "notes:write"
assert classify_function("nc_notes_append_content") == "notes:write"
assert classify_function("nc_calendar_create_event") == "calendar:write"
assert classify_function("nc_calendar_update_event") == "notes:write"
assert classify_function("nc_calendar_manage_calendar") == "notes:write"
assert classify_function("nc_webdav_write_file") == "files:write"
assert classify_function("nc_webdav_move_resource") == "notes:write"
assert classify_function("nc_contacts_create_contact") == "notes:write"
assert classify_function("nc_cookbook_import_recipe") == "notes:write"
assert classify_function("nc_tables_insert_row") == "notes:write"
assert classify_function("deck_archive_card") == "notes:write"
assert classify_function("deck_assign_label_to_card") == "notes:write"
assert classify_function("nc_notes_create_note") == "notes.write"
assert classify_function("nc_notes_update_note") == "notes.write"
assert classify_function("nc_notes_delete_note") == "notes.write"
assert classify_function("nc_notes_append_content") == "notes.write"
assert classify_function("nc_calendar_create_event") == "calendar.write"
assert classify_function("nc_calendar_update_event") == "notes.write"
assert classify_function("nc_calendar_manage_calendar") == "notes.write"
assert classify_function("nc_webdav_write_file") == "files.write"
assert classify_function("nc_webdav_move_resource") == "notes.write"
assert classify_function("nc_contacts_create_contact") == "notes.write"
assert classify_function("nc_cookbook_import_recipe") == "notes.write"
assert classify_function("nc_tables_insert_row") == "notes.write"
assert classify_function("deck_archive_card") == "notes.write"
assert classify_function("deck_assign_label_to_card") == "notes.write"
@pytest.mark.skip(reason="Script no longer exists - decorators are already in place")
tests/test_cli.py
+9 -9
View File
@@ -251,15 +251,15 @@ def test_default_values(runner, clean_env, monkeypatch):
# Verify default values
assert captured_env["NEXTCLOUD_OIDC_SCOPES"] == (
"openid profile email "
"notes:read notes:write "
"calendar:read calendar:write "
"todo:read todo:write "
"contacts:read contacts:write "
"cookbook:read cookbook:write "
"deck:read deck:write "
"tables:read tables:write "
"files:read files:write "
"sharing:read sharing:write"
"notes.read notes.write "
"calendar.read calendar.write "
"todo.read todo.write "
"contacts.read contacts.write "
"cookbook.read cookbook.write "
"deck.read deck.write "
"tables.read tables.write "
"files.read files.write "
"sharing.read sharing.write"
)
assert captured_env["NEXTCLOUD_OIDC_TOKEN_TYPE"] == "bearer"
assert captured_env["NEXTCLOUD_MCP_SERVER_URL"] == "http://localhost:8000"
tests/unit/api/test_access.py
+8 -8
View File
@@ -100,7 +100,7 @@ class TestGetUserAccess:
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="test-app-pw",
scopes=["notes:read", "calendar:write"],
scopes=["notes.read", "calendar.write"],
username="alice_nc",
)
@@ -115,7 +115,7 @@ class TestGetUserAccess:
data = resp.json()
assert data["success"] is True
assert data["provisioned"] is True
assert set(data["scopes"]) == {"notes:read", "calendar:write"}
assert set(data["scopes"]) == {"notes.read", "calendar.write"}
assert data["username"] == "alice_nc"
async def test_missing_auth_header(self, temp_storage):
@@ -146,7 +146,7 @@ class TestUpdateUserScopes:
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="test-app-pw",
scopes=["notes:read"],
scopes=["notes.read"],
username="alice_nc",
)
@@ -156,19 +156,19 @@ class TestUpdateUserScopes:
resp = client.patch(
"/api/v1/users/alice/scopes",
headers={"Authorization": create_basic_auth_header("alice", "pw")},
json={"scopes": ["notes:read", "notes:write", "calendar:read"]},
json={"scopes": ["notes.read", "notes.write", "calendar.read"]},
)
assert resp.status_code == 200
data = resp.json()
assert data["success"] is True
assert set(data["scopes"]) == {"notes:read", "notes:write", "calendar:read"}
assert set(data["scopes"]) == {"notes.read", "notes.write", "calendar.read"}
async def test_invalid_scopes(self, temp_storage):
"""Returns 400 for invalid scope names."""
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="test-app-pw",
scopes=["notes:read"],
scopes=["notes.read"],
)
app = create_test_app(temp_storage)
@@ -177,7 +177,7 @@ class TestUpdateUserScopes:
resp = client.patch(
"/api/v1/users/alice/scopes",
headers={"Authorization": create_basic_auth_header("alice", "pw")},
json={"scopes": ["notes:read", "invalid:scope"]},
json={"scopes": ["notes.read", "invalid:scope"]},
)
assert resp.status_code == 400
data = resp.json()
@@ -192,7 +192,7 @@ class TestUpdateUserScopes:
resp = client.patch(
"/api/v1/users/alice/scopes",
headers={"Authorization": create_basic_auth_header("alice", "pw")},
json={"scopes": ["notes:read"]},
json={"scopes": ["notes.read"]},
)
assert resp.status_code == 404
data = resp.json()
tests/unit/test_auth_tools.py
+12 -12
View File
@@ -38,14 +38,14 @@ async def test_store_app_password_with_scopes(temp_storage):
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee",
scopes=["notes:read", "notes:write"],
scopes=["notes.read", "notes.write"],
username="alice_nc",
)
data = await temp_storage.get_app_password_with_scopes("alice")
assert data is not None
assert data["app_password"] == "aaaaa-bbbbb-ccccc-ddddd-eeeee"
assert data["scopes"] == ["notes:read", "notes:write"]
assert data["scopes"] == ["notes.read", "notes.write"]
assert data["username"] == "alice_nc"
assert data["created_at"] is not None
assert data["updated_at"] is not None
@@ -70,18 +70,18 @@ async def test_store_app_password_with_scopes_replaces(temp_storage):
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee",
scopes=["notes:read"],
scopes=["notes.read"],
)
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="xxxxx-yyyyy-zzzzz-aaaaa-bbbbb",
scopes=["notes:read", "calendar:read"],
scopes=["notes.read", "calendar.read"],
username="alice_nc",
)
data = await temp_storage.get_app_password_with_scopes("alice")
assert data["app_password"] == "xxxxx-yyyyy-zzzzz-aaaaa-bbbbb"
assert data["scopes"] == ["notes:read", "calendar:read"]
assert data["scopes"] == ["notes.read", "calendar.read"]
async def test_get_app_password_with_scopes_nonexistent(temp_storage):
@@ -99,14 +99,14 @@ async def test_store_and_get_login_flow_session(temp_storage):
user_id="alice",
poll_token="secret-poll-token",
poll_endpoint="https://cloud.example.com/login/v2/poll",
requested_scopes=["notes:read", "notes:write"],
requested_scopes=["notes.read", "notes.write"],
)
session = await temp_storage.get_login_flow_session("alice")
assert session is not None
assert session["poll_token"] == "secret-poll-token"
assert session["poll_endpoint"] == "https://cloud.example.com/login/v2/poll"
assert session["requested_scopes"] == ["notes:read", "notes:write"]
assert session["requested_scopes"] == ["notes.read", "notes.write"]
assert session["created_at"] is not None
assert session["expires_at"] is not None
@@ -187,11 +187,11 @@ async def test_delete_expired_login_flow_sessions(temp_storage):
def test_all_supported_scopes():
"""Test that ALL_SUPPORTED_SCOPES contains expected scopes."""
assert "notes:read" in ALL_SUPPORTED_SCOPES
assert "notes:write" in ALL_SUPPORTED_SCOPES
assert "calendar:read" in ALL_SUPPORTED_SCOPES
assert "files:read" in ALL_SUPPORTED_SCOPES
assert "deck:read" in ALL_SUPPORTED_SCOPES
assert "notes.read" in ALL_SUPPORTED_SCOPES
assert "notes.write" in ALL_SUPPORTED_SCOPES
assert "calendar.read" in ALL_SUPPORTED_SCOPES
assert "files.read" in ALL_SUPPORTED_SCOPES
assert "deck.read" in ALL_SUPPORTED_SCOPES
# Scopes should be in pairs (read/write)
read_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":read")]
write_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":write")]
tests/unit/test_scope_authorization_stored.py
+2 -2
View File
@@ -29,7 +29,7 @@ async def test_get_stored_scopes_with_scopes():
mock_storage = AsyncMock()
mock_storage.get_app_password_with_scopes.return_value = {
"app_password": "xxxxx",
"scopes": ["notes:read", "calendar:read"],
"scopes": ["notes.read", "calendar.read"],
"username": "alice",
"created_at": 1000,
"updated_at": 1000,
@@ -41,7 +41,7 @@ async def test_get_stored_scopes_with_scopes():
):
result = await _get_stored_scopes("alice")
assert result == ["notes:read", "calendar:read"]
assert result == ["notes.read", "calendar.read"]
async def test_get_stored_scopes_null_scopes():
tests/unit/test_scope_decorator.py
+8 -8
View File
@@ -12,24 +12,24 @@ from nextcloud_mcp_server.auth.scope_authorization import (
def test_scope_decorator_stores_metadata():
"""Test that @require_scopes decorator stores scope requirements as function metadata."""
@require_scopes("notes:read", "notes:write")
@require_scopes("notes.read", "notes.write")
async def example_function():
pass
# Verify metadata is stored
assert hasattr(example_function, "_required_scopes")
assert example_function._required_scopes == ["notes:read", "notes:write"]
assert example_function._required_scopes == ["notes.read", "notes.write"]
@pytest.mark.unit
def test_scope_decorator_with_single_scope():
"""Test decorator with a single scope requirement."""
@require_scopes("calendar:read")
@require_scopes("calendar.read")
async def example_function():
pass
assert example_function._required_scopes == ["calendar:read"]
assert example_function._required_scopes == ["calendar.read"]
@pytest.mark.unit
@@ -46,18 +46,18 @@ def test_scope_decorator_with_no_scopes():
@pytest.mark.unit
def test_insufficient_scope_error():
"""Test InsufficientScopeError exception structure."""
missing = ["notes:write", "calendar:write"]
missing = ["notes.write", "calendar.write"]
error = InsufficientScopeError(missing)
assert error.missing_scopes == missing
assert "notes:write" in str(error)
assert "calendar:write" in str(error)
assert "notes.write" in str(error)
assert "calendar.write" in str(error)
@pytest.mark.unit
def test_insufficient_scope_error_with_custom_message():
"""Test InsufficientScopeError with custom message."""
missing = ["files:write"]
missing = ["files.write"]
custom_msg = "You need more permissions"
error = InsufficientScopeError(missing, custom_msg)
tests/unit/test_token_broker.py
+3 -3
View File
@@ -379,7 +379,7 @@ class TestRefreshTokenRotation:
expires_in,
) = await broker._refresh_access_token_with_scopes(
refresh_token="old_refresh_token_123",
required_scopes=["notes:read"],
required_scopes=["notes.read"],
user_id="admin",
)
@@ -424,7 +424,7 @@ class TestRefreshTokenRotation:
):
await broker._refresh_access_token_with_scopes(
refresh_token="same_refresh_token",
required_scopes=["notes:read"],
required_scopes=["notes.read"],
user_id="admin",
)
@@ -460,7 +460,7 @@ class TestRefreshTokenRotation:
):
await broker._refresh_access_token_with_scopes(
refresh_token="old_token",
required_scopes=["notes:read"],
required_scopes=["notes.read"],
user_id=None, # No user_id
)