feat: add operational sync controls
This commit is contained in:
62
tests/web/test_operations.py
Normal file
62
tests/web/test_operations.py
Normal file
@@ -0,0 +1,62 @@
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def test_manual_sync_calls_shared_manager(authenticated_client, fake_sync_manager) -> None:
|
||||
response = authenticated_client.post(
|
||||
"/users/1/sync",
|
||||
data={"csrf_token": authenticated_client.csrf_token},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert fake_sync_manager.user_calls == [1]
|
||||
|
||||
|
||||
def test_manual_sync_reports_already_running(authenticated_client, fake_sync_manager) -> None:
|
||||
fake_sync_manager.raise_already_running = True
|
||||
response = authenticated_client.post(
|
||||
"/users/1/sync",
|
||||
data={"csrf_token": authenticated_client.csrf_token},
|
||||
)
|
||||
assert response.status_code == 409
|
||||
assert "already running" in response.text.lower()
|
||||
|
||||
|
||||
def test_sync_all_calls_shared_manager(authenticated_client, fake_sync_manager) -> None:
|
||||
response = authenticated_client.post(
|
||||
"/sync-all",
|
||||
data={"csrf_token": authenticated_client.csrf_token},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert fake_sync_manager.all_calls == 1
|
||||
|
||||
|
||||
def test_manual_sync_requires_admin(client: TestClient) -> None:
|
||||
response = client.post(
|
||||
"/users/1/sync",
|
||||
data={"csrf_token": "whatever"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 303
|
||||
assert response.headers["location"] == "/login"
|
||||
|
||||
|
||||
def test_manual_sync_rejects_invalid_csrf(authenticated_client) -> None:
|
||||
response = authenticated_client.post(
|
||||
"/users/1/sync",
|
||||
data={"csrf_token": "invalid-token"},
|
||||
)
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_system_page_shows_scheduler_state(app, authenticated_client) -> None:
|
||||
class FakeScheduler:
|
||||
def __init__(self) -> None:
|
||||
self.last_tick = None
|
||||
self.next_tick = None
|
||||
|
||||
app.state.scheduler = FakeScheduler()
|
||||
|
||||
response = authenticated_client.get("/system")
|
||||
assert response.status_code == 200
|
||||
assert "1.0.0" in response.text
|
||||
assert "5" in response.text # sync_interval_minutes
|
||||
assert "0" in response.text # user_count / activity_count fresh DB
|
||||
Reference in New Issue
Block a user