Changeset View
Changeset View
Standalone View
Standalone View
swh/journal/tests/test_inmemory.py
# Copyright (C) 2019-2022 The Software Heritage developers | |||||
# See the AUTHORS file at the top-level directory of this distribution | |||||
# License: GNU General Public License version 3, or any later version | |||||
# See top-level LICENSE file for more information | |||||
import pytest | import pytest | ||||
from swh.journal.writer import model_object_dict_sanitizer | from swh.journal.writer import model_object_dict_sanitizer | ||||
from swh.journal.writer.inmemory import InMemoryJournalWriter | from swh.journal.writer.inmemory import InMemoryJournalWriter | ||||
from swh.model.model import BaseModel | from swh.model.model import BaseModel | ||||
from swh.model.tests.swh_model_data import TEST_OBJECTS | from swh.model.tests.swh_model_data import TEST_OBJECTS | ||||
def test_write_additions_with_test_objects(): | def test_write_additions_with_test_objects(): | ||||
writer = InMemoryJournalWriter[BaseModel]( | writer = InMemoryJournalWriter[BaseModel]( | ||||
value_sanitizer=model_object_dict_sanitizer | value_sanitizer=model_object_dict_sanitizer | ||||
) | ) | ||||
expected = [] | expected = [] | ||||
priv_expected = [] | |||||
for object_type, objects in TEST_OBJECTS.items(): | for object_type, objects in TEST_OBJECTS.items(): | ||||
writer.write_additions(object_type, objects) | writer.write_additions(object_type, objects) | ||||
for object in objects: | for object in objects: | ||||
if object.anonymize(): | |||||
expected.append((object_type, object.anonymize())) | |||||
priv_expected.append((object_type, object)) | |||||
else: | |||||
expected.append((object_type, object)) | expected.append((object_type, object)) | ||||
assert list(writer.privileged_objects) == [] | assert set(priv_expected) == set(writer.privileged_objects) | ||||
assert set(expected) == set(writer.objects) | assert set(expected) == set(writer.objects) | ||||
def test_write_additions_with_privileged_test_objects(): | |||||
writer = InMemoryJournalWriter[BaseModel]( | |||||
value_sanitizer=model_object_dict_sanitizer | |||||
) | |||||
expected = [] | |||||
for object_type, objects in TEST_OBJECTS.items(): | |||||
writer.write_additions(object_type, objects, True) | |||||
for object in objects: | |||||
expected.append((object_type, object)) | |||||
assert list(writer.objects) == [] | |||||
assert set(expected) == set(writer.privileged_objects) | |||||
def test_write_addition_errors_without_unique_key(): | def test_write_addition_errors_without_unique_key(): | ||||
writer = InMemoryJournalWriter[BaseModel]( | writer = InMemoryJournalWriter[BaseModel]( | ||||
value_sanitizer=model_object_dict_sanitizer | value_sanitizer=model_object_dict_sanitizer | ||||
) | ) | ||||
with pytest.raises(NotImplementedError): | with pytest.raises(NotImplementedError): | ||||
writer.write_addition("BaseModel", BaseModel()) | writer.write_addition("BaseModel", BaseModel()) |