71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_push_tomorrow_resolves_unread_update_and_persists_canonical_later_item(
|
|
monkeypatch,
|
|
):
|
|
resolved = []
|
|
applied = []
|
|
|
|
async def confirmed_login():
|
|
return "timmy"
|
|
|
|
async def resolve(kind, repository, number, notification_id):
|
|
resolved.append((kind, repository, number, notification_id))
|
|
return {
|
|
"kind": "update",
|
|
"notification_id": 42,
|
|
"repository": "stackchain/api",
|
|
"title": "Retry failed deploy",
|
|
}
|
|
|
|
class Store:
|
|
def apply(
|
|
self,
|
|
login,
|
|
operation_id,
|
|
action,
|
|
item_id,
|
|
*,
|
|
wake_at=None,
|
|
base_revision=None,
|
|
):
|
|
applied.append(
|
|
(login, operation_id, action, item_id, wake_at, base_revision)
|
|
)
|
|
return {"revision": 3, "records": {item_id: wake_at}}
|
|
|
|
monkeypatch.setattr(main, "_confirmed_login", confirmed_login)
|
|
monkeypatch.setattr(main.gitea_proxy, "resolve_work_route", resolve)
|
|
monkeypatch.setattr(main, "_later_store", lambda: Store())
|
|
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.patch(
|
|
"/api/v1/notifications/42/later",
|
|
json={"wake_at": "2026-08-12T09:00:00.000Z"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"id": 42,
|
|
"status": "deferred",
|
|
"item_id": "update:stackchain/api::42",
|
|
"wake_at": "2026-08-12T09:00:00.000Z",
|
|
"revision": 3,
|
|
}
|
|
assert resolved == [("update", None, None, 42)]
|
|
assert applied == [(
|
|
"timmy",
|
|
"push-tomorrow:42:2026-08-12T09:00:00.000Z",
|
|
"defer",
|
|
"update:stackchain/api::42",
|
|
"2026-08-12T09:00:00.000Z",
|
|
None,
|
|
)]
|