54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from samples.peripheral.auth import Identity
|
|
from samples.peripheral.config import SampleSettings
|
|
from samples.peripheral.store import PostgresStore
|
|
|
|
|
|
class EventConnection:
|
|
def __init__(self):
|
|
self.statements = []
|
|
self.event_log = True
|
|
|
|
def execute(self, statement, parameters=None):
|
|
self.statements.append((statement, parameters))
|
|
return self
|
|
|
|
def fetchone(self):
|
|
return {"event_seq": 42, "event_log": self.event_log}
|
|
|
|
|
|
@pytest.mark.parametrize(("revision", "deleted", "change"), [
|
|
(1, False, "created"), (2, False, "updated"), (3, True, "deleted"),
|
|
])
|
|
@pytest.mark.parametrize("database", ["tjwater_v2", "fengyang_v2", "physical-project-db"])
|
|
def test_work_order_signals_use_same_connection_and_contain_only_identity(revision, deleted, change, database):
|
|
conn = EventConnection()
|
|
who = Identity(str(uuid4()), str(uuid4()), True)
|
|
order = {"work_order_id": str(uuid4()), "revision": revision,
|
|
"updated_at": "2026-10-08T00:00:00Z", "title": "private title"}
|
|
|
|
PostgresStore(SampleSettings(db_name=database))._work_order_event(conn, who, order, deleted=deleted)
|
|
|
|
statement, parameters = conn.statements[1]
|
|
assert "workbench.domain_event_log" in statement
|
|
assert "'workbench.work_order.changed','peripheral_work_order'" in statement
|
|
assert parameters[0] == who.project_id
|
|
assert parameters[2:5] == (order["work_order_id"], revision, order["updated_at"])
|
|
assert parameters[5].obj == {
|
|
"source": "peripheral_sample", "work_order_id": order["work_order_id"], "change": change,
|
|
}
|
|
assert conn.statements[2] == (
|
|
"SELECT pg_notify('workbench_events', %s)", (f"{who.project_id}:42",),
|
|
)
|
|
|
|
|
|
def test_event_publication_does_not_create_missing_storage():
|
|
conn = EventConnection()
|
|
conn.event_log = False
|
|
store = PostgresStore(SampleSettings(db_name="another_project"))
|
|
store._work_order_event(conn, Identity(str(uuid4()), str(uuid4()), True), {})
|
|
assert conn.statements == [("SELECT to_regclass('workbench.domain_event_log') AS event_log", None)]
|