75 lines
1.9 KiB
Python
75 lines
1.9 KiB
Python
from contextlib import contextmanager
|
|
from uuid import uuid4
|
|
|
|
from scripts.seed_peripheral_sample import (
|
|
ACTOR_ID,
|
|
previous_seed_result,
|
|
restore_seeded_test_data,
|
|
)
|
|
|
|
|
|
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_test_data",
|
|
"result": {"record_id": resource_id},
|
|
})
|
|
seed_request_id = uuid4()
|
|
|
|
result = previous_seed_result(
|
|
store,
|
|
"project-1",
|
|
seed_request_id=seed_request_id,
|
|
command_key="create_test_data",
|
|
)
|
|
|
|
assert result == {"record_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_test_data_clears_soft_delete_fields():
|
|
store = FakeStore()
|
|
resource_ids = [str(uuid4()), str(uuid4())]
|
|
|
|
restore_seeded_test_data(store, "project-1", resource_ids)
|
|
|
|
statement, parameters = store.connection_value.calls[0]
|
|
assert "UPDATE peripheral_sample.test_data_records" 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)
|