Source code for aries_cloudagent.storage.tests.test_basic_storage

import pytest

from aries_cloudagent.storage.error import (
    StorageDuplicateError,
    StorageError,
    StorageNotFoundError,
    StorageSearchError,
)

from aries_cloudagent.storage.record import StorageRecord

from aries_cloudagent.storage.basic import BasicStorage



[docs]@pytest.fixture()
def store():
    yield BasicStorage()




[docs]def test_record(tags={}):
    return StorageRecord(type="TYPE", value="TEST", tags=tags)




[docs]def test_missing_record(tags={}):
    return StorageRecord(type="__MISSING__", value="000000000")




[docs]class TestBasicStorage:

[docs]    @pytest.mark.asyncio
    async def test_add_required(self, store):
        with pytest.raises(StorageError):
            await store.add_record(None)



[docs]    @pytest.mark.asyncio
    async def test_add_id_required(self, store):
        record = test_record()._replace(id=None)
        with pytest.raises(StorageError):
            await store.add_record(record)



[docs]    @pytest.mark.asyncio
    async def test_retrieve_missing(self, store):
        missing = test_missing_record()
        with pytest.raises(StorageNotFoundError):
            await store.get_record(missing.type, missing.id)



[docs]    @pytest.mark.asyncio
    async def test_add_retrieve(self, store):
        record = test_record()
        await store.add_record(record)
        result = await store.get_record(record.type, record.id)
        assert result
        assert result.id == record.id
        assert result.type == record.type
        assert result.value == record.value
        assert result.tags == record.tags

        with pytest.raises(StorageDuplicateError):
            await store.add_record(record)



[docs]    @pytest.mark.asyncio
    async def test_delete(self, store):
        record = test_record()
        await store.add_record(record)
        await store.delete_record(record)
        with pytest.raises(StorageNotFoundError):
            await store.get_record(record.type, record.id)



[docs]    @pytest.mark.asyncio
    async def test_delete_missing(self, store):
        missing = test_missing_record()
        with pytest.raises(StorageNotFoundError):
            await store.delete_record(missing)



[docs]    @pytest.mark.asyncio
    async def test_update_value(self, store):
        init_value = "a"
        upd_value = "b"
        record = test_record()._replace(value=init_value)
        await store.add_record(record)
        assert record.value == init_value
        await store.update_record_value(record, upd_value)
        result = await store.get_record(record.type, record.id)
        assert result.value == upd_value



[docs]    @pytest.mark.asyncio
    async def test_update_missing(self, store):
        missing = test_missing_record()
        with pytest.raises(StorageNotFoundError):
            await store.update_record_value(missing, missing.value)



[docs]    @pytest.mark.asyncio
    async def test_update_tags(self, store):
        record = test_record({})
        assert record.tags == {}
        await store.add_record(record)
        await store.update_record_tags(record, {"a": "A"})
        result = await store.get_record(record.type, record.id)
        assert result.tags.get("a") == "A"



[docs]    @pytest.mark.asyncio
    async def test_update_tags_missing(self, store):
        missing = test_missing_record()
        with pytest.raises(StorageNotFoundError):
            await store.update_record_tags(missing, {})



[docs]    @pytest.mark.asyncio
    async def test_delete_tags(self, store):
        record = test_record({"a": "A"})
        await store.add_record(record)
        await store.delete_record_tags(record, {"a": "A"})
        result = await store.get_record(record.type, record.id)
        assert result.tags.get("a") is None



[docs]    @pytest.mark.asyncio
    async def test_delete_tags_missing(self, store):
        missing = test_missing_record()
        with pytest.raises(StorageNotFoundError):
            await store.delete_record_tags(missing, {"a": "A"})