57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
|
"""Schema/code agreement must be observable (STATE-WP-0083-T07).
|
||
|
|
|
||
|
|
Central served two revisions behind the code with no signal at all. The API
|
||
|
|
started fine and would only have failed when a request touched a missing table.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from api.services import schema_state as ss
|
||
|
|
|
||
|
|
|
||
|
|
class _Session:
|
||
|
|
def __init__(self, revision=None, raises=False):
|
||
|
|
self._revision = revision
|
||
|
|
self._raises = raises
|
||
|
|
|
||
|
|
async def execute(self, *_a, **_k):
|
||
|
|
if self._raises:
|
||
|
|
raise RuntimeError("no connection")
|
||
|
|
rev = self._revision
|
||
|
|
|
||
|
|
class R:
|
||
|
|
def scalar(self_inner):
|
||
|
|
return rev
|
||
|
|
|
||
|
|
return R()
|
||
|
|
|
||
|
|
|
||
|
|
def test_code_head_is_readable_from_the_shipped_migrations():
|
||
|
|
"""Must not require a database: it is what the code expects, not what ran."""
|
||
|
|
assert ss.code_head_revision()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_matching_revision_reports_ok(monkeypatch):
|
||
|
|
monkeypatch.setattr(ss, "code_head_revision", lambda: "abc123")
|
||
|
|
assert (await ss.schema_state(_Session("abc123")))["status"] == "ok"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_behind_revision_is_reported_with_both_values(monkeypatch):
|
||
|
|
monkeypatch.setattr(ss, "code_head_revision", lambda: "newrev")
|
||
|
|
state = await ss.schema_state(_Session("oldrev"))
|
||
|
|
assert state["status"] == "behind"
|
||
|
|
assert state["applied"] == "oldrev" and state["expected"] == "newrev"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_unknown_is_not_reported_as_ok(monkeypatch):
|
||
|
|
"""An instance that cannot establish agreement must not claim it."""
|
||
|
|
monkeypatch.setattr(ss, "code_head_revision", lambda: "newrev")
|
||
|
|
assert (await ss.schema_state(_Session(raises=True)))["status"] == "unknown"
|
||
|
|
monkeypatch.setattr(ss, "code_head_revision", lambda: None)
|
||
|
|
assert (await ss.schema_state(_Session("oldrev")))["status"] == "unknown"
|