mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-15 17:25:26 -04:00
174 lines
4.7 KiB
Python
174 lines
4.7 KiB
Python
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
|
|
class _AuthManager:
|
|
is_configured = True
|
|
|
|
def __init__(self, admins=()):
|
|
self._admins = set(admins)
|
|
|
|
def is_admin(self, user):
|
|
return user in self._admins
|
|
|
|
|
|
class _Request:
|
|
def __init__(self, body, *, user="alice", admins=()):
|
|
self._body = body
|
|
self.state = SimpleNamespace(current_user=user)
|
|
self.client = SimpleNamespace(host="127.0.0.1")
|
|
self.app = SimpleNamespace(
|
|
state=SimpleNamespace(auth_manager=_AuthManager(admins))
|
|
)
|
|
|
|
async def json(self):
|
|
return self._body
|
|
|
|
|
|
class _Query:
|
|
def __init__(self, note):
|
|
self.note = note
|
|
|
|
def filter(self, *args, **kwargs):
|
|
return self
|
|
|
|
def first(self):
|
|
return self.note
|
|
|
|
|
|
class _Db:
|
|
def __init__(self, note):
|
|
self.note = note
|
|
self.closed = False
|
|
|
|
def query(self, model):
|
|
return _Query(self.note)
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def _endpoint(monkeypatch, note=None):
|
|
import routes.note_routes as note_routes
|
|
|
|
calls = []
|
|
db = _Db(note)
|
|
|
|
async def fake_dispatch_reminder(**kwargs):
|
|
calls.append(kwargs)
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr(note_routes, "SessionLocal", lambda: db)
|
|
monkeypatch.setattr(note_routes, "dispatch_reminder", fake_dispatch_reminder)
|
|
|
|
router = note_routes.setup_note_routes()
|
|
endpoint = next(
|
|
route.endpoint for route in router.routes
|
|
if route.path == "/api/notes/fire-reminder" and "POST" in route.methods
|
|
)
|
|
return endpoint, calls, db
|
|
|
|
|
|
def _note(**overrides):
|
|
data = {
|
|
"id": "note-1",
|
|
"owner": "alice",
|
|
"title": "Stored title",
|
|
"content": "Stored body",
|
|
"items": None,
|
|
}
|
|
data.update(overrides)
|
|
return SimpleNamespace(**data)
|
|
|
|
|
|
def test_real_reminder_requires_owned_note(monkeypatch):
|
|
endpoint, calls, _db = _endpoint(monkeypatch, _note(owner="bob"))
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(endpoint(_Request({"note_id": "note-1"}, user="alice")))
|
|
|
|
assert exc.value.status_code == 404
|
|
assert calls == []
|
|
|
|
|
|
def test_real_reminder_uses_stored_note_and_ignores_overrides(monkeypatch):
|
|
endpoint, calls, db = _endpoint(monkeypatch, _note())
|
|
|
|
result = asyncio.run(endpoint(_Request({
|
|
"note_id": "note-1",
|
|
"title": "Forged title",
|
|
"body": "Forged body",
|
|
"channel": "webhook",
|
|
"webhook_integration_id": "global-webhook",
|
|
"webhook_payload_template": '{"content":"owned"}',
|
|
}, user="alice")))
|
|
|
|
assert result == {"ok": True}
|
|
assert db.closed is True
|
|
assert calls == [{
|
|
"title": "Stored title",
|
|
"note_body": "Stored body",
|
|
"note_id": "note-1",
|
|
"owner": "alice",
|
|
"queue_browser": False,
|
|
"settings_override": None,
|
|
}]
|
|
|
|
|
|
def test_real_checklist_reminder_body_is_built_from_stored_items(monkeypatch):
|
|
endpoint, calls, _db = _endpoint(monkeypatch, _note(items=(
|
|
'[{"text":"first","done":false},'
|
|
'{"text":"finished","done":true},'
|
|
'{"text":"second","checked":false}]'
|
|
)))
|
|
|
|
asyncio.run(endpoint(_Request({"note_id": "note-1"}, user="alice")))
|
|
|
|
assert calls[0]["note_body"] == "Pending (2):\n- first\n- second"
|
|
|
|
|
|
def test_non_admin_cannot_fire_synthetic_test_reminder(monkeypatch):
|
|
endpoint, calls, _db = _endpoint(monkeypatch)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(endpoint(_Request({
|
|
"note_id": "test-123",
|
|
"title": "Test Reminder",
|
|
"body": "Test body",
|
|
"channel": "webhook",
|
|
"webhook_integration_id": "global-webhook",
|
|
}, user="alice")))
|
|
|
|
assert exc.value.status_code == 403
|
|
assert calls == []
|
|
|
|
|
|
def test_admin_test_reminder_can_use_current_ui_overrides(monkeypatch):
|
|
endpoint, calls, _db = _endpoint(monkeypatch)
|
|
|
|
result = asyncio.run(endpoint(_Request({
|
|
"note_id": "test-123",
|
|
"title": "Test Reminder",
|
|
"body": "Test body",
|
|
"channel": "webhook",
|
|
"webhook_integration_id": "global-webhook",
|
|
"webhook_payload_template": '{"content":"{{message}}"}',
|
|
}, user="admin", admins={"admin"})))
|
|
|
|
assert result == {"ok": True}
|
|
assert calls == [{
|
|
"title": "Test Reminder",
|
|
"note_body": "Test body",
|
|
"note_id": "test-123",
|
|
"owner": "admin",
|
|
"queue_browser": False,
|
|
"settings_override": {
|
|
"reminder_channel": "webhook",
|
|
"reminder_webhook_integration_id": "global-webhook",
|
|
"reminder_webhook_payload_template": '{"content":"{{message}}"}',
|
|
},
|
|
}]
|