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 |