Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 224 Lines • ▼ Show 20 Lines | def test_content_add_collision(self, swh_storage): | ||||
| sha256_array[0] += 1 | sha256_array[0] += 1 | ||||
| cont1b['sha256'] = bytes(sha256_array) | cont1b['sha256'] = bytes(sha256_array) | ||||
| with pytest.raises(HashCollision) as cm: | with pytest.raises(HashCollision) as cm: | ||||
| swh_storage.content_add([cont1, cont1b]) | swh_storage.content_add([cont1, cont1b]) | ||||
| assert cm.value.args[0] in ['sha1', 'sha1_git', 'blake2s256'] | assert cm.value.args[0] in ['sha1', 'sha1_git', 'blake2s256'] | ||||
| def test_content_update(self, swh_storage): | |||||
| swh_storage.journal_writer = None # TODO, not supported | |||||
| cont = copy.deepcopy(data.cont) | |||||
| swh_storage.content_add([cont]) | |||||
| # alter the sha1_git for example | |||||
| cont['sha1_git'] = hash_to_bytes( | |||||
| '3a60a5275d0333bf13468e8b3dcab90f4046e654') | |||||
| swh_storage.content_update([cont], keys=['sha1_git']) | |||||
| results = swh_storage.content_get_metadata([cont['sha1']]) | |||||
| del cont['data'] | |||||
| assert results == {cont['sha1']: [cont]} | |||||
| def test_content_add_metadata(self, swh_storage): | def test_content_add_metadata(self, swh_storage): | ||||
| cont = data.cont | cont = data.cont | ||||
| del cont['data'] | del cont['data'] | ||||
| cont['ctime'] = datetime.datetime.now() | cont['ctime'] = datetime.datetime.now() | ||||
| 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, | ||||
| ▲ Show 20 Lines • Show All 3,372 Lines • ▼ Show 20 Lines | |||||
| @pytest.mark.db | @pytest.mark.db | ||||
| class TestPgStorage: | class TestPgStorage: | ||||
| """This class is dedicated for the rare case where the schema needs to | """This class is dedicated for the rare case where the schema needs to | ||||
| be altered dynamically. | be altered dynamically. | ||||
| Otherwise, the tests could be blocking when ran altogether. | Otherwise, the tests could be blocking when ran altogether. | ||||
| """ | """ | ||||
| def test_content_update(self, swh_storage): | |||||
| swh_storage.journal_writer = None # TODO, not supported | |||||
| cont = copy.deepcopy(data.cont) | |||||
| swh_storage.content_add([cont]) | |||||
| # alter the sha1_git for example | |||||
| cont['sha1_git'] = hash_to_bytes( | |||||
| '3a60a5275d0333bf13468e8b3dcab90f4046e654') | |||||
| swh_storage.content_update([cont], keys=['sha1_git']) | |||||
| with db_transaction(swh_storage) as (_, cur): | |||||
| cur.execute('SELECT sha1, sha1_git, sha256, length, status' | |||||
| ' FROM content WHERE sha1 = %s', | |||||
| (cont['sha1'],)) | |||||
| datum = cur.fetchone() | |||||
| assert datum == (cont['sha1'], cont['sha1_git'], cont['sha256'], | |||||
| cont['length'], 'visible') | |||||
| def test_content_update_with_new_cols(self, swh_storage): | def test_content_update_with_new_cols(self, swh_storage): | ||||
| swh_storage.journal_writer = None # TODO, not supported | swh_storage.journal_writer = None # TODO, not supported | ||||
| with db_transaction(swh_storage) as (_, cur): | with db_transaction(swh_storage) as (_, cur): | ||||
| cur.execute("""alter table content | cur.execute("""alter table content | ||||
| add column test text default null, | add column test text default null, | ||||
| add column test2 text default null""") | add column test2 text default null""") | ||||
| ▲ Show 20 Lines • Show All 106 Lines • Show Last 20 Lines | |||||