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

fix: address PR #571 review comments

Browse Source 
- Move httpx import to top-level and use anyio task group for concurrent
  validation in cleanup_invalid_app_passwords (storage.py)
- Respect Retry-After header for 429 responses, capped at 300s (oauth_sync.py)
- Soften pre-validation exceptions so transient failures don't crash the
  background sync task (oauth_sync.py)
- Replace f-string SQL with blanket DELETE and add returncode checks (conftest.py)
- Extract clear_stale_test_state() helper to deduplicate cleanup logic
  in astrolabe background sync tests

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-02-20 08:03:55 +01:00
parent 3779ec3e17
commit 81efa6e263
4 changed files with 112 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+24 -22
View File
@@ -2400,29 +2400,31 @@ async def test_users_setup(anyio_backend, nc_client: NextcloudClient):
except Exception as e:
logger.warning(f"Error deleting test user {username}: {e}")
# Clean up app passwords from MCP server to prevent stale scanners
for username in created_users:
try:
import subprocess
# Clean up all app passwords from MCP server to prevent stale scanners
import subprocess
subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"mcp-multi-user-basic",
"sqlite3",
"/app/data/tokens.db",
f"DELETE FROM app_passwords WHERE user_id = '{username}';",
],
capture_output=True,
text=True,
timeout=10,
)
logger.info(f"Cleaned up app password for {username}")
except Exception as e:
logger.debug(f"App password cleanup for {username}: {e}")
result = subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"mcp-multi-user-basic",
"sqlite3",
"/app/data/tokens.db",
"DELETE FROM app_passwords;",
],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode != 0:
logger.warning(
f"Failed to clean up app passwords (rc={result.returncode}): "
f"{result.stderr}"
)
else:
logger.info("Cleaned up all test app passwords")
async def _get_oauth_token_for_user(
tests/integration/test_astrolabe_multi_user_background_sync.py
+66 -87
View File
@@ -834,6 +834,70 @@ async def verify_app_password_created(username: str) -> bool:
return False
def clear_stale_test_state(clear_preferences: bool = False) -> None:
"""Clear stale app passwords, bruteforce entries, and optionally Astrolabe preferences."""
commands: list[tuple[list[str], str]] = [
(
[
"docker",
"compose",
"exec",
"-T",
"mcp-multi-user-basic",
"sqlite3",
"/app/data/tokens.db",
"DELETE FROM app_passwords;",
],
"app passwords",
),
(
[
"docker",
"compose",
"exec",
"-T",
"db",
"mariadb",
"-u",
"root",
"-ppassword",
"nextcloud",
"-e",
"DELETE FROM oc_bruteforce_attempts;",
],
"bruteforce entries",
),
]
if clear_preferences:
commands.append(
(
[
"docker",
"compose",
"exec",
"-T",
"db",
"mariadb",
"-u",
"root",
"-ppassword",
"nextcloud",
"-e",
"DELETE FROM oc_preferences WHERE appid = 'astrolabe';",
],
"Astrolabe preferences",
),
)
for cmd, label in commands:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if result.returncode != 0:
logger.warning(
f"Failed to clear {label} (rc={result.returncode}): {result.stderr}"
)
else:
logger.debug(f"Cleared {label}")
@pytest.mark.integration
@pytest.mark.oauth
async def test_multi_user_astrolabe_background_sync_enablement(
@@ -863,40 +927,7 @@ async def test_multi_user_astrolabe_background_sync_enablement(
"""
# Clear stale state from previous test runs
logger.info("Clearing stale app passwords and bruteforce entries...")
subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"mcp-multi-user-basic",
"sqlite3",
"/app/data/tokens.db",
"DELETE FROM app_passwords;",
],
capture_output=True,
text=True,
timeout=10,
)
subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"db",
"mariadb",
"-u",
"root",
"-ppassword",
"nextcloud",
"-e",
"DELETE FROM oc_bruteforce_attempts;",
],
capture_output=True,
text=True,
timeout=10,
)
clear_stale_test_state()
# Configure Astrolabe to point to the mcp-multi-user-basic server
logger.info("Configuring Astrolabe for mcp-multi-user-basic server...")
@@ -1239,59 +1270,7 @@ async def test_revoke_background_sync_access(
logger.info(
"Clearing stale app passwords, bruteforce entries, and Astrolabe preferences..."
)
subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"mcp-multi-user-basic",
"sqlite3",
"/app/data/tokens.db",
"DELETE FROM app_passwords;",
],
capture_output=True,
text=True,
timeout=10,
)
subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"db",
"mariadb",
"-u",
"root",
"-ppassword",
"nextcloud",
"-e",
"DELETE FROM oc_bruteforce_attempts;",
],
capture_output=True,
text=True,
timeout=10,
)
subprocess.run(
[
"docker",
"compose",
"exec",
"-T",
"db",
"mariadb",
"-u",
"root",
"-ppassword",
"nextcloud",
"-e",
"DELETE FROM oc_preferences WHERE appid = 'astrolabe';",
],
capture_output=True,
text=True,
timeout=10,
)
clear_stale_test_state(clear_preferences=True)
# Configure Astrolabe to point to the mcp-multi-user-basic server
logger.info("Configuring Astrolabe for mcp-multi-user-basic server...")