Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 3,641 Lines • ▼ Show 20 Lines | |||||
| @pytest.mark.db | @pytest.mark.db | ||||
| class TestLocalStorage: | class TestLocalStorage: | ||||
| """Test the local storage""" | """Test the local storage""" | ||||
| # This test is only relevant on the local storage, with an actual | # This test is only relevant on the local storage, with an actual | ||||
| # objstorage raising an exception | # objstorage raising an exception | ||||
| def test_content_add_objstorage_exception(self, swh_storage): | def test_content_add_objstorage_exception(self, swh_storage): | ||||
| swh_storage.objstorage.add = Mock( | swh_storage.objstorage.content_add = Mock( | ||||
| side_effect=Exception('mocked broken objstorage') | side_effect=Exception('mocked broken objstorage') | ||||
| ) | ) | ||||
| with pytest.raises(Exception) as e: | with pytest.raises(Exception) as e: | ||||
| swh_storage.content_add([data.cont]) | swh_storage.content_add([data.cont]) | ||||
| assert e.value.args == ('mocked broken objstorage',) | assert e.value.args == ('mocked broken objstorage',) | ||||
| missing = list(swh_storage.content_missing([data.cont])) | missing = list(swh_storage.content_missing([data.cont])) | ||||
| ▲ Show 20 Lines • Show All 82 Lines • ▼ Show 20 Lines | def test_content_add_db(self, swh_storage): | ||||
| actual_result = swh_storage.content_add([cont]) | actual_result = swh_storage.content_add([cont]) | ||||
| assert actual_result == { | assert actual_result == { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:add:bytes': cont['length'], | 'content:add:bytes': cont['length'], | ||||
| } | } | ||||
| if hasattr(swh_storage, 'objstorage'): | if hasattr(swh_storage, 'objstorage'): | ||||
| assert cont['sha1'] in swh_storage.objstorage | assert cont['sha1'] in swh_storage.objstorage.objstorage | ||||
| with db_transaction(swh_storage) as (_, cur): | with db_transaction(swh_storage) as (_, cur): | ||||
| cur.execute('SELECT sha1, sha1_git, sha256, length, status' | cur.execute('SELECT sha1, sha1_git, sha256, length, status' | ||||
| ' FROM content WHERE sha1 = %s', | ' FROM content WHERE sha1 = %s', | ||||
| (cont['sha1'],)) | (cont['sha1'],)) | ||||
| datum = cur.fetchone() | datum = cur.fetchone() | ||||
| assert datum == (cont['sha1'], cont['sha1_git'], cont['sha256'], | assert datum == (cont['sha1'], cont['sha1_git'], cont['sha256'], | ||||
| Show All 13 Lines | def test_content_add_metadata_db(self, swh_storage): | ||||
| actual_result = swh_storage.content_add_metadata([cont]) | actual_result = swh_storage.content_add_metadata([cont]) | ||||
| assert actual_result == { | assert actual_result == { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| } | } | ||||
| if hasattr(swh_storage, 'objstorage'): | if hasattr(swh_storage, 'objstorage'): | ||||
| assert cont['sha1'] not in swh_storage.objstorage | assert cont['sha1'] not in swh_storage.objstorage.objstorage | ||||
| with db_transaction(swh_storage) as (_, cur): | with db_transaction(swh_storage) as (_, cur): | ||||
| cur.execute('SELECT sha1, sha1_git, sha256, length, status' | cur.execute('SELECT sha1, sha1_git, sha256, length, status' | ||||
| ' FROM content WHERE sha1 = %s', | ' FROM content WHERE sha1 = %s', | ||||
| (cont['sha1'],)) | (cont['sha1'],)) | ||||
| datum = cur.fetchone() | datum = cur.fetchone() | ||||
| assert datum == (cont['sha1'], cont['sha1_git'], cont['sha256'], | assert datum == (cont['sha1'], cont['sha1_git'], cont['sha256'], | ||||
| cont['length'], 'visible') | cont['length'], 'visible') | ||||
| Show All 27 Lines | |||||