92 lines
2.3 KiB
Python
92 lines
2.3 KiB
Python
from contextlib import contextmanager
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from scripts.seed_peripheral_sample import (
|
|
ACTOR_ID,
|
|
previous_seed_result,
|
|
restore_seeded_resources,
|
|
)
|
|
|
|
|
|
class FakeResult:
|
|
def __init__(self, row=None):
|
|
self.row = row
|
|
|
|
def fetchone(self):
|
|
return self.row
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self, row=None):
|
|
self.row = row
|
|
self.calls = []
|
|
|
|
def execute(self, statement, parameters):
|
|
self.calls.append((" ".join(statement.split()), parameters))
|
|
return FakeResult(self.row)
|
|
|
|
|
|
class FakeStore:
|
|
def __init__(self, row=None):
|
|
self.connection_value = FakeConnection(row)
|
|
self.readonly = None
|
|
|
|
@contextmanager
|
|
def connection(self, readonly=False):
|
|
self.readonly = readonly
|
|
yield self.connection_value
|
|
|
|
|
|
def test_previous_seed_result_reuses_legacy_receipt():
|
|
resource_id = str(uuid4())
|
|
store = FakeStore({
|
|
"command_key": "create_work_order",
|
|
"result": {"work_order_id": resource_id},
|
|
})
|
|
seed_request_id = uuid4()
|
|
|
|
result = previous_seed_result(
|
|
store,
|
|
"project-1",
|
|
seed_request_id=seed_request_id,
|
|
command_key="create_work_order",
|
|
)
|
|
|
|
assert result == {"work_order_id": resource_id}
|
|
assert store.readonly is True
|
|
assert store.connection_value.calls[0][1] == (
|
|
"project-1",
|
|
str(ACTOR_ID),
|
|
seed_request_id,
|
|
)
|
|
|
|
|
|
def test_restore_seeded_resources_only_restores_allowed_tables():
|
|
store = FakeStore()
|
|
resource_ids = [str(uuid4()), str(uuid4())]
|
|
|
|
restore_seeded_resources(
|
|
store,
|
|
"project-1",
|
|
table="work_orders",
|
|
id_column="work_order_id",
|
|
resource_ids=resource_ids,
|
|
)
|
|
|
|
statement, parameters = store.connection_value.calls[0]
|
|
assert "UPDATE peripheral_sample.work_orders" in statement
|
|
assert "document=document - 'deleted_at' - 'deleted_by'" in statement
|
|
assert "deleted_at=NULL, deleted_by=NULL" in statement
|
|
assert parameters == ("project-1", resource_ids)
|
|
|
|
with pytest.raises(ValueError, match="unsupported seeded resource"):
|
|
restore_seeded_resources(
|
|
store,
|
|
"project-1",
|
|
table="requests",
|
|
id_column="request_id",
|
|
resource_ids=resource_ids,
|
|
)
|