Files
odysseus/tests/test_note_reminder_fire_scope.py

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}}"}',
},
}]