test: Add tests for dcr endpoint and update oidc app
This commit is contained in:
Chris Coutinho
@@ -0,0 +1,232 @@
|
||||
"""
|
||||
Test the new DCR deletion implementation.
|
||||
|
||||
This test verifies that the recently implemented DCR deletion branch works correctly.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
pytestmark = [pytest.mark.integration, pytest.mark.oauth]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_new_dcr_registration_includes_access_token(
|
||||
anyio_backend,
|
||||
oauth_callback_server,
|
||||
):
|
||||
"""
|
||||
Test that registration now includes registration_access_token.
|
||||
|
||||
The new DCR deletion implementation should provide a registration_access_token
|
||||
in the registration response per RFC 7592.
|
||||
"""
|
||||
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
|
||||
if not nextcloud_host:
|
||||
pytest.skip("Test requires NEXTCLOUD_HOST")
|
||||
|
||||
auth_states, callback_url = oauth_callback_server
|
||||
|
||||
# Discover OIDC endpoints
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
|
||||
discovery_response = await client.get(discovery_url)
|
||||
discovery_response.raise_for_status()
|
||||
oidc_config = discovery_response.json()
|
||||
|
||||
registration_endpoint = oidc_config.get("registration_endpoint")
|
||||
|
||||
# Register a client and inspect the full response
|
||||
client_metadata = {
|
||||
"client_name": "DCR New Implementation Test",
|
||||
"redirect_uris": [callback_url],
|
||||
"token_endpoint_auth_method": "client_secret_post",
|
||||
"grant_types": ["authorization_code", "refresh_token"],
|
||||
"response_types": ["code"],
|
||||
"scope": "openid profile email",
|
||||
"token_type": "Bearer",
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
logger.info("Registering client to check for registration_access_token...")
|
||||
response = await client.post(
|
||||
registration_endpoint,
|
||||
json=client_metadata,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
registration_data = response.json()
|
||||
|
||||
# Log the full response
|
||||
logger.info(f"\n{'=' * 70}")
|
||||
logger.info("REGISTRATION RESPONSE")
|
||||
logger.info(f"{'=' * 70}")
|
||||
logger.info(f"Response keys: {sorted(registration_data.keys())}")
|
||||
logger.info("\nFull response:")
|
||||
for key, value in sorted(registration_data.items()):
|
||||
if key in ["client_secret", "registration_access_token"]:
|
||||
# Truncate secrets for security
|
||||
logger.info(f" {key}: {value[:20]}... (truncated)")
|
||||
else:
|
||||
logger.info(f" {key}: {value}")
|
||||
|
||||
# Check for RFC 7592 required fields
|
||||
logger.info(f"\n{'=' * 70}")
|
||||
logger.info("RFC 7592 COMPLIANCE CHECK")
|
||||
logger.info(f"{'=' * 70}")
|
||||
|
||||
has_token = "registration_access_token" in registration_data
|
||||
has_uri = "registration_client_uri" in registration_data
|
||||
|
||||
logger.info(f"registration_access_token present: {has_token}")
|
||||
logger.info(f"registration_client_uri present: {has_uri}")
|
||||
|
||||
if has_token and has_uri:
|
||||
logger.info(
|
||||
"\n✓ PASS: Registration response includes RFC 7592 management fields!"
|
||||
)
|
||||
logger.info(
|
||||
" This means DCR deletion should now work with Bearer token authentication."
|
||||
)
|
||||
|
||||
# Store these for deletion test
|
||||
client_id = registration_data["client_id"]
|
||||
registration_access_token = registration_data["registration_access_token"]
|
||||
registration_client_uri = registration_data.get("registration_client_uri")
|
||||
|
||||
# Now test deletion with the registration_access_token
|
||||
logger.info(f"\n{'=' * 70}")
|
||||
logger.info("TESTING DCR DELETION WITH REGISTRATION_ACCESS_TOKEN")
|
||||
logger.info(f"{'=' * 70}")
|
||||
|
||||
deletion_endpoint = (
|
||||
registration_client_uri
|
||||
or f"{nextcloud_host}/apps/oidc/register/{client_id}"
|
||||
)
|
||||
logger.info(f"Deletion endpoint: {deletion_endpoint}")
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
# Try deletion with Bearer token (RFC 7592 standard)
|
||||
logger.info("\nAttempting deletion with Bearer token...")
|
||||
delete_response = await client.delete(
|
||||
deletion_endpoint,
|
||||
headers={"Authorization": f"Bearer {registration_access_token}"},
|
||||
)
|
||||
|
||||
logger.info(f"Response status: {delete_response.status_code}")
|
||||
logger.info(f"Response body: {delete_response.text[:200]}")
|
||||
|
||||
if delete_response.status_code == 204:
|
||||
logger.info(
|
||||
"\n✓✓✓ SUCCESS! DCR deletion works with new implementation!"
|
||||
)
|
||||
logger.info(" RFC 7592 deletion is now fully functional.")
|
||||
assert True
|
||||
elif delete_response.status_code == 401:
|
||||
logger.error(
|
||||
"\n✗ FAIL: Still getting 401 even with registration_access_token"
|
||||
)
|
||||
logger.error(
|
||||
" The token may not be recognized or there's a middleware issue."
|
||||
)
|
||||
pytest.fail(
|
||||
"DCR deletion failed with 401 even with registration_access_token"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"\n? UNEXPECTED: Got status {delete_response.status_code}"
|
||||
)
|
||||
pytest.fail(
|
||||
f"Unexpected status code: {delete_response.status_code}, body: {delete_response.text[:500]}"
|
||||
)
|
||||
|
||||
else:
|
||||
logger.warning(
|
||||
"\n✗ FAIL: Registration response still missing RFC 7592 management fields"
|
||||
)
|
||||
logger.warning(
|
||||
" The new DCR deletion implementation may not be active or needs configuration."
|
||||
)
|
||||
pytest.fail(
|
||||
f"Registration response missing RFC 7592 fields. "
|
||||
f"Has token: {has_token}, Has URI: {has_uri}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dcr_deletion_with_basic_auth_new_impl(
|
||||
anyio_backend,
|
||||
oauth_callback_server,
|
||||
):
|
||||
"""
|
||||
Verify whether HTTP Basic Auth is now supported for deletion.
|
||||
|
||||
Some implementations support both Bearer token and Basic Auth.
|
||||
"""
|
||||
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
|
||||
if not nextcloud_host:
|
||||
pytest.skip("Test requires NEXTCLOUD_HOST")
|
||||
|
||||
auth_states, callback_url = oauth_callback_server
|
||||
|
||||
# Discover and register
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
|
||||
discovery_response = await client.get(discovery_url)
|
||||
discovery_response.raise_for_status()
|
||||
oidc_config = discovery_response.json()
|
||||
registration_endpoint = oidc_config.get("registration_endpoint")
|
||||
|
||||
# Register
|
||||
client_metadata = {
|
||||
"client_name": "DCR Basic Auth Test",
|
||||
"redirect_uris": [callback_url],
|
||||
"token_endpoint_auth_method": "client_secret_post",
|
||||
"grant_types": ["authorization_code"],
|
||||
"response_types": ["code"],
|
||||
"scope": "openid profile email",
|
||||
"token_type": "Bearer",
|
||||
}
|
||||
|
||||
response = await client.post(
|
||||
registration_endpoint,
|
||||
json=client_metadata,
|
||||
)
|
||||
response.raise_for_status()
|
||||
reg_data = response.json()
|
||||
|
||||
client_id = reg_data["client_id"]
|
||||
client_secret = reg_data["client_secret"]
|
||||
deletion_endpoint = f"{nextcloud_host}/apps/oidc/register/{client_id}"
|
||||
|
||||
logger.info(f"\n{'=' * 70}")
|
||||
logger.info("TESTING DCR DELETION WITH HTTP BASIC AUTH")
|
||||
logger.info(f"{'=' * 70}")
|
||||
logger.info(f"Endpoint: {deletion_endpoint}")
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
response = await client.delete(
|
||||
deletion_endpoint,
|
||||
auth=(client_id, client_secret),
|
||||
)
|
||||
|
||||
logger.info(f"Status: {response.status_code}")
|
||||
logger.info(f"Body: {response.text[:200]}")
|
||||
|
||||
if response.status_code == 204:
|
||||
logger.info("\n✓ SUCCESS: HTTP Basic Auth works for deletion!")
|
||||
elif response.status_code == 401:
|
||||
logger.info(
|
||||
"\n✗ HTTP Basic Auth not supported - use registration_access_token instead"
|
||||
)
|
||||
else:
|
||||
logger.warning(f"\n? Unexpected status: {response.status_code}")
|
||||
|
||||
# This test is informational - we don't fail if Basic Auth doesn't work
|
||||
# as long as Bearer token works
|
||||
assert True
|
||||
Reference in New Issue
Block a user