8d73dea7dd
"Sync now" and "Sync all now" now return the freshly reloaded rider row(s) plus an out-of-band toast instead of navigating to a separate result page. The "sync already running" case is a 200 + info toast now instead of a 409 special case. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
261 lines
8.4 KiB
Python
261 lines
8.4 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from app.db.repositories import SchedulerSettingsRepository, SystemLogRepository, UserRepository
|
|
from app.sync.states import SyncOutcome
|
|
|
|
|
|
def _create_user(app, name="Alex") -> int:
|
|
with app.state.session_factory() as session:
|
|
user = UserRepository(session).create(
|
|
name=name,
|
|
enabled=True,
|
|
mywhoosh_email_enc="mw",
|
|
mywhoosh_password_enc="mw-pw",
|
|
garmin_email_enc="g",
|
|
garmin_password_enc="g-pw",
|
|
)
|
|
return user.id
|
|
|
|
|
|
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_updates_row_and_shows_toast(app, authenticated_client, fake_sync_manager) -> None:
|
|
user_id = _create_user(app, "Alex")
|
|
|
|
response = authenticated_client.post(
|
|
f"/users/{user_id}/sync",
|
|
data={"csrf_token": authenticated_client.csrf_token},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert f'id="user-row-{user_id}"' in response.text
|
|
assert 'hx-swap-oob="true"' in response.text
|
|
assert "Alex" in response.text
|
|
assert "0 imported, 0 failed" in response.text
|
|
|
|
|
|
def test_manual_sync_reports_already_running_as_toast(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 == 200
|
|
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_sync_all_updates_each_affected_row_and_shows_summary_toast(app, authenticated_client, fake_sync_manager) -> None:
|
|
user_id = _create_user(app, "Alex")
|
|
|
|
async def fake_sync_all_enabled():
|
|
return [SyncOutcome(user_id=user_id, status="success", discovered=2, imported=2, skipped=0, failed=0)]
|
|
|
|
fake_sync_manager.sync_all_enabled = fake_sync_all_enabled
|
|
|
|
response = authenticated_client.post(
|
|
"/sync-all",
|
|
data={"csrf_token": authenticated_client.csrf_token},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert f'id="user-row-{user_id}"' in response.text
|
|
assert "Synced 1 riders" in response.text
|
|
|
|
|
|
def test_sync_all_shows_toast_when_nothing_to_sync(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 "No riders to sync" in response.text
|
|
|
|
|
|
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
|
|
|
|
|
|
class _FakeScheduler:
|
|
def __init__(self) -> None:
|
|
self.last_tick = None
|
|
self.next_tick = None
|
|
|
|
|
|
def test_system_page_shows_empty_log_state(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
|
|
response = authenticated_client.get("/system")
|
|
assert response.status_code == 200
|
|
assert "No system log entries" in response.text
|
|
|
|
|
|
def test_system_page_shows_recorded_log_entries(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
with app.state.session_factory() as session:
|
|
SystemLogRepository(session).add(
|
|
source="email_notification", message="Failed to email alerts@example.com: SMTP timeout"
|
|
)
|
|
|
|
response = authenticated_client.get("/system")
|
|
assert response.status_code == 200
|
|
assert "email_notification" in response.text
|
|
assert "SMTP timeout" in response.text
|
|
|
|
|
|
def _extract_csrf(html: str) -> str:
|
|
marker = 'name="csrf_token" value="'
|
|
start = html.index(marker) + len(marker)
|
|
end = html.index('"', start)
|
|
return html[start:end]
|
|
|
|
|
|
def test_scheduler_settings_page_prefills_defaults(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
|
|
response = authenticated_client.get("/system")
|
|
|
|
assert response.status_code == 200
|
|
assert 'id="day_start_hour"' in response.text
|
|
assert 'value="6"' in response.text
|
|
assert 'value="22"' in response.text
|
|
|
|
|
|
def test_update_scheduler_settings_persists_and_takes_effect_next_tick(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
page = authenticated_client.get("/system")
|
|
csrf = _extract_csrf(page.text)
|
|
|
|
response = authenticated_client.post(
|
|
"/system/scheduler-settings",
|
|
data={
|
|
"csrf_token": csrf,
|
|
"day_start_hour": "8",
|
|
"night_start_hour": "20",
|
|
"day_interval_minutes": "3",
|
|
"night_interval_minutes": "45",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/system"
|
|
|
|
with app.state.session_factory() as session:
|
|
row = SchedulerSettingsRepository(session).get_or_create(default_minutes=5)
|
|
assert row.day_start_hour == 8
|
|
assert row.night_start_hour == 20
|
|
assert row.day_interval_minutes == 3
|
|
assert row.night_interval_minutes == 45
|
|
|
|
|
|
def test_update_scheduler_settings_rejects_out_of_range_hour(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
page = authenticated_client.get("/system")
|
|
csrf = _extract_csrf(page.text)
|
|
|
|
response = authenticated_client.post(
|
|
"/system/scheduler-settings",
|
|
data={
|
|
"csrf_token": csrf,
|
|
"day_start_hour": "24",
|
|
"night_start_hour": "22",
|
|
"day_interval_minutes": "5",
|
|
"night_interval_minutes": "5",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_update_scheduler_settings_rejects_non_positive_interval(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
page = authenticated_client.get("/system")
|
|
csrf = _extract_csrf(page.text)
|
|
|
|
response = authenticated_client.post(
|
|
"/system/scheduler-settings",
|
|
data={
|
|
"csrf_token": csrf,
|
|
"day_start_hour": "6",
|
|
"night_start_hour": "22",
|
|
"day_interval_minutes": "0",
|
|
"night_interval_minutes": "5",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_update_scheduler_settings_requires_admin(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/system/scheduler-settings",
|
|
data={
|
|
"csrf_token": "whatever",
|
|
"day_start_hour": "6",
|
|
"night_start_hour": "22",
|
|
"day_interval_minutes": "5",
|
|
"night_interval_minutes": "5",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
|
|
|
|
def test_update_scheduler_settings_rejects_invalid_csrf(app, authenticated_client) -> None:
|
|
app.state.scheduler = _FakeScheduler()
|
|
response = authenticated_client.post(
|
|
"/system/scheduler-settings",
|
|
data={
|
|
"csrf_token": "invalid-token",
|
|
"day_start_hour": "6",
|
|
"night_start_hour": "22",
|
|
"day_interval_minutes": "5",
|
|
"night_interval_minutes": "5",
|
|
},
|
|
)
|
|
assert response.status_code == 403
|