Page MenuHomeSoftware Heritage
Differential D3592 Diff 12627 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 152 Lines • ▼ Show 20 Lines def test_types(self, swh_storage_backend_config):
assert expected_signature == actual_signature, meth_name assert expected_signature == actual_signature, meth_name
assert missing_methods == [] assert missing_methods == []
def test_check_config(self, swh_storage): def test_check_config(self, swh_storage):
assert swh_storage.check_config(check_write=True) assert swh_storage.check_config(check_write=True)
assert swh_storage.check_config(check_write=False) assert swh_storage.check_config(check_write=False)
def test_content_add(self, swh_storage, sample_data_model): def test_content_add(self, swh_storage, sample_data):
cont = sample_data_model["content"][0] cont = sample_data["content"][0]
insertion_start_time = now() insertion_start_time = now()
actual_result = swh_storage.content_add([cont]) actual_result = swh_storage.content_add([cont])
insertion_end_time = now() insertion_end_time = now()
assert actual_result == { assert actual_result == {
"content:add": 1, "content:add": 1,
"content:add:bytes": cont.length, "content:add:bytes": cont.length,
Show All 14 Lines def test_content_add(self, swh_storage, sample_data):
for obj in contents: for obj in contents:
assert insertion_start_time <= obj.ctime assert insertion_start_time <= obj.ctime
assert obj.ctime <= insertion_end_time assert obj.ctime <= insertion_end_time
assert obj == expected_cont assert obj == expected_cont
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["content"] == 1 assert swh_storage.stat_counters()["content"] == 1
def test_content_add_from_generator(self, swh_storage, sample_data_model): def test_content_add_from_generator(self, swh_storage, sample_data):
cont = sample_data_model["content"][0] cont = sample_data["content"][0]
def _cnt_gen(): def _cnt_gen():
yield cont yield cont
actual_result = swh_storage.content_add(_cnt_gen()) actual_result = swh_storage.content_add(_cnt_gen())
assert actual_result == { assert actual_result == {
"content:add": 1, "content:add": 1,
"content:add:bytes": cont.length, "content:add:bytes": cont.length,
} }
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["content"] == 1 assert swh_storage.stat_counters()["content"] == 1
def test_content_add_from_lazy_content(self, swh_storage, sample_data_model): def test_content_add_from_lazy_content(self, swh_storage, sample_data):
cont = sample_data_model["content"][0] cont = sample_data["content"][0]
lazy_content = LazyContent.from_dict(cont.to_dict()) lazy_content = LazyContent.from_dict(cont.to_dict())
insertion_start_time = now() insertion_start_time = now()
actual_result = swh_storage.content_add([lazy_content]) actual_result = swh_storage.content_add([lazy_content])
insertion_end_time = now() insertion_end_time = now()
Show All 18 Lines def test_content_add_from_lazy_content(self, swh_storage, sample_data):
for obj in contents: for obj in contents:
assert insertion_start_time <= obj.ctime assert insertion_start_time <= obj.ctime
assert obj.ctime <= insertion_end_time assert obj.ctime <= insertion_end_time
assert attr.evolve(obj, ctime=None).to_dict() == expected_cont.to_dict() assert attr.evolve(obj, ctime=None).to_dict() == expected_cont.to_dict()
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["content"] == 1 assert swh_storage.stat_counters()["content"] == 1
def test_content_get_missing(self, swh_storage, sample_data_model): def test_content_get_missing(self, swh_storage, sample_data):
cont, cont2 = sample_data_model["content"][:2] cont, cont2 = sample_data["content"][:2]
swh_storage.content_add([cont]) swh_storage.content_add([cont])
# Query a single missing content # Query a single missing content
results = list(swh_storage.content_get([cont2.sha1])) results = list(swh_storage.content_get([cont2.sha1]))
assert results == [None] assert results == [None]
# Check content_get does not abort after finding a missing content # Check content_get does not abort after finding a missing content
results = list(swh_storage.content_get([cont.sha1, cont2.sha1])) results = list(swh_storage.content_get([cont.sha1, cont2.sha1]))
assert results == [{"sha1": cont.sha1, "data": cont.data}, None] assert results == [{"sha1": cont.sha1, "data": cont.data}, None]
# Check content_get does not discard found countent when it finds # Check content_get does not discard found countent when it finds
# a missing content. # a missing content.
results = list(swh_storage.content_get([cont2.sha1, cont.sha1])) results = list(swh_storage.content_get([cont2.sha1, cont.sha1]))
assert results == [None, {"sha1": cont.sha1, "data": cont.data}] assert results == [None, {"sha1": cont.sha1, "data": cont.data}]
def test_content_add_different_input(self, swh_storage, sample_data_model): def test_content_add_different_input(self, swh_storage, sample_data):
cont, cont2 = sample_data_model["content"][:2] cont, cont2 = sample_data["content"][:2]
actual_result = swh_storage.content_add([cont, cont2]) actual_result = swh_storage.content_add([cont, cont2])
assert actual_result == { assert actual_result == {
"content:add": 2, "content:add": 2,
"content:add:bytes": cont.length + cont2.length, "content:add:bytes": cont.length + cont2.length,
} }
def test_content_add_twice(self, swh_storage, sample_data_model): def test_content_add_twice(self, swh_storage, sample_data):
cont, cont2 = sample_data_model["content"][:2] cont, cont2 = sample_data["content"][:2]
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,
} }
assert len(swh_storage.journal_writer.journal.objects) == 1 assert len(swh_storage.journal_writer.journal.objects) == 1
actual_result = swh_storage.content_add([cont, cont2]) actual_result = swh_storage.content_add([cont, cont2])
assert actual_result == { assert actual_result == {
"content:add": 1, "content:add": 1,
"content:add:bytes": cont2.length, "content:add:bytes": cont2.length,
} }
assert 2 <= len(swh_storage.journal_writer.journal.objects) <= 3 assert 2 <= len(swh_storage.journal_writer.journal.objects) <= 3
assert len(swh_storage.content_find(cont.to_dict())) == 1 assert len(swh_storage.content_find(cont.to_dict())) == 1
assert len(swh_storage.content_find(cont2.to_dict())) == 1 assert len(swh_storage.content_find(cont2.to_dict())) == 1
def test_content_add_collision(self, swh_storage, sample_data_model): def test_content_add_collision(self, swh_storage, sample_data):
cont1 = sample_data_model["content"][0] cont1 = sample_data["content"][0]
# create (corrupted) content with same sha1{,_git} but != sha256 # create (corrupted) content with same sha1{,_git} but != sha256
sha256_array = bytearray(cont1.sha256) sha256_array = bytearray(cont1.sha256)
sha256_array[0] += 1 sha256_array[0] += 1
cont1b = attr.evolve(cont1, sha256=bytes(sha256_array)) cont1b = attr.evolve(cont1, 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])
Show All 9 Lines def test_content_add_collision(self, swh_storage, sample_data):
content_hex_hashes(cont1.hashes()), content_hex_hashes(cont1.hashes()),
content_hex_hashes(cont1b.hashes()), content_hex_hashes(cont1b.hashes()),
] ]
assert exc.colliding_content_hashes() == [ assert exc.colliding_content_hashes() == [
cont1.hashes(), cont1.hashes(),
cont1b.hashes(), cont1b.hashes(),
] ]
def test_content_add_duplicate(self, swh_storage, sample_data_model): def test_content_add_duplicate(self, swh_storage, sample_data):
cont = sample_data_model["content"][0] cont = sample_data["content"][0]
swh_storage.content_add([cont, cont]) swh_storage.content_add([cont, cont])
assert list(swh_storage.content_get([cont.sha1])) == [ assert list(swh_storage.content_get([cont.sha1])) == [
{"sha1": cont.sha1, "data": cont.data} {"sha1": cont.sha1, "data": cont.data}
] ]
def test_content_update(self, swh_storage, sample_data_model): def test_content_update(self, swh_storage, sample_data):
cont1 = sample_data_model["content"][0] cont1 = sample_data["content"][0]
if hasattr(swh_storage, "journal_writer"): if hasattr(swh_storage, "journal_writer"):
swh_storage.journal_writer.journal = None # TODO, not supported swh_storage.journal_writer.journal = None # TODO, not supported
swh_storage.content_add([cont1]) swh_storage.content_add([cont1])
# alter the sha1_git for example # alter the sha1_git for example
cont1b = attr.evolve( cont1b = attr.evolve(
cont1, sha1_git=hash_to_bytes("3a60a5275d0333bf13468e8b3dcab90f4046e654") cont1, sha1_git=hash_to_bytes("3a60a5275d0333bf13468e8b3dcab90f4046e654")
) )
swh_storage.content_update([cont1b.to_dict()], keys=["sha1_git"]) swh_storage.content_update([cont1b.to_dict()], keys=["sha1_git"])
results = swh_storage.content_get_metadata([cont1.sha1]) results = swh_storage.content_get_metadata([cont1.sha1])
expected_content = attr.evolve(cont1b, data=None).to_dict() expected_content = attr.evolve(cont1b, data=None).to_dict()
del expected_content["ctime"] del expected_content["ctime"]
assert tuple(results[cont1.sha1]) == (expected_content,) assert tuple(results[cont1.sha1]) == (expected_content,)
def test_content_add_metadata(self, swh_storage, sample_data_model): def test_content_add_metadata(self, swh_storage, sample_data):
cont = attr.evolve(sample_data_model["content"][0], data=None, ctime=now()) cont = attr.evolve(sample_data["content"][0], data=None, ctime=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,
} }
expected_cont = cont.to_dict() expected_cont = cont.to_dict()
del expected_cont["ctime"] del expected_cont["ctime"]
assert tuple(swh_storage.content_get_metadata([cont.sha1])[cont.sha1]) == ( assert tuple(swh_storage.content_get_metadata([cont.sha1])[cont.sha1]) == (
expected_cont, expected_cont,
) )
contents = [ contents = [
obj obj
for (obj_type, obj) in swh_storage.journal_writer.journal.objects for (obj_type, obj) in swh_storage.journal_writer.journal.objects
if obj_type == "content" if obj_type == "content"
] ]
assert len(contents) == 1 assert len(contents) == 1
for obj in contents: for obj in contents:
obj = attr.evolve(obj, ctime=None) obj = attr.evolve(obj, ctime=None)
assert obj == cont assert obj == cont
def test_content_add_metadata_different_input(self, swh_storage, sample_data_model): def test_content_add_metadata_different_input(self, swh_storage, sample_data):
contents = sample_data_model["content"][:2] contents = sample_data["content"][:2]
cont = attr.evolve(contents[0], data=None, ctime=now()) cont = attr.evolve(contents[0], data=None, ctime=now())
cont2 = attr.evolve(contents[1], data=None, ctime=now()) cont2 = attr.evolve(contents[1], data=None, ctime=now())
actual_result = swh_storage.content_add_metadata([cont, cont2]) actual_result = swh_storage.content_add_metadata([cont, cont2])
assert actual_result == { assert actual_result == {
"content:add": 2, "content:add": 2,
} }
def test_content_add_metadata_collision(self, swh_storage, sample_data_model): def test_content_add_metadata_collision(self, swh_storage, sample_data):
cont1 = attr.evolve(sample_data_model["content"][0], data=None, ctime=now()) cont1 = attr.evolve(sample_data["content"][0], data=None, ctime=now())
# create (corrupted) content with same sha1{,_git} but != sha256 # create (corrupted) content with same sha1{,_git} but != sha256
sha1_git_array = bytearray(cont1.sha256) sha1_git_array = bytearray(cont1.sha256)
sha1_git_array[0] += 1 sha1_git_array[0] += 1
cont1b = attr.evolve(cont1, sha256=bytes(sha1_git_array)) cont1b = attr.evolve(cont1, sha256=bytes(sha1_git_array))
with pytest.raises(HashCollision) as cm: with pytest.raises(HashCollision) as cm:
swh_storage.content_add_metadata([cont1, cont1b]) swh_storage.content_add_metadata([cont1, cont1b])
Show All 9 Lines def test_content_add_metadata_collision(self, swh_storage, sample_data):
content_hex_hashes(cont1.hashes()), content_hex_hashes(cont1.hashes()),
content_hex_hashes(cont1b.hashes()), content_hex_hashes(cont1b.hashes()),
] ]
assert exc.colliding_content_hashes() == [ assert exc.colliding_content_hashes() == [
cont1.hashes(), cont1.hashes(),
cont1b.hashes(), cont1b.hashes(),
] ]
def test_skipped_content_add(self, swh_storage, sample_data_model): def test_skipped_content_add(self, swh_storage, sample_data):
contents = sample_data_model["skipped_content"][:2] contents = sample_data["skipped_content"][:2]
cont = contents[0] cont = contents[0]
cont2 = attr.evolve(contents[1], blake2s256=None) cont2 = attr.evolve(contents[1], blake2s256=None)
contents_dict = [c.to_dict() for c in [cont, cont2]] contents_dict = [c.to_dict() for c in [cont, cont2]]
missing = list(swh_storage.skipped_content_missing(contents_dict)) missing = list(swh_storage.skipped_content_missing(contents_dict))
assert missing == [cont.hashes(), cont2.hashes()] assert missing == [cont.hashes(), cont2.hashes()]
actual_result = swh_storage.skipped_content_add([cont, cont, cont2]) actual_result = swh_storage.skipped_content_add([cont, cont, cont2])
assert 2 <= actual_result.pop("skipped_content:add") <= 3 assert 2 <= actual_result.pop("skipped_content:add") <= 3
assert actual_result == {} assert actual_result == {}
missing = list(swh_storage.skipped_content_missing(contents_dict)) missing = list(swh_storage.skipped_content_missing(contents_dict))
assert missing == [] assert missing == []
def test_skipped_content_add_missing_hashes(self, swh_storage, sample_data_model): def test_skipped_content_add_missing_hashes(self, swh_storage, sample_data):
cont, cont2 = [ cont, cont2 = [
attr.evolve(c, sha1_git=None) attr.evolve(c, sha1_git=None) for c in sample_data["skipped_content"][:2]
for c in sample_data_model["skipped_content"][:2]
] ]
contents_dict = [c.to_dict() for c in [cont, cont2]] contents_dict = [c.to_dict() for c in [cont, cont2]]
missing = list(swh_storage.skipped_content_missing(contents_dict)) missing = list(swh_storage.skipped_content_missing(contents_dict))
assert len(missing) == 2 assert len(missing) == 2
actual_result = swh_storage.skipped_content_add([cont, cont, cont2]) actual_result = swh_storage.skipped_content_add([cont, cont, cont2])
assert 2 <= actual_result.pop("skipped_content:add") <= 3 assert 2 <= actual_result.pop("skipped_content:add") <= 3
assert actual_result == {} assert actual_result == {}
missing = list(swh_storage.skipped_content_missing(contents_dict)) missing = list(swh_storage.skipped_content_missing(contents_dict))
assert missing == [] assert missing == []
def test_skipped_content_missing_partial_hash(self, swh_storage, sample_data_model): def test_skipped_content_missing_partial_hash(self, swh_storage, sample_data):
cont = sample_data_model["skipped_content"][0] cont = sample_data["skipped_content"][0]
cont2 = attr.evolve(cont, sha1_git=None) cont2 = attr.evolve(cont, sha1_git=None)
contents_dict = [c.to_dict() for c in [cont, cont2]] contents_dict = [c.to_dict() for c in [cont, cont2]]
missing = list(swh_storage.skipped_content_missing(contents_dict)) missing = list(swh_storage.skipped_content_missing(contents_dict))
assert len(missing) == 2 assert len(missing) == 2
actual_result = swh_storage.skipped_content_add([cont]) actual_result = swh_storage.skipped_content_add([cont])
▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines def test_content_missing_unknown_algo(self, swh_storage, algos):
missing_per_hash["sha1"] missing_per_hash["sha1"]
) )
for hash in algos: for hash in algos:
assert set( assert set(
swh_storage.content_missing(test_contents, key_hash=hash) swh_storage.content_missing(test_contents, key_hash=hash)
) == set(missing_per_hash[hash]) ) == set(missing_per_hash[hash])
def test_content_missing_per_sha1(self, swh_storage, sample_data_model): def test_content_missing_per_sha1(self, swh_storage, sample_data):
# given # given
cont = sample_data_model["content"][0] cont = sample_data["content"][0]
missing_cont = sample_data_model["skipped_content"][0] missing_cont = sample_data["skipped_content"][0]
swh_storage.content_add([cont]) swh_storage.content_add([cont])
# when # when
gen = swh_storage.content_missing_per_sha1([cont.sha1, missing_cont.sha1]) gen = swh_storage.content_missing_per_sha1([cont.sha1, missing_cont.sha1])
# then # then
assert list(gen) == [missing_cont.sha1] assert list(gen) == [missing_cont.sha1]
def test_content_missing_per_sha1_git(self, swh_storage, sample_data_model): def test_content_missing_per_sha1_git(self, swh_storage, sample_data):
cont, cont2 = sample_data_model["content"][:2] cont, cont2 = sample_data["content"][:2]
missing_cont = sample_data_model["skipped_content"][0] missing_cont = sample_data["skipped_content"][0]
swh_storage.content_add([cont, cont2]) swh_storage.content_add([cont, cont2])
contents = [cont.sha1_git, cont2.sha1_git, missing_cont.sha1_git] contents = [cont.sha1_git, cont2.sha1_git, missing_cont.sha1_git]
missing_contents = swh_storage.content_missing_per_sha1_git(contents) missing_contents = swh_storage.content_missing_per_sha1_git(contents)
assert list(missing_contents) == [missing_cont.sha1_git] assert list(missing_contents) == [missing_cont.sha1_git]
▲ Show 20 Lines • Show All 67 Lines • ▼ Show 20 Lines def test_generate_content_get_partition_pagination(self, swh_storage, swh_contents):
actual_contents.extend(actual_result["contents"]) actual_contents.extend(actual_result["contents"])
page_token = actual_result["next_page_token"] page_token = actual_result["next_page_token"]
if page_token is None: if page_token is None:
break break
assert_contents_ok(expected_contents, actual_contents, ["sha1"]) assert_contents_ok(expected_contents, actual_contents, ["sha1"])
def test_content_get_metadata(self, swh_storage, sample_data_model): def test_content_get_metadata(self, swh_storage, sample_data):
cont1, cont2 = sample_data_model["content"][:2] cont1, cont2 = sample_data["content"][:2]
swh_storage.content_add([cont1, cont2]) swh_storage.content_add([cont1, cont2])
actual_md = swh_storage.content_get_metadata([cont1.sha1, cont2.sha1]) actual_md = swh_storage.content_get_metadata([cont1.sha1, cont2.sha1])
# we only retrieve the metadata so no data nor ctime within # we only retrieve the metadata so no data nor ctime within
expected_cont1, expected_cont2 = [ expected_cont1, expected_cont2 = [
attr.evolve(c, data=None).to_dict() for c in [cont1, cont2] attr.evolve(c, data=None).to_dict() for c in [cont1, cont2]
] ]
expected_cont1.pop("ctime") expected_cont1.pop("ctime")
expected_cont2.pop("ctime") expected_cont2.pop("ctime")
assert tuple(actual_md[cont1.sha1]) == (expected_cont1,) assert tuple(actual_md[cont1.sha1]) == (expected_cont1,)
assert tuple(actual_md[cont2.sha1]) == (expected_cont2,) assert tuple(actual_md[cont2.sha1]) == (expected_cont2,)
assert len(actual_md.keys()) == 2 assert len(actual_md.keys()) == 2
def test_content_get_metadata_missing_sha1(self, swh_storage, sample_data_model): def test_content_get_metadata_missing_sha1(self, swh_storage, sample_data):
cont1, cont2 = sample_data_model["content"][:2] cont1, cont2 = sample_data["content"][:2]
missing_cont = sample_data_model["skipped_content"][0] missing_cont = sample_data["skipped_content"][0]
swh_storage.content_add([cont1, cont2]) swh_storage.content_add([cont1, cont2])
actual_contents = swh_storage.content_get_metadata([missing_cont.sha1]) actual_contents = swh_storage.content_get_metadata([missing_cont.sha1])
assert len(actual_contents) == 1 assert len(actual_contents) == 1
assert tuple(actual_contents[missing_cont.sha1]) == () assert tuple(actual_contents[missing_cont.sha1]) == ()
def test_content_get_random(self, swh_storage, sample_data_model): def test_content_get_random(self, swh_storage, sample_data):
cont, cont2, cont3 = sample_data_model["content"][:3] cont, cont2, cont3 = sample_data["content"][:3]
swh_storage.content_add([cont, cont2, cont3]) swh_storage.content_add([cont, cont2, cont3])
assert swh_storage.content_get_random() in { assert swh_storage.content_get_random() in {
cont.sha1_git, cont.sha1_git,
cont2.sha1_git, cont2.sha1_git,
cont3.sha1_git, cont3.sha1_git,
} }
def test_directory_add(self, swh_storage, sample_data_model): def test_directory_add(self, swh_storage, sample_data):
directory = sample_data_model["directory"][1] directory = sample_data["directory"][1]
init_missing = list(swh_storage.directory_missing([directory.id])) init_missing = list(swh_storage.directory_missing([directory.id]))
assert [directory.id] == init_missing assert [directory.id] == init_missing
actual_result = swh_storage.directory_add([directory]) actual_result = swh_storage.directory_add([directory])
assert actual_result == {"directory:add": 1} assert actual_result == {"directory:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("directory", directory) ("directory", directory)
] ]
actual_data = list(swh_storage.directory_ls(directory.id)) actual_data = list(swh_storage.directory_ls(directory.id))
expected_data = list(transform_entries(directory)) expected_data = list(transform_entries(directory))
assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir) assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir)
after_missing = list(swh_storage.directory_missing([directory.id])) after_missing = list(swh_storage.directory_missing([directory.id]))
assert after_missing == [] assert after_missing == []
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["directory"] == 1 assert swh_storage.stat_counters()["directory"] == 1
def test_directory_add_from_generator(self, swh_storage, sample_data_model): def test_directory_add_from_generator(self, swh_storage, sample_data):
directory = sample_data_model["directory"][1] directory = sample_data["directory"][1]
def _dir_gen(): def _dir_gen():
yield directory yield directory
actual_result = swh_storage.directory_add(directories=_dir_gen()) actual_result = swh_storage.directory_add(directories=_dir_gen())
assert actual_result == {"directory:add": 1} assert actual_result == {"directory:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("directory", directory) ("directory", directory)
] ]
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["directory"] == 1 assert swh_storage.stat_counters()["directory"] == 1
def test_directory_add_twice(self, swh_storage, sample_data_model): def test_directory_add_twice(self, swh_storage, sample_data):
directory = sample_data_model["directory"][1] directory = sample_data["directory"][1]
actual_result = swh_storage.directory_add([directory]) actual_result = swh_storage.directory_add([directory])
assert actual_result == {"directory:add": 1} assert actual_result == {"directory:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("directory", directory) ("directory", directory)
] ]
actual_result = swh_storage.directory_add([directory]) actual_result = swh_storage.directory_add([directory])
assert actual_result == {"directory:add": 0} assert actual_result == {"directory:add": 0}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("directory", directory) ("directory", directory)
] ]
def test_directory_get_recursive(self, swh_storage, sample_data_model): def test_directory_get_recursive(self, swh_storage, sample_data):
dir1, dir2, dir3 = sample_data_model["directory"][:3] dir1, dir2, dir3 = sample_data["directory"][:3]
init_missing = list(swh_storage.directory_missing([dir1.id])) init_missing = list(swh_storage.directory_missing([dir1.id]))
assert init_missing == [dir1.id] assert init_missing == [dir1.id]
actual_result = swh_storage.directory_add([dir1, dir2, dir3]) actual_result = swh_storage.directory_add([dir1, dir2, dir3])
assert actual_result == {"directory:add": 3} assert actual_result == {"directory:add": 3}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
Show All 17 Lines def test_directory_get_recursive(self, swh_storage, sample_data):
actual_data = list(swh_storage.directory_ls(dir3.id, recursive=True)) actual_data = list(swh_storage.directory_ls(dir3.id, recursive=True))
expected_data = list( expected_data = list(
itertools.chain( itertools.chain(
transform_entries(dir3), transform_entries(dir2, prefix=b"subdir/"), transform_entries(dir3), transform_entries(dir2, prefix=b"subdir/"),
) )
) )
assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir) assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir)
def test_directory_get_non_recursive(self, swh_storage, sample_data_model): def test_directory_get_non_recursive(self, swh_storage, sample_data):
dir1, dir2, dir3 = sample_data_model["directory"][:3] dir1, dir2, dir3 = sample_data["directory"][:3]
init_missing = list(swh_storage.directory_missing([dir1.id])) init_missing = list(swh_storage.directory_missing([dir1.id]))
assert init_missing == [dir1.id] assert init_missing == [dir1.id]
actual_result = swh_storage.directory_add([dir1, dir2, dir3]) actual_result = swh_storage.directory_add([dir1, dir2, dir3])
assert actual_result == {"directory:add": 3} assert actual_result == {"directory:add": 3}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
Show All 13 Lines def test_directory_get_non_recursive(self, swh_storage, sample_data):
assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir) assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir)
# List directory containing a known subdirectory, entries should # List directory containing a known subdirectory, entries should
# only be those of the parent directory, not of the subdir # only be those of the parent directory, not of the subdir
actual_data = list(swh_storage.directory_ls(dir3.id)) actual_data = list(swh_storage.directory_ls(dir3.id))
expected_data = list(transform_entries(dir3)) expected_data = list(transform_entries(dir3))
assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir) assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir)
def test_directory_entry_get_by_path(self, swh_storage, sample_data_model): def test_directory_entry_get_by_path(self, swh_storage, sample_data):
cont = sample_data_model["content"][0] cont = sample_data["content"][0]
dir1, dir2, dir3, dir4, dir5 = sample_data_model["directory"][:5] dir1, dir2, dir3, dir4, dir5 = sample_data["directory"][:5]
# given # given
init_missing = list(swh_storage.directory_missing([dir3.id])) init_missing = list(swh_storage.directory_missing([dir3.id]))
assert init_missing == [dir3.id] assert init_missing == [dir3.id]
actual_result = swh_storage.directory_add([dir3, dir4]) actual_result = swh_storage.directory_add([dir3, dir4])
assert actual_result == {"directory:add": 2} assert actual_result == {"directory:add": 2}
▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines def test_directory_entry_get_by_path(self, swh_storage, sample_data):
# when (nothing should be found here since `dir` is not persisted.) # when (nothing should be found here since `dir` is not persisted.)
for entry in dir2.entries: for entry in dir2.entries:
actual_entry = swh_storage.directory_entry_get_by_path( actual_entry = swh_storage.directory_entry_get_by_path(
dir2.id, [entry.name] dir2.id, [entry.name]
) )
assert actual_entry is None assert actual_entry is None
def test_directory_get_random(self, swh_storage, sample_data_model): def test_directory_get_random(self, swh_storage, sample_data):
dir1, dir2, dir3 = sample_data_model["directory"][:3] dir1, dir2, dir3 = sample_data["directory"][:3]
swh_storage.directory_add([dir1, dir2, dir3]) swh_storage.directory_add([dir1, dir2, dir3])
assert swh_storage.directory_get_random() in { assert swh_storage.directory_get_random() in {
dir1.id, dir1.id,
dir2.id, dir2.id,
dir3.id, dir3.id,
} }
def test_revision_add(self, swh_storage, sample_data_model): def test_revision_add(self, swh_storage, sample_data):
revision = sample_data_model["revision"][0] revision = sample_data["revision"][0]
init_missing = swh_storage.revision_missing([revision.id]) init_missing = swh_storage.revision_missing([revision.id])
assert list(init_missing) == [revision.id] assert list(init_missing) == [revision.id]
actual_result = swh_storage.revision_add([revision]) actual_result = swh_storage.revision_add([revision])
assert actual_result == {"revision:add": 1} assert actual_result == {"revision:add": 1}
end_missing = swh_storage.revision_missing([revision.id]) end_missing = swh_storage.revision_missing([revision.id])
assert list(end_missing) == [] assert list(end_missing) == []
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("revision", revision) ("revision", revision)
] ]
# already there so nothing added # already there so nothing added
actual_result = swh_storage.revision_add([revision]) actual_result = swh_storage.revision_add([revision])
assert actual_result == {"revision:add": 0} assert actual_result == {"revision:add": 0}
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["revision"] == 1 assert swh_storage.stat_counters()["revision"] == 1
def test_revision_add_from_generator(self, swh_storage, sample_data_model): def test_revision_add_from_generator(self, swh_storage, sample_data):
revision = sample_data_model["revision"][0] revision = sample_data["revision"][0]
def _rev_gen(): def _rev_gen():
yield revision yield revision
actual_result = swh_storage.revision_add(_rev_gen()) actual_result = swh_storage.revision_add(_rev_gen())
assert actual_result == {"revision:add": 1} assert actual_result == {"revision:add": 1}
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["revision"] == 1 assert swh_storage.stat_counters()["revision"] == 1
def test_revision_add_twice(self, swh_storage, sample_data_model): def test_revision_add_twice(self, swh_storage, sample_data):
revision, revision2 = sample_data_model["revision"][:2] revision, revision2 = sample_data["revision"][:2]
actual_result = swh_storage.revision_add([revision]) actual_result = swh_storage.revision_add([revision])
assert actual_result == {"revision:add": 1} assert actual_result == {"revision:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("revision", revision) ("revision", revision)
] ]
actual_result = swh_storage.revision_add([revision, revision2]) actual_result = swh_storage.revision_add([revision, revision2])
assert actual_result == {"revision:add": 1} assert actual_result == {"revision:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("revision", revision), ("revision", revision),
("revision", revision2), ("revision", revision2),
] ]
def test_revision_add_name_clash(self, swh_storage, sample_data_model): def test_revision_add_name_clash(self, swh_storage, sample_data):
revision, revision2 = sample_data_model["revision"][:2] revision, revision2 = sample_data["revision"][:2]
revision1 = attr.evolve( revision1 = attr.evolve(
revision, revision,
author=Person( author=Person(
fullname=b"John Doe <john.doe@example.com>", fullname=b"John Doe <john.doe@example.com>",
name=b"John Doe", name=b"John Doe",
email=b"john.doe@example.com", email=b"john.doe@example.com",
), ),
) )
revision2 = attr.evolve( revision2 = attr.evolve(
revision2, revision2,
author=Person( author=Person(
fullname=b"John Doe <john.doe@example.com>", fullname=b"John Doe <john.doe@example.com>",
name=b"John Doe ", name=b"John Doe ",
email=b"john.doe@example.com ", email=b"john.doe@example.com ",
), ),
) )
actual_result = swh_storage.revision_add([revision1, revision2]) actual_result = swh_storage.revision_add([revision1, revision2])
assert actual_result == {"revision:add": 2} assert actual_result == {"revision:add": 2}
def test_revision_get_order(self, swh_storage, sample_data_model): def test_revision_get_order(self, swh_storage, sample_data):
revision, revision2 = sample_data_model["revision"][:2] revision, revision2 = sample_data["revision"][:2]
add_result = swh_storage.revision_add([revision, revision2]) add_result = swh_storage.revision_add([revision, revision2])
assert add_result == {"revision:add": 2} assert add_result == {"revision:add": 2}
# order 1 # order 1
res1 = swh_storage.revision_get([revision.id, revision2.id]) res1 = swh_storage.revision_get([revision.id, revision2.id])
assert [Revision.from_dict(r) for r in res1] == [revision, revision2] assert [Revision.from_dict(r) for r in res1] == [revision, revision2]
# order 2 # order 2
res2 = swh_storage.revision_get([revision2.id, revision.id]) res2 = swh_storage.revision_get([revision2.id, revision.id])
assert [Revision.from_dict(r) for r in res2] == [revision2, revision] assert [Revision.from_dict(r) for r in res2] == [revision2, revision]
def test_revision_log(self, swh_storage, sample_data_model): def test_revision_log(self, swh_storage, sample_data):
revision1, revision2, revision3, revision4 = sample_data_model["revision"][:4] revision1, revision2, revision3, revision4 = sample_data["revision"][:4]
# rev4 -is-child-of-> rev3 -> rev1, (rev2 -> rev1) # rev4 -is-child-of-> rev3 -> rev1, (rev2 -> rev1)
swh_storage.revision_add([revision1, revision2, revision3, revision4]) swh_storage.revision_add([revision1, revision2, revision3, revision4])
# when # when
results = list(swh_storage.revision_log([revision4.id])) results = list(swh_storage.revision_log([revision4.id]))
# for comparison purposes # for comparison purposes
actual_results = [Revision.from_dict(r) for r in results] actual_results = [Revision.from_dict(r) for r in results]
assert len(actual_results) == 4 # rev4 -child-> rev3 -> rev1, (rev2 -> rev1) assert len(actual_results) == 4 # rev4 -child-> rev3 -> rev1, (rev2 -> rev1)
assert actual_results == [revision4, revision3, revision1, revision2] assert actual_results == [revision4, revision3, revision1, revision2]
def test_revision_log_with_limit(self, swh_storage, sample_data_model): def test_revision_log_with_limit(self, swh_storage, sample_data):
revision1, revision2, revision3, revision4 = sample_data_model["revision"][:4] revision1, revision2, revision3, revision4 = sample_data["revision"][:4]
# revision4 -is-child-of-> revision3 # revision4 -is-child-of-> revision3
swh_storage.revision_add([revision3, revision4]) swh_storage.revision_add([revision3, revision4])
results = list(swh_storage.revision_log([revision4.id], 1)) results = list(swh_storage.revision_log([revision4.id], 1))
actual_results = [Revision.from_dict(r) for r in results] actual_results = [Revision.from_dict(r) for r in results]
assert len(actual_results) == 1 assert len(actual_results) == 1
assert actual_results[0] == revision4 assert actual_results[0] == revision4
def test_revision_log_unknown_revision(self, swh_storage, sample_data_model): def test_revision_log_unknown_revision(self, swh_storage, sample_data):
revision = sample_data_model["revision"][0] revision = sample_data["revision"][0]
rev_log = list(swh_storage.revision_log([revision.id])) rev_log = list(swh_storage.revision_log([revision.id]))
assert rev_log == [] assert rev_log == []
def test_revision_shortlog(self, swh_storage, sample_data_model): def test_revision_shortlog(self, swh_storage, sample_data):
revision1, revision2, revision3, revision4 = sample_data_model["revision"][:4] revision1, revision2, revision3, revision4 = sample_data["revision"][:4]
# rev4 -is-child-of-> rev3 -> (rev1, rev2); rev2 -> rev1 # rev4 -is-child-of-> rev3 -> (rev1, rev2); rev2 -> rev1
swh_storage.revision_add([revision1, revision2, revision3, revision4]) swh_storage.revision_add([revision1, revision2, revision3, revision4])
results = list(swh_storage.revision_shortlog([revision4.id])) results = list(swh_storage.revision_shortlog([revision4.id]))
actual_results = [[id, tuple(parents)] for (id, parents) in results] actual_results = [[id, tuple(parents)] for (id, parents) in results]
assert len(actual_results) == 4 assert len(actual_results) == 4
assert actual_results == [ assert actual_results == [
[revision4.id, revision4.parents], [revision4.id, revision4.parents],
[revision3.id, revision3.parents], [revision3.id, revision3.parents],
[revision1.id, revision1.parents], [revision1.id, revision1.parents],
[revision2.id, revision2.parents], [revision2.id, revision2.parents],
] ]
def test_revision_shortlog_with_limit(self, swh_storage, sample_data_model): def test_revision_shortlog_with_limit(self, swh_storage, sample_data):
revision1, revision2, revision3, revision4 = sample_data_model["revision"][:4] revision1, revision2, revision3, revision4 = sample_data["revision"][:4]
# revision4 -is-child-of-> revision3 # revision4 -is-child-of-> revision3
swh_storage.revision_add([revision1, revision2, revision3, revision4]) swh_storage.revision_add([revision1, revision2, revision3, revision4])
results = list(swh_storage.revision_shortlog([revision4.id], 1)) results = list(swh_storage.revision_shortlog([revision4.id], 1))
actual_results = [[id, tuple(parents)] for (id, parents) in results] actual_results = [[id, tuple(parents)] for (id, parents) in results]
assert len(actual_results) == 1 assert len(actual_results) == 1
assert list(actual_results[0]) == [revision4.id, revision4.parents] assert list(actual_results[0]) == [revision4.id, revision4.parents]
def test_revision_get(self, swh_storage, sample_data_model): def test_revision_get(self, swh_storage, sample_data):
revision, revision2 = sample_data_model["revision"][:2] revision, revision2 = sample_data["revision"][:2]
swh_storage.revision_add([revision]) swh_storage.revision_add([revision])
actual_revisions = list(swh_storage.revision_get([revision.id, revision2.id])) actual_revisions = list(swh_storage.revision_get([revision.id, revision2.id]))
assert len(actual_revisions) == 2 assert len(actual_revisions) == 2
assert Revision.from_dict(actual_revisions[0]) == revision assert Revision.from_dict(actual_revisions[0]) == revision
assert actual_revisions[1] is None assert actual_revisions[1] is None
def test_revision_get_no_parents(self, swh_storage, sample_data_model): def test_revision_get_no_parents(self, swh_storage, sample_data):
revision = sample_data_model["revision"][0] revision = sample_data["revision"][0]
swh_storage.revision_add([revision]) swh_storage.revision_add([revision])
get = list(swh_storage.revision_get([revision.id])) get = list(swh_storage.revision_get([revision.id]))
assert len(get) == 1 assert len(get) == 1
assert revision.parents == () assert revision.parents == ()
assert tuple(get[0]["parents"]) == () # no parents on this one assert tuple(get[0]["parents"]) == () # no parents on this one
def test_revision_get_random(self, swh_storage, sample_data_model): def test_revision_get_random(self, swh_storage, sample_data):
revision1, revision2, revision3 = sample_data_model["revision"][:3] revision1, revision2, revision3 = sample_data["revision"][:3]
swh_storage.revision_add([revision1, revision2, revision3]) swh_storage.revision_add([revision1, revision2, revision3])
assert swh_storage.revision_get_random() in { assert swh_storage.revision_get_random() in {
revision1.id, revision1.id,
revision2.id, revision2.id,
revision3.id, revision3.id,
} }
def test_release_add(self, swh_storage, sample_data_model): def test_release_add(self, swh_storage, sample_data):
release, release2 = sample_data_model["release"][:2] release, release2 = sample_data["release"][:2]
init_missing = swh_storage.release_missing([release.id, release2.id]) init_missing = swh_storage.release_missing([release.id, release2.id])
assert list(init_missing) == [release.id, release2.id] assert list(init_missing) == [release.id, release2.id]
actual_result = swh_storage.release_add([release, release2]) actual_result = swh_storage.release_add([release, release2])
assert actual_result == {"release:add": 2} assert actual_result == {"release:add": 2}
end_missing = swh_storage.release_missing([release.id, release2.id]) end_missing = swh_storage.release_missing([release.id, release2.id])
assert list(end_missing) == [] assert list(end_missing) == []
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("release", release), ("release", release),
("release", release2), ("release", release2),
] ]
# already present so nothing added # already present so nothing added
actual_result = swh_storage.release_add([release, release2]) actual_result = swh_storage.release_add([release, release2])
assert actual_result == {"release:add": 0} assert actual_result == {"release:add": 0}
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["release"] == 2 assert swh_storage.stat_counters()["release"] == 2
def test_release_add_from_generator(self, swh_storage, sample_data_model): def test_release_add_from_generator(self, swh_storage, sample_data):
release, release2 = sample_data_model["release"][:2] release, release2 = sample_data["release"][:2]
def _rel_gen(): def _rel_gen():
yield release yield release
yield release2 yield release2
actual_result = swh_storage.release_add(_rel_gen()) actual_result = swh_storage.release_add(_rel_gen())
assert actual_result == {"release:add": 2} assert actual_result == {"release:add": 2}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("release", release), ("release", release),
("release", release2), ("release", release2),
] ]
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["release"] == 2 assert swh_storage.stat_counters()["release"] == 2
def test_release_add_no_author_date(self, swh_storage, sample_data_model): def test_release_add_no_author_date(self, swh_storage, sample_data):
full_release = sample_data_model["release"][0] full_release = sample_data["release"][0]
release = attr.evolve(full_release, author=None, date=None) release = attr.evolve(full_release, author=None, date=None)
actual_result = swh_storage.release_add([release]) actual_result = swh_storage.release_add([release])
assert actual_result == {"release:add": 1} assert actual_result == {"release:add": 1}
end_missing = swh_storage.release_missing([release.id]) end_missing = swh_storage.release_missing([release.id])
assert list(end_missing) == [] assert list(end_missing) == []
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("release", release) ("release", release)
] ]
def test_release_add_twice(self, swh_storage, sample_data_model): def test_release_add_twice(self, swh_storage, sample_data):
release, release2 = sample_data_model["release"][:2] release, release2 = sample_data["release"][:2]
actual_result = swh_storage.release_add([release]) actual_result = swh_storage.release_add([release])
assert actual_result == {"release:add": 1} assert actual_result == {"release:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("release", release) ("release", release)
] ]
actual_result = swh_storage.release_add([release, release2, release, release2]) actual_result = swh_storage.release_add([release, release2, release, release2])
assert actual_result == {"release:add": 1} assert actual_result == {"release:add": 1}
assert set(swh_storage.journal_writer.journal.objects) == set( assert set(swh_storage.journal_writer.journal.objects) == set(
[("release", release), ("release", release2),] [("release", release), ("release", release2),]
) )
def test_release_add_name_clash(self, swh_storage, sample_data_model): def test_release_add_name_clash(self, swh_storage, sample_data):
release, release2 = [ release, release2 = [
attr.evolve( attr.evolve(
c, c,
author=Person( author=Person(
fullname=b"John Doe <john.doe@example.com>", fullname=b"John Doe <john.doe@example.com>",
name=b"John Doe", name=b"John Doe",
email=b"john.doe@example.com", email=b"john.doe@example.com",
), ),
) )
for c in sample_data_model["release"][:2] for c in sample_data["release"][:2]
] ]
actual_result = swh_storage.release_add([release, release2]) actual_result = swh_storage.release_add([release, release2])
assert actual_result == {"release:add": 2} assert actual_result == {"release:add": 2}
def test_release_get(self, swh_storage, sample_data_model): def test_release_get(self, swh_storage, sample_data):
release, release2, release3 = sample_data_model["release"][:3] release, release2, release3 = sample_data["release"][:3]
# given # given
swh_storage.release_add([release, release2]) swh_storage.release_add([release, release2])
# when # when
releases = list(swh_storage.release_get([release.id, release2.id])) releases = list(swh_storage.release_get([release.id, release2.id]))
actual_releases = [Release.from_dict(r) for r in releases] actual_releases = [Release.from_dict(r) for r in releases]
# then # then
assert actual_releases == [release, release2] assert actual_releases == [release, release2]
unknown_releases = list(swh_storage.release_get([release3.id])) unknown_releases = list(swh_storage.release_get([release3.id]))
assert unknown_releases[0] is None assert unknown_releases[0] is None
def test_release_get_order(self, swh_storage, sample_data_model): def test_release_get_order(self, swh_storage, sample_data):
release, release2 = sample_data_model["release"][:2] release, release2 = sample_data["release"][:2]
add_result = swh_storage.release_add([release, release2]) add_result = swh_storage.release_add([release, release2])
assert add_result == {"release:add": 2} assert add_result == {"release:add": 2}
# order 1 # order 1
res1 = swh_storage.release_get([release.id, release2.id]) res1 = swh_storage.release_get([release.id, release2.id])
assert list(res1) == [release.to_dict(), release2.to_dict()] assert list(res1) == [release.to_dict(), release2.to_dict()]
# order 2 # order 2
res2 = swh_storage.release_get([release2.id, release.id]) res2 = swh_storage.release_get([release2.id, release.id])
assert list(res2) == [release2.to_dict(), release.to_dict()] assert list(res2) == [release2.to_dict(), release.to_dict()]
def test_release_get_random(self, swh_storage, sample_data_model): def test_release_get_random(self, swh_storage, sample_data):
release, release2, release3 = sample_data_model["release"][:3] release, release2, release3 = sample_data["release"][:3]
swh_storage.release_add([release, release2, release3]) swh_storage.release_add([release, release2, release3])
assert swh_storage.release_get_random() in { assert swh_storage.release_get_random() in {
release.id, release.id,
release2.id, release2.id,
release3.id, release3.id,
} }
def test_origin_add(self, swh_storage, sample_data_model): def test_origin_add(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]]
assert swh_storage.origin_get([origin_dict])[0] is None assert swh_storage.origin_get([origin_dict])[0] is None
stats = swh_storage.origin_add([origin, origin2]) stats = swh_storage.origin_add([origin, origin2])
assert stats == {"origin:add": 2} assert stats == {"origin:add": 2}
actual_origin = swh_storage.origin_get([origin_dict])[0] actual_origin = swh_storage.origin_get([origin_dict])[0]
assert actual_origin["url"] == origin.url assert actual_origin["url"] == origin.url
actual_origin2 = swh_storage.origin_get([origin2_dict])[0] actual_origin2 = swh_storage.origin_get([origin2_dict])[0]
assert actual_origin2["url"] == origin2.url assert actual_origin2["url"] == origin2.url
assert set(swh_storage.journal_writer.journal.objects) == set( assert set(swh_storage.journal_writer.journal.objects) == set(
[("origin", origin), ("origin", origin2),] [("origin", origin), ("origin", origin2),]
) )
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["origin"] == 2 assert swh_storage.stat_counters()["origin"] == 2
def test_origin_add_from_generator(self, swh_storage, sample_data_model): def test_origin_add_from_generator(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]]
def _ori_gen(): def _ori_gen():
yield origin yield origin
yield origin2 yield origin2
stats = swh_storage.origin_add(_ori_gen()) stats = swh_storage.origin_add(_ori_gen())
assert stats == {"origin:add": 2} assert stats == {"origin:add": 2}
actual_origin = swh_storage.origin_get([origin_dict])[0] actual_origin = swh_storage.origin_get([origin_dict])[0]
assert actual_origin["url"] == origin.url assert actual_origin["url"] == origin.url
actual_origin2 = swh_storage.origin_get([origin2_dict])[0] actual_origin2 = swh_storage.origin_get([origin2_dict])[0]
assert actual_origin2["url"] == origin2.url assert actual_origin2["url"] == origin2.url
assert set(swh_storage.journal_writer.journal.objects) == set( assert set(swh_storage.journal_writer.journal.objects) == set(
[("origin", origin), ("origin", origin2),] [("origin", origin), ("origin", origin2),]
) )
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["origin"] == 2 assert swh_storage.stat_counters()["origin"] == 2
def test_origin_add_twice(self, swh_storage, sample_data_model): def test_origin_add_twice(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]]
add1 = swh_storage.origin_add([origin, origin2]) add1 = swh_storage.origin_add([origin, origin2])
assert set(swh_storage.journal_writer.journal.objects) == set( assert set(swh_storage.journal_writer.journal.objects) == set(
[("origin", origin), ("origin", origin2),] [("origin", origin), ("origin", origin2),]
) )
assert add1 == {"origin:add": 2} assert add1 == {"origin:add": 2}
add2 = swh_storage.origin_add([origin, origin2]) add2 = swh_storage.origin_add([origin, origin2])
assert set(swh_storage.journal_writer.journal.objects) == set( assert set(swh_storage.journal_writer.journal.objects) == set(
[("origin", origin), ("origin", origin2),] [("origin", origin), ("origin", origin2),]
) )
assert add2 == {"origin:add": 0} assert add2 == {"origin:add": 0}
def test_origin_get_legacy(self, swh_storage, sample_data_model): def test_origin_get_legacy(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]]
assert swh_storage.origin_get(origin_dict) is None assert swh_storage.origin_get(origin_dict) is None
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
actual_origin0 = swh_storage.origin_get(origin_dict) actual_origin0 = swh_storage.origin_get(origin_dict)
assert actual_origin0["url"] == origin.url assert actual_origin0["url"] == origin.url
def test_origin_get(self, swh_storage, sample_data_model): def test_origin_get(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]]
assert swh_storage.origin_get(origin_dict) is None assert swh_storage.origin_get(origin_dict) is None
assert swh_storage.origin_get([origin_dict]) == [None] assert swh_storage.origin_get([origin_dict]) == [None]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
actual_origins = swh_storage.origin_get([origin_dict]) actual_origins = swh_storage.origin_get([origin_dict])
assert len(actual_origins) == 1 assert len(actual_origins) == 1
Show All 19 Lines def _generate_random_visits(self, nb_visits=100, start=0, end=7):
days = random.randint(0, 28) days = random.randint(0, 28)
weeks = random.randint(start, end) weeks = random.randint(start, end)
date_visit = today - timedelta( date_visit = today - timedelta(
weeks=weeks, hours=hours, minutes=minutes, seconds=seconds, days=days weeks=weeks, hours=hours, minutes=minutes, seconds=seconds, days=days
) )
visits.append(date_visit) visits.append(date_visit)
return visits return visits
def test_origin_visit_get_all(self, swh_storage, sample_data_model): def test_origin_visit_get_all(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visits = swh_storage.origin_visit_add( visits = swh_storage.origin_visit_add(
[ [
OriginVisit( OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
), ),
OriginVisit( OriginVisit(
origin=origin.url, date=data.date_visit2, type=data.type_visit2, origin=origin.url, date=data.date_visit2, type=data.type_visit2,
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines def test_origin_visit_get_all(self, swh_storage, sample_data):
origin.url, last_visit=ov3["visit"], order="desc", limit=1 origin.url, last_visit=ov3["visit"], order="desc", limit=1
) )
) )
assert all_visits8 == [ov2] assert all_visits8 == [ov2]
def test_origin_visit_get__unknown_origin(self, swh_storage): def test_origin_visit_get__unknown_origin(self, swh_storage):
assert [] == list(swh_storage.origin_visit_get("foo")) assert [] == list(swh_storage.origin_visit_get("foo"))
def test_origin_visit_get_random(self, swh_storage, sample_data_model): def test_origin_visit_get_random(self, swh_storage, sample_data):
origins = sample_data_model["origin"][:2] origins = sample_data["origin"][:2]
swh_storage.origin_add(origins) swh_storage.origin_add(origins)
# Add some random visits within the selection range # Add some random visits within the selection range
visits = self._generate_random_visits() visits = self._generate_random_visits()
visit_type = "git" visit_type = "git"
# Add visits to those origins # Add visits to those origins
for origin in origins: for origin in origins:
Show All 19 Lines def test_origin_visit_get_random(self, swh_storage, sample_data):
assert stats["origin"] == len(origins) assert stats["origin"] == len(origins)
assert stats["origin_visit"] == len(origins) * len(visits) assert stats["origin_visit"] == len(origins) * len(visits)
random_origin_visit = swh_storage.origin_visit_get_random(visit_type) random_origin_visit = swh_storage.origin_visit_get_random(visit_type)
assert random_origin_visit assert random_origin_visit
assert random_origin_visit["origin"] is not None assert random_origin_visit["origin"] is not None
assert random_origin_visit["origin"] in [o.url for o in origins] assert random_origin_visit["origin"] in [o.url for o in origins]
def test_origin_visit_get_random_nothing_found( def test_origin_visit_get_random_nothing_found(self, swh_storage, sample_data):
self, swh_storage, sample_data_model origins = sample_data["origin"]
):
origins = sample_data_model["origin"]
swh_storage.origin_add(origins) swh_storage.origin_add(origins)
visit_type = "hg" visit_type = "hg"
# Add some visits outside of the random generation selection so nothing # Add some visits outside of the random generation selection so nothing
# will be found by the random selection # will be found by the random selection
visits = self._generate_random_visits(nb_visits=3, start=13, end=24) visits = self._generate_random_visits(nb_visits=3, start=13, end=24)
for origin in origins: for origin in origins:
for date_visit in visits: for date_visit in visits:
visit = swh_storage.origin_visit_add( visit = swh_storage.origin_visit_add(
Show All 9 Lines def test_origin_visit_get_random_nothing_found(self, swh_storage, sample_data):
snapshot=None, snapshot=None,
) )
] ]
) )
random_origin_visit = swh_storage.origin_visit_get_random(visit_type) random_origin_visit = swh_storage.origin_visit_get_random(visit_type)
assert random_origin_visit is None assert random_origin_visit is None
def test_origin_get_by_sha1(self, swh_storage, sample_data_model): def test_origin_get_by_sha1(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
assert swh_storage.origin_get(origin.to_dict()) is None assert swh_storage.origin_get(origin.to_dict()) is None
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)])) origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)]))
assert len(origins) == 1 assert len(origins) == 1
assert origins[0]["url"] == origin.url assert origins[0]["url"] == origin.url
def test_origin_get_by_sha1_not_found(self, swh_storage, sample_data_model): def test_origin_get_by_sha1_not_found(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
assert swh_storage.origin_get(origin.to_dict()) is None assert swh_storage.origin_get(origin.to_dict()) is None
origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)])) origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)]))
assert len(origins) == 1 assert len(origins) == 1
assert origins[0] is None assert origins[0] is None
def test_origin_search_single_result(self, swh_storage, sample_data_model): def test_origin_search_single_result(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
found_origins = list(swh_storage.origin_search(origin.url)) found_origins = list(swh_storage.origin_search(origin.url))
assert len(found_origins) == 0 assert len(found_origins) == 0
found_origins = list(swh_storage.origin_search(origin.url, regexp=True)) found_origins = list(swh_storage.origin_search(origin.url, regexp=True))
assert len(found_origins) == 0 assert len(found_origins) == 0
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
Show All 16 Lines def test_origin_search_single_result(self, swh_storage, sample_data):
assert found_origins[0] == origin2_data assert found_origins[0] == origin2_data
found_origins = list( found_origins = list(
swh_storage.origin_search(f".{origin2.url[1:-1]}.", regexp=True) swh_storage.origin_search(f".{origin2.url[1:-1]}.", regexp=True)
) )
assert len(found_origins) == 1 assert len(found_origins) == 1
assert found_origins[0] == origin2_data assert found_origins[0] == origin2_data
def test_origin_search_no_regexp(self, swh_storage, sample_data_model): def test_origin_search_no_regexp(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dicts = [o.to_dict() for o in [origin, origin2]] origin_dicts = [o.to_dict() for o in [origin, origin2]]
swh_storage.origin_add([origin, origin2]) swh_storage.origin_add([origin, origin2])
# no pagination # no pagination
found_origins = list(swh_storage.origin_search("/")) found_origins = list(swh_storage.origin_search("/"))
assert len(found_origins) == 2 assert len(found_origins) == 2
# offset=0 # offset=0
found_origins0 = list(swh_storage.origin_search("/", offset=0, limit=1)) found_origins0 = list(swh_storage.origin_search("/", offset=0, limit=1))
assert len(found_origins0) == 1 assert len(found_origins0) == 1
assert found_origins0[0] in origin_dicts assert found_origins0[0] in origin_dicts
# offset=1 # offset=1
found_origins1 = list(swh_storage.origin_search("/", offset=1, limit=1)) found_origins1 = list(swh_storage.origin_search("/", offset=1, limit=1))
assert len(found_origins1) == 1 assert len(found_origins1) == 1
assert found_origins1[0] in origin_dicts assert found_origins1[0] in origin_dicts
# check both origins were returned # check both origins were returned
assert found_origins0 != found_origins1 assert found_origins0 != found_origins1
def test_origin_search_regexp_substring(self, swh_storage, sample_data_model): def test_origin_search_regexp_substring(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dicts = [o.to_dict() for o in [origin, origin2]] origin_dicts = [o.to_dict() for o in [origin, origin2]]
swh_storage.origin_add([origin, origin2]) swh_storage.origin_add([origin, origin2])
# no pagination # no pagination
found_origins = list(swh_storage.origin_search("/", regexp=True)) found_origins = list(swh_storage.origin_search("/", regexp=True))
assert len(found_origins) == 2 assert len(found_origins) == 2
Show All 9 Lines def test_origin_search_regexp_substring(self, swh_storage, sample_data):
swh_storage.origin_search("/", offset=1, limit=1, regexp=True) swh_storage.origin_search("/", offset=1, limit=1, regexp=True)
) )
assert len(found_origins1) == 1 assert len(found_origins1) == 1
assert found_origins1[0] in origin_dicts assert found_origins1[0] in origin_dicts
# check both origins were returned # check both origins were returned
assert found_origins0 != found_origins1 assert found_origins0 != found_origins1
def test_origin_search_regexp_fullstring(self, swh_storage, sample_data_model): def test_origin_search_regexp_fullstring(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
origin_dicts = [o.to_dict() for o in [origin, origin2]] origin_dicts = [o.to_dict() for o in [origin, origin2]]
swh_storage.origin_add([origin, origin2]) swh_storage.origin_add([origin, origin2])
# no pagination # no pagination
found_origins = list(swh_storage.origin_search(".*/.*", regexp=True)) found_origins = list(swh_storage.origin_search(".*/.*", regexp=True))
assert len(found_origins) == 2 assert len(found_origins) == 2
Show All 9 Lines def test_origin_search_regexp_fullstring(self, swh_storage, sample_data):
swh_storage.origin_search(".*/.*", offset=1, limit=1, regexp=True) swh_storage.origin_search(".*/.*", offset=1, limit=1, regexp=True)
) )
assert len(found_origins1) == 1 assert len(found_origins1) == 1
assert found_origins1[0] in origin_dicts assert found_origins1[0] in origin_dicts
# check both origins were returned # check both origins were returned
assert found_origins0 != found_origins1 assert found_origins0 != found_origins1
def test_origin_visit_add(self, swh_storage, sample_data_model): def test_origin_visit_add(self, swh_storage, sample_data):
origin1 = sample_data_model["origin"][1] origin1 = sample_data["origin"][1]
swh_storage.origin_add([origin1]) swh_storage.origin_add([origin1])
date_visit = now() date_visit = now()
date_visit2 = date_visit + datetime.timedelta(minutes=1) date_visit2 = date_visit + datetime.timedelta(minutes=1)
date_visit = round_to_milliseconds(date_visit) date_visit = round_to_milliseconds(date_visit)
date_visit2 = round_to_milliseconds(date_visit2) date_visit2 = round_to_milliseconds(date_visit2)
▲ Show 20 Lines • Show All 69 Lines • ▼ Show 20 Lines def test_origin_visit_status_add_validation(self, swh_storage):
snapshot=None, snapshot=None,
) )
with pytest.raises(StorageArgumentException, match="Unknown origin"): with pytest.raises(StorageArgumentException, match="Unknown origin"):
swh_storage.origin_visit_status_add([visit_status1]) swh_storage.origin_visit_status_add([visit_status1])
objects = list(swh_storage.journal_writer.journal.objects) objects = list(swh_storage.journal_writer.journal.objects)
assert not objects assert not objects
def test_origin_visit_status_add(self, swh_storage, sample_data_model): def test_origin_visit_status_add(self, swh_storage, sample_data):
"""Correct origin visit statuses should add a new visit status """Correct origin visit statuses should add a new visit status
""" """
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
origin1 = sample_data_model["origin"][1] origin1 = sample_data["origin"][1]
origin2 = Origin(url="new-origin") origin2 = Origin(url="new-origin")
swh_storage.origin_add([origin1, origin2]) swh_storage.origin_add([origin1, origin2])
ov1, ov2 = swh_storage.origin_visit_add( ov1, ov2 = swh_storage.origin_visit_add(
[ [
OriginVisit( OriginVisit(
origin=origin1.url, date=data.date_visit1, type=data.type_visit1, origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
), ),
▲ Show 20 Lines • Show All 64 Lines • ▼ Show 20 Lines def test_origin_visit_status_add(self, swh_storage, sample_data):
[("origin", o) for o in expected_origins] [("origin", o) for o in expected_origins]
+ [("origin_visit", v) for v in expected_visits] + [("origin_visit", v) for v in expected_visits]
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses] + [("origin_visit_status", ovs) for ovs in expected_visit_statuses]
) )
for obj in expected_objects: for obj in expected_objects:
assert obj in actual_objects assert obj in actual_objects
def test_origin_visit_status_add_twice(self, swh_storage, sample_data_model): def test_origin_visit_status_add_twice(self, swh_storage, sample_data):
"""Correct origin visit statuses should add a new visit status """Correct origin visit statuses should add a new visit status
""" """
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
origin1 = sample_data_model["origin"][1] origin1 = sample_data["origin"][1]
swh_storage.origin_add([origin1]) swh_storage.origin_add([origin1])
ov1 = swh_storage.origin_visit_add( ov1 = swh_storage.origin_visit_add(
[ [
OriginVisit( OriginVisit(
origin=origin1.url, date=data.date_visit1, type=data.type_visit1, origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
), ),
] ]
)[0] )[0]
Show All 37 Lines def test_origin_visit_status_add_twice(self, swh_storage, sample_data):
[("origin", o) for o in expected_origins] [("origin", o) for o in expected_origins]
+ [("origin_visit", v) for v in expected_visits] + [("origin_visit", v) for v in expected_visits]
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses] + [("origin_visit_status", ovs) for ovs in expected_visit_statuses]
) )
for obj in expected_objects: for obj in expected_objects:
assert obj in actual_objects assert obj in actual_objects
def test_origin_visit_find_by_date(self, swh_storage, sample_data_model): def test_origin_visit_find_by_date(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visit1 = OriginVisit( visit1 = OriginVisit(
origin=origin.url, date=data.date_visit2, type=data.type_visit1, origin=origin.url, date=data.date_visit2, type=data.type_visit1,
) )
visit2 = OriginVisit( visit2 = OriginVisit(
origin=origin.url, date=data.date_visit3, type=data.type_visit2, origin=origin.url, date=data.date_visit3, type=data.type_visit2,
) )
visit3 = OriginVisit( visit3 = OriginVisit(
Show All 30 Lines def test_origin_visit_find_by_date(self, swh_storage, sample_data):
# There are two visits at the same date, the latest must be returned # There are two visits at the same date, the latest must be returned
visit = swh_storage.origin_visit_find_by_date(origin.url, data.date_visit2) visit = swh_storage.origin_visit_find_by_date(origin.url, data.date_visit2)
assert visit["visit"] == ov3.visit assert visit["visit"] == ov3.visit
def test_origin_visit_find_by_date__unknown_origin(self, swh_storage): def test_origin_visit_find_by_date__unknown_origin(self, swh_storage):
swh_storage.origin_visit_find_by_date("foo", data.date_visit2) swh_storage.origin_visit_find_by_date("foo", data.date_visit2)
def test_origin_visit_get_by(self, swh_storage, sample_data_model): def test_origin_visit_get_by(self, swh_storage, sample_data):
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
origins = sample_data_model["origin"][:2] origins = sample_data["origin"][:2]
swh_storage.origin_add(origins) swh_storage.origin_add(origins)
origin_url, origin_url2 = [o.url for o in origins] origin_url, origin_url2 = [o.url for o in origins]
visit = OriginVisit( visit = OriginVisit(
origin=origin_url, date=data.date_visit2, type=data.type_visit2, origin=origin_url, date=data.date_visit2, type=data.type_visit2,
) )
origin_visit1 = swh_storage.origin_visit_add([visit])[0] origin_visit1 = swh_storage.origin_visit_add([visit])[0]
▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines def test_origin_visit_get_by(self, swh_storage, sample_data):
) )
# then # then
assert actual_origin_visit1 == expected_origin_visit assert actual_origin_visit1 == expected_origin_visit
def test_origin_visit_get_by__unknown_origin(self, swh_storage): def test_origin_visit_get_by__unknown_origin(self, swh_storage):
assert swh_storage.origin_visit_get_by("foo", 10) is None assert swh_storage.origin_visit_get_by("foo", 10) is None
def test_origin_visit_get_by_no_result(self, swh_storage, sample_data_model): def test_origin_visit_get_by_no_result(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
actual_origin_visit = swh_storage.origin_visit_get_by(origin.url, 999) actual_origin_visit = swh_storage.origin_visit_get_by(origin.url, 999)
assert actual_origin_visit is None assert actual_origin_visit is None
def test_origin_visit_get_latest_none(self, swh_storage, sample_data_model): def test_origin_visit_get_latest_none(self, swh_storage, sample_data):
"""Origin visit get latest on unknown objects should return nothing """Origin visit get latest on unknown objects should return nothing
""" """
# unknown origin so no result # unknown origin so no result
assert swh_storage.origin_visit_get_latest("unknown-origin") is None assert swh_storage.origin_visit_get_latest("unknown-origin") is None
# unknown type # unknown type
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
assert swh_storage.origin_visit_get_latest(origin.url, type="unknown") is None assert swh_storage.origin_visit_get_latest(origin.url, type="unknown") is None
def test_origin_visit_get_latest_filter_type(self, swh_storage, sample_data_model): def test_origin_visit_get_latest_filter_type(self, swh_storage, sample_data):
"""Filtering origin visit get latest with filter type should be ok """Filtering origin visit get latest with filter type should be ok
""" """
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visit1 = OriginVisit( visit1 = OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
visit2 = OriginVisit( visit2 = OriginVisit(
origin=origin.url, date=data.date_visit2, type=data.type_visit2, origin=origin.url, date=data.date_visit2, type=data.type_visit2,
) )
# Add a visit with the same date as the previous one # Add a visit with the same date as the previous one
Show All 26 Lines def test_origin_visit_get_latest_filter_type(self, swh_storage, sample_data):
assert ( assert (
swh_storage.origin_visit_get_latest( swh_storage.origin_visit_get_latest(
origin.url, type=new_type, # no visit matching that type origin.url, type=new_type, # no visit matching that type
) )
is None is None
) )
def test_origin_visit_get_latest(self, swh_storage, sample_data_model): def test_origin_visit_get_latest(self, swh_storage, sample_data):
empty_snapshot, complete_snapshot = sample_data_model["snapshot"][1:3] empty_snapshot, complete_snapshot = sample_data["snapshot"][1:3]
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visit1 = OriginVisit( visit1 = OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
visit2 = OriginVisit( visit2 = OriginVisit(
origin=origin.url, date=data.date_visit2, type=data.type_visit2, origin=origin.url, date=data.date_visit2, type=data.type_visit2,
) )
▲ Show 20 Lines • Show All 128 Lines • ▼ Show 20 Lines def test_origin_visit_get_latest(self, swh_storage, sample_data):
} == swh_storage.origin_visit_get_latest(origin.url) } == swh_storage.origin_visit_get_latest(origin.url)
assert { assert {
**origin_visit3, **origin_visit3,
"snapshot": complete_snapshot.id, "snapshot": complete_snapshot.id,
"status": "ongoing", "status": "ongoing",
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) } == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)
def test_origin_visit_status_get_latest(self, swh_storage, sample_data_model): def test_origin_visit_status_get_latest(self, swh_storage, sample_data):
snapshot = sample_data_model["snapshot"][2] snapshot = sample_data["snapshot"][2]
origin1 = sample_data_model["origin"][0] origin1 = sample_data["origin"][0]
swh_storage.origin_add([origin1]) swh_storage.origin_add([origin1])
# to have some reference visits # to have some reference visits
ov1, ov2 = swh_storage.origin_visit_add( ov1, ov2 = swh_storage.origin_visit_add(
[ [
OriginVisit( OriginVisit(
origin=origin1.url, date=data.date_visit1, type=data.type_visit1, origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines def test_origin_visit_status_get_latest(self, swh_storage, sample_data):
# nothing matches so nothing # nothing matches so nothing
assert actual_origin_visit is None # there is no visit with status full assert actual_origin_visit is None # there is no visit with status full
actual_origin_visit3 = swh_storage.origin_visit_status_get_latest( actual_origin_visit3 = swh_storage.origin_visit_status_get_latest(
origin1.url, ov2.visit, allowed_statuses=["ongoing"] origin1.url, ov2.visit, allowed_statuses=["ongoing"]
) )
assert actual_origin_visit3 == ovs3 assert actual_origin_visit3 == ovs3
def test_person_fullname_unicity(self, swh_storage, sample_data_model): def test_person_fullname_unicity(self, swh_storage, sample_data):
revision, rev2 = sample_data_model["revision"][0:2] revision, rev2 = sample_data["revision"][0:2]
# create a revision with same committer fullname but wo name and email # create a revision with same committer fullname but wo name and email
revision2 = attr.evolve( revision2 = attr.evolve(
rev2, rev2,
committer=Person( committer=Person(
fullname=revision.committer.fullname, name=None, email=None fullname=revision.committer.fullname, name=None, email=None
), ),
) )
swh_storage.revision_add([revision, revision2]) swh_storage.revision_add([revision, revision2])
# when getting added revisions # when getting added revisions
revisions = list(swh_storage.revision_get([revision.id, revision2.id])) revisions = list(swh_storage.revision_get([revision.id, revision2.id]))
# then check committers are the same # then check committers are the same
assert revisions[0]["committer"] == revisions[1]["committer"] assert revisions[0]["committer"] == revisions[1]["committer"]
def test_snapshot_add_get_empty(self, swh_storage, sample_data_model): def test_snapshot_add_get_empty(self, swh_storage, sample_data):
empty_snapshot = sample_data_model["snapshot"][1] empty_snapshot = sample_data["snapshot"][1]
empty_snapshot_dict = empty_snapshot.to_dict() empty_snapshot_dict = empty_snapshot.to_dict()
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
ov1 = swh_storage.origin_visit_add( ov1 = swh_storage.origin_visit_add(
[ [
OriginVisit( OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
] ]
)[0] )[0]
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Lines def test_snapshot_add_get_empty(self, swh_storage, sample_data):
("origin_visit", ov1), ("origin_visit", ov1),
("origin_visit_status", ovs1,), ("origin_visit_status", ovs1,),
("snapshot", empty_snapshot), ("snapshot", empty_snapshot),
("origin_visit_status", ovs2,), ("origin_visit_status", ovs2,),
] ]
for obj in expected_objects: for obj in expected_objects:
assert obj in actual_objects assert obj in actual_objects
def test_snapshot_add_get_complete(self, swh_storage, sample_data_model): def test_snapshot_add_get_complete(self, swh_storage, sample_data):
complete_snapshot = sample_data_model["snapshot"][2] complete_snapshot = sample_data["snapshot"][2]
complete_snapshot_dict = complete_snapshot.to_dict() complete_snapshot_dict = complete_snapshot.to_dict()
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visit = OriginVisit( visit = OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
origin_visit1 = swh_storage.origin_visit_add([visit])[0] origin_visit1 = swh_storage.origin_visit_add([visit])[0]
visit_id = origin_visit1.visit visit_id = origin_visit1.visit
Show All 12 Lines def test_snapshot_add_get_complete(self, swh_storage, sample_data):
assert actual_result == {"snapshot:add": 1} assert actual_result == {"snapshot:add": 1}
by_id = swh_storage.snapshot_get(complete_snapshot.id) by_id = swh_storage.snapshot_get(complete_snapshot.id)
assert by_id == {**complete_snapshot_dict, "next_branch": None} assert by_id == {**complete_snapshot_dict, "next_branch": None}
by_ov = swh_storage.snapshot_get_by_origin_visit(origin.url, visit_id) by_ov = swh_storage.snapshot_get_by_origin_visit(origin.url, visit_id)
assert by_ov == {**complete_snapshot_dict, "next_branch": None} assert by_ov == {**complete_snapshot_dict, "next_branch": None}
def test_snapshot_add_many(self, swh_storage, sample_data_model): def test_snapshot_add_many(self, swh_storage, sample_data):
snapshot, _, complete_snapshot = sample_data_model["snapshot"][:3] snapshot, _, complete_snapshot = sample_data["snapshot"][:3]
actual_result = swh_storage.snapshot_add([snapshot, complete_snapshot]) actual_result = swh_storage.snapshot_add([snapshot, complete_snapshot])
assert actual_result == {"snapshot:add": 2} assert actual_result == {"snapshot:add": 2}
assert swh_storage.snapshot_get(complete_snapshot.id) == { assert swh_storage.snapshot_get(complete_snapshot.id) == {
**complete_snapshot.to_dict(), **complete_snapshot.to_dict(),
"next_branch": None, "next_branch": None,
} }
assert swh_storage.snapshot_get(snapshot.id) == { assert swh_storage.snapshot_get(snapshot.id) == {
**snapshot.to_dict(), **snapshot.to_dict(),
"next_branch": None, "next_branch": None,
} }
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["snapshot"] == 2 assert swh_storage.stat_counters()["snapshot"] == 2
def test_snapshot_add_many_from_generator(self, swh_storage, sample_data_model): def test_snapshot_add_many_from_generator(self, swh_storage, sample_data):
snapshot, _, complete_snapshot = sample_data_model["snapshot"][:3] snapshot, _, complete_snapshot = sample_data["snapshot"][:3]
def _snp_gen(): def _snp_gen():
yield from [snapshot, complete_snapshot] yield from [snapshot, complete_snapshot]
actual_result = swh_storage.snapshot_add(_snp_gen()) actual_result = swh_storage.snapshot_add(_snp_gen())
assert actual_result == {"snapshot:add": 2} assert actual_result == {"snapshot:add": 2}
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
assert swh_storage.stat_counters()["snapshot"] == 2 assert swh_storage.stat_counters()["snapshot"] == 2
def test_snapshot_add_many_incremental(self, swh_storage, sample_data_model): def test_snapshot_add_many_incremental(self, swh_storage, sample_data):
snapshot, _, complete_snapshot = sample_data_model["snapshot"][:3] snapshot, _, complete_snapshot = sample_data["snapshot"][:3]
actual_result = swh_storage.snapshot_add([complete_snapshot]) actual_result = swh_storage.snapshot_add([complete_snapshot])
assert actual_result == {"snapshot:add": 1} assert actual_result == {"snapshot:add": 1}
actual_result2 = swh_storage.snapshot_add([snapshot, complete_snapshot]) actual_result2 = swh_storage.snapshot_add([snapshot, complete_snapshot])
assert actual_result2 == {"snapshot:add": 1} assert actual_result2 == {"snapshot:add": 1}
assert swh_storage.snapshot_get(complete_snapshot.id) == { assert swh_storage.snapshot_get(complete_snapshot.id) == {
**complete_snapshot.to_dict(), **complete_snapshot.to_dict(),
"next_branch": None, "next_branch": None,
} }
assert swh_storage.snapshot_get(snapshot.id) == { assert swh_storage.snapshot_get(snapshot.id) == {
**snapshot.to_dict(), **snapshot.to_dict(),
"next_branch": None, "next_branch": None,
} }
def test_snapshot_add_twice(self, swh_storage, sample_data_model): def test_snapshot_add_twice(self, swh_storage, sample_data):
snapshot, empty_snapshot = sample_data_model["snapshot"][:2] snapshot, empty_snapshot = sample_data["snapshot"][:2]
actual_result = swh_storage.snapshot_add([empty_snapshot]) actual_result = swh_storage.snapshot_add([empty_snapshot])
assert actual_result == {"snapshot:add": 1} assert actual_result == {"snapshot:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("snapshot", empty_snapshot) ("snapshot", empty_snapshot)
] ]
actual_result = swh_storage.snapshot_add([snapshot]) actual_result = swh_storage.snapshot_add([snapshot])
assert actual_result == {"snapshot:add": 1} assert actual_result == {"snapshot:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("snapshot", empty_snapshot), ("snapshot", empty_snapshot),
("snapshot", snapshot), ("snapshot", snapshot),
] ]
def test_snapshot_add_count_branches(self, swh_storage, sample_data_model): def test_snapshot_add_count_branches(self, swh_storage, sample_data):
complete_snapshot = sample_data_model["snapshot"][2] complete_snapshot = sample_data["snapshot"][2]
actual_result = swh_storage.snapshot_add([complete_snapshot]) actual_result = swh_storage.snapshot_add([complete_snapshot])
assert actual_result == {"snapshot:add": 1} assert actual_result == {"snapshot:add": 1}
snp_size = swh_storage.snapshot_count_branches(complete_snapshot.id) snp_size = swh_storage.snapshot_count_branches(complete_snapshot.id)
expected_snp_size = { expected_snp_size = {
"alias": 1, "alias": 1,
"content": 1, "content": 1,
"directory": 2, "directory": 2,
"release": 1, "release": 1,
"revision": 1, "revision": 1,
"snapshot": 1, "snapshot": 1,
None: 1, None: 1,
} }
assert snp_size == expected_snp_size assert snp_size == expected_snp_size
def test_snapshot_add_get_paginated(self, swh_storage, sample_data_model): def test_snapshot_add_get_paginated(self, swh_storage, sample_data):
complete_snapshot = sample_data_model["snapshot"][2] complete_snapshot = sample_data["snapshot"][2]
swh_storage.snapshot_add([complete_snapshot]) swh_storage.snapshot_add([complete_snapshot])
snp_id = complete_snapshot.id snp_id = complete_snapshot.id
branches = complete_snapshot.to_dict()["branches"] branches = complete_snapshot.to_dict()["branches"]
branch_names = list(sorted(branches)) branch_names = list(sorted(branches))
# Test branch_from # Test branch_from
Show All 30 Lines def test_snapshot_add_get_paginated(self, swh_storage, sample_data):
"branches": { "branches": {
name: branches[name] for name in branch_names[dir_idx : dir_idx + 3] name: branches[name] for name in branch_names[dir_idx : dir_idx + 3]
}, },
"next_branch": branch_names[dir_idx + 3], "next_branch": branch_names[dir_idx + 3],
} }
assert snapshot == expected_snapshot assert snapshot == expected_snapshot
def test_snapshot_add_get_filtered(self, swh_storage, sample_data_model): def test_snapshot_add_get_filtered(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
complete_snapshot = sample_data_model["snapshot"][2] complete_snapshot = sample_data["snapshot"][2]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visit = OriginVisit( visit = OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
origin_visit1 = swh_storage.origin_visit_add([visit])[0] origin_visit1 = swh_storage.origin_visit_add([visit])[0]
swh_storage.snapshot_add([complete_snapshot]) swh_storage.snapshot_add([complete_snapshot])
Show All 37 Lines def test_snapshot_add_get_filtered(self, swh_storage, sample_data):
for name, tgt in branches.items() for name, tgt in branches.items()
if tgt and tgt["target_type"] == "alias" if tgt and tgt["target_type"] == "alias"
}, },
"next_branch": None, "next_branch": None,
} }
assert snapshot == expected_snapshot assert snapshot == expected_snapshot
def test_snapshot_add_get_filtered_and_paginated( def test_snapshot_add_get_filtered_and_paginated(self, swh_storage, sample_data):
self, swh_storage, sample_data_model complete_snapshot = sample_data["snapshot"][2]
):
complete_snapshot = sample_data_model["snapshot"][2]
swh_storage.snapshot_add([complete_snapshot]) swh_storage.snapshot_add([complete_snapshot])
snp_id = complete_snapshot.id snp_id = complete_snapshot.id
branches = complete_snapshot.to_dict()["branches"] branches = complete_snapshot.to_dict()["branches"]
branch_names = list(sorted(branches)) branch_names = list(sorted(branches))
# Test branch_from # Test branch_from
▲ Show 20 Lines • Show All 51 Lines • ▼ Show 20 Lines def test_snapshot_add_get_filtered_and_paginated(self, swh_storage, sample_data):
expected_snapshot = { expected_snapshot = {
"id": snp_id, "id": snp_id,
"branches": {branch_names[dir_idx]: branches[branch_names[dir_idx]],}, "branches": {branch_names[dir_idx]: branches[branch_names[dir_idx]],},
"next_branch": b"release", "next_branch": b"release",
} }
assert snapshot == expected_snapshot assert snapshot == expected_snapshot
def test_snapshot_add_get_branch_by_type(self, swh_storage, sample_data_model): def test_snapshot_add_get_branch_by_type(self, swh_storage, sample_data):
complete_snapshot = sample_data_model["snapshot"][2] complete_snapshot = sample_data["snapshot"][2]
snapshot = complete_snapshot.to_dict() snapshot = complete_snapshot.to_dict()
alias1 = b"alias1" alias1 = b"alias1"
alias2 = b"alias2" alias2 = b"alias2"
target1 = random.choice(list(snapshot["branches"].keys())) target1 = random.choice(list(snapshot["branches"].keys()))
target2 = random.choice(list(snapshot["branches"].keys())) target2 = random.choice(list(snapshot["branches"].keys()))
snapshot["branches"][alias2] = { snapshot["branches"][alias2] = {
Show All 14 Lines def test_snapshot_add_get_branch_by_type(self, swh_storage, sample_data):
target_types=["alias"], target_types=["alias"],
branches_from=alias1, branches_from=alias1,
branches_count=1, branches_count=1,
)["branches"] )["branches"]
assert len(branches) == 1 assert len(branches) == 1
assert alias1 in branches assert alias1 in branches
def test_snapshot_add_get(self, swh_storage, sample_data_model): def test_snapshot_add_get(self, swh_storage, sample_data):
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
visit = OriginVisit( visit = OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
origin_visit1 = swh_storage.origin_visit_add([visit])[0] origin_visit1 = swh_storage.origin_visit_add([visit])[0]
visit_id = origin_visit1.visit visit_id = origin_visit1.visit
Show All 16 Lines def test_snapshot_add_get(self, swh_storage, sample_data):
assert by_id == expected_snapshot assert by_id == expected_snapshot
by_ov = swh_storage.snapshot_get_by_origin_visit(origin.url, visit_id) by_ov = swh_storage.snapshot_get_by_origin_visit(origin.url, visit_id)
assert by_ov == expected_snapshot assert by_ov == expected_snapshot
origin_visit_info = swh_storage.origin_visit_get_by(origin.url, visit_id) origin_visit_info = swh_storage.origin_visit_get_by(origin.url, visit_id)
assert origin_visit_info["snapshot"] == snapshot.id assert origin_visit_info["snapshot"] == snapshot.id
def test_snapshot_add_twice__by_origin_visit(self, swh_storage, sample_data_model): def test_snapshot_add_twice__by_origin_visit(self, swh_storage, sample_data):
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
swh_storage.origin_add([origin]) swh_storage.origin_add([origin])
ov1 = swh_storage.origin_visit_add( ov1 = swh_storage.origin_visit_add(
[ [
OriginVisit( OriginVisit(
origin=origin.url, date=data.date_visit1, type=data.type_visit1, origin=origin.url, date=data.date_visit1, type=data.type_visit1,
) )
] ]
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines def test_snapshot_add_twice__by_origin_visit(self, swh_storage, sample_data):
("origin_visit", ov2), ("origin_visit", ov2),
("origin_visit_status", ovs3), ("origin_visit_status", ovs3),
("origin_visit_status", ovs4), ("origin_visit_status", ovs4),
] ]
for obj in expected_objects: for obj in expected_objects:
assert obj in actual_objects assert obj in actual_objects
def test_snapshot_get_random(self, swh_storage, sample_data_model): def test_snapshot_get_random(self, swh_storage, sample_data):
snapshot, empty_snapshot, complete_snapshot = sample_data_model["snapshot"][:3] snapshot, empty_snapshot, complete_snapshot = sample_data["snapshot"][:3]
swh_storage.snapshot_add([snapshot, empty_snapshot, complete_snapshot]) swh_storage.snapshot_add([snapshot, empty_snapshot, complete_snapshot])
assert swh_storage.snapshot_get_random() in { assert swh_storage.snapshot_get_random() in {
snapshot.id, snapshot.id,
empty_snapshot.id, empty_snapshot.id,
complete_snapshot.id, complete_snapshot.id,
} }
def test_snapshot_missing(self, swh_storage, sample_data_model): def test_snapshot_missing(self, swh_storage, sample_data):
snapshot, missing_snapshot = sample_data_model["snapshot"][:2] snapshot, missing_snapshot = sample_data["snapshot"][:2]
snapshots = [snapshot.id, missing_snapshot.id] snapshots = [snapshot.id, missing_snapshot.id]
swh_storage.snapshot_add([snapshot]) swh_storage.snapshot_add([snapshot])
missing_snapshots = swh_storage.snapshot_missing(snapshots) missing_snapshots = swh_storage.snapshot_missing(snapshots)
assert list(missing_snapshots) == [missing_snapshot.id] assert list(missing_snapshots) == [missing_snapshot.id]
def test_stat_counters(self, swh_storage, sample_data_model): def test_stat_counters(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
revision = sample_data_model["revision"][0] revision = sample_data["revision"][0]
release = sample_data_model["release"][0] release = sample_data["release"][0]
directory = sample_data_model["directory"][0] directory = sample_data["directory"][0]
content = sample_data_model["content"][0] content = sample_data["content"][0]
expected_keys = ["content", "directory", "origin", "revision"] expected_keys = ["content", "directory", "origin", "revision"]
# Initially, all counters are 0 # Initially, all counters are 0
swh_storage.refresh_stat_counters() swh_storage.refresh_stat_counters()
counters = swh_storage.stat_counters() counters = swh_storage.stat_counters()
assert set(expected_keys) <= set(counters) assert set(expected_keys) <= set(counters)
▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines def test_stat_counters(self, swh_storage, sample_data):
assert counters["origin"] == 1 assert counters["origin"] == 1
assert counters["origin_visit"] == 1 assert counters["origin_visit"] == 1
assert counters["revision"] == 1 assert counters["revision"] == 1
assert counters["release"] == 1 assert counters["release"] == 1
assert counters["snapshot"] == 1 assert counters["snapshot"] == 1
if "person" in counters: if "person" in counters:
assert counters["person"] == 3 assert counters["person"] == 3
def test_content_find_ctime(self, swh_storage, sample_data_model): def test_content_find_ctime(self, swh_storage, sample_data):
origin_content = sample_data_model["content"][0] origin_content = sample_data["content"][0]
ctime = round_to_milliseconds(now()) ctime = round_to_milliseconds(now())
content = attr.evolve(origin_content, data=None, ctime=ctime) content = attr.evolve(origin_content, data=None, ctime=ctime)
swh_storage.content_add_metadata([content]) swh_storage.content_add_metadata([content])
actually_present = swh_storage.content_find({"sha1": content.sha1}) actually_present = swh_storage.content_find({"sha1": content.sha1})
assert actually_present[0] == content.to_dict() assert actually_present[0] == content.to_dict()
def test_content_find_with_present_content(self, swh_storage, sample_data_model): def test_content_find_with_present_content(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
expected_content = content.to_dict() expected_content = content.to_dict()
del expected_content["data"] del expected_content["data"]
del expected_content["ctime"] del expected_content["ctime"]
# 1. with something to find # 1. with something to find
swh_storage.content_add([content]) swh_storage.content_add([content])
actually_present = swh_storage.content_find({"sha1": content.sha1}) actually_present = swh_storage.content_find({"sha1": content.sha1})
Show All 14 Lines def test_content_find_with_present_content(self, swh_storage, sample_data):
assert actually_present[0] == expected_content assert actually_present[0] == expected_content
# 4. with something to find # 4. with something to find
actually_present = swh_storage.content_find(content.hashes()) actually_present = swh_storage.content_find(content.hashes())
assert 1 == len(actually_present) assert 1 == len(actually_present)
actually_present[0].pop("ctime") actually_present[0].pop("ctime")
assert actually_present[0] == expected_content assert actually_present[0] == expected_content
def test_content_find_with_non_present_content( def test_content_find_with_non_present_content(self, swh_storage, sample_data):
self, swh_storage, sample_data_model missing_content = sample_data["skipped_content"][0]
):
missing_content = sample_data_model["skipped_content"][0]
# 1. with something that does not exist # 1. with something that does not exist
actually_present = swh_storage.content_find({"sha1": missing_content.sha1}) actually_present = swh_storage.content_find({"sha1": missing_content.sha1})
assert actually_present == [] assert actually_present == []
# 2. with something that does not exist # 2. with something that does not exist
actually_present = swh_storage.content_find( actually_present = swh_storage.content_find(
{"sha1_git": missing_content.sha1_git} {"sha1_git": missing_content.sha1_git}
) )
assert actually_present == [] assert actually_present == []
# 3. with something that does not exist # 3. with something that does not exist
actually_present = swh_storage.content_find({"sha256": missing_content.sha256}) actually_present = swh_storage.content_find({"sha256": missing_content.sha256})
assert actually_present == [] assert actually_present == []
def test_content_find_with_duplicate_input(self, swh_storage, sample_data_model): def test_content_find_with_duplicate_input(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
# Create fake data with colliding sha256 and blake2s256 # Create fake data with colliding sha256 and blake2s256
sha1_array = bytearray(content.sha1) sha1_array = bytearray(content.sha1)
sha1_array[0] += 1 sha1_array[0] += 1
sha1git_array = bytearray(content.sha1_git) sha1git_array = bytearray(content.sha1_git)
sha1git_array[0] += 1 sha1git_array[0] += 1
duplicated_content = attr.evolve( duplicated_content = attr.evolve(
content, sha1=bytes(sha1_array), sha1_git=bytes(sha1git_array) content, sha1=bytes(sha1_array), sha1_git=bytes(sha1git_array)
Show All 22 Lines def test_content_find_with_duplicate_input(self, swh_storage, sample_data):
actual_result[1], actual_result[1],
]: ]:
dict_.pop(key, None) dict_.pop(key, None)
expected_result = [expected_content, expected_duplicated_content] expected_result = [expected_content, expected_duplicated_content]
for result in expected_result: for result in expected_result:
assert result in actual_result assert result in actual_result
def test_content_find_with_duplicate_sha256(self, swh_storage, sample_data_model): def test_content_find_with_duplicate_sha256(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
hashes = {} hashes = {}
# Create fake data with colliding sha256 # Create fake data with colliding sha256
for hashalgo in ("sha1", "sha1_git", "blake2s256"): for hashalgo in ("sha1", "sha1_git", "blake2s256"):
value = bytearray(getattr(content, hashalgo)) value = bytearray(getattr(content, hashalgo))
value[0] += 1 value[0] += 1
hashes[hashalgo] = bytes(value) hashes[hashalgo] = bytes(value)
Show All 38 Lines def test_content_find_with_duplicate_sha256(self, swh_storage, sample_data):
) )
) )
assert len(actual_result) == 1 assert len(actual_result) == 1
actual_result[0].pop("ctime") actual_result[0].pop("ctime")
assert actual_result == [expected_duplicated_content] assert actual_result == [expected_duplicated_content]
def test_content_find_with_duplicate_blake2s256( def test_content_find_with_duplicate_blake2s256(self, swh_storage, sample_data):
self, swh_storage, sample_data_model content = sample_data["content"][0]
):
content = sample_data_model["content"][0]
# Create fake data with colliding sha256 and blake2s256 # Create fake data with colliding sha256 and blake2s256
sha1_array = bytearray(content.sha1) sha1_array = bytearray(content.sha1)
sha1_array[0] += 1 sha1_array[0] += 1
sha1git_array = bytearray(content.sha1_git) sha1git_array = bytearray(content.sha1_git)
sha1git_array[0] += 1 sha1git_array[0] += 1
sha256_array = bytearray(content.sha256) sha256_array = bytearray(content.sha256)
sha256_array[0] += 1 sha256_array[0] += 1
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines def test_content_find_bad_input(self, swh_storage):
# 1. with bad input # 1. with bad input
with pytest.raises(StorageArgumentException): with pytest.raises(StorageArgumentException):
swh_storage.content_find({}) # empty is bad swh_storage.content_find({}) # empty is bad
# 2. with bad input # 2. with bad input
with pytest.raises(StorageArgumentException): with pytest.raises(StorageArgumentException):
swh_storage.content_find({"unknown-sha1": "something"}) # not the right key swh_storage.content_find({"unknown-sha1": "something"}) # not the right key
def test_object_find_by_sha1_git(self, swh_storage, sample_data_model): def test_object_find_by_sha1_git(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
directory = sample_data_model["directory"][0] directory = sample_data["directory"][0]
revision = sample_data_model["revision"][0] revision = sample_data["revision"][0]
release = sample_data_model["release"][0] release = sample_data["release"][0]
sha1_gits = [b"00000000000000000000"] sha1_gits = [b"00000000000000000000"]
expected = { expected = {
b"00000000000000000000": [], b"00000000000000000000": [],
} }
swh_storage.content_add([content]) swh_storage.content_add([content])
sha1_gits.append(content.sha1_git) sha1_gits.append(content.sha1_git)
Show All 13 Lines def test_object_find_by_sha1_git(self, swh_storage, sample_data):
swh_storage.release_add([release]) swh_storage.release_add([release])
sha1_gits.append(release.id) sha1_gits.append(release.id)
expected[release.id] = [{"sha1_git": release.id, "type": "release",}] expected[release.id] = [{"sha1_git": release.id, "type": "release",}]
ret = swh_storage.object_find_by_sha1_git(sha1_gits) ret = swh_storage.object_find_by_sha1_git(sha1_gits)
assert expected == ret assert expected == ret
def test_metadata_fetcher_add_get(self, swh_storage, sample_data_model): def test_metadata_fetcher_add_get(self, swh_storage, sample_data):
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
actual_fetcher = swh_storage.metadata_fetcher_get(fetcher.name, fetcher.version) actual_fetcher = swh_storage.metadata_fetcher_get(fetcher.name, fetcher.version)
assert actual_fetcher is None # does not exist assert actual_fetcher is None # does not exist
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
res = swh_storage.metadata_fetcher_get(fetcher.name, fetcher.version) res = swh_storage.metadata_fetcher_get(fetcher.name, fetcher.version)
assert res == fetcher assert res == fetcher
def test_metadata_authority_add_get(self, swh_storage, sample_data_model): def test_metadata_authority_add_get(self, swh_storage, sample_data):
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
actual_authority = swh_storage.metadata_authority_get( actual_authority = swh_storage.metadata_authority_get(
authority.type, authority.url authority.type, authority.url
) )
assert actual_authority is None # does not exist assert actual_authority is None # does not exist
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
res = swh_storage.metadata_authority_get(authority.type, authority.url) res = swh_storage.metadata_authority_get(authority.type, authority.url)
assert res == authority assert res == authority
def test_content_metadata_add(self, swh_storage, sample_data_model): def test_content_metadata_add(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
content_metadata = sample_data_model["content_metadata"][:2] content_metadata = sample_data["content_metadata"][:2]
content_swhid = SWHID( content_swhid = SWHID(
object_type="content", object_id=hash_to_bytes(content.sha1_git) object_type="content", object_id=hash_to_bytes(content.sha1_git)
) )
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add(content_metadata) swh_storage.object_metadata_add(content_metadata)
result = swh_storage.object_metadata_get( result = swh_storage.object_metadata_get(
MetadataTargetType.CONTENT, content_swhid, authority MetadataTargetType.CONTENT, content_swhid, authority
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert list(sorted(result["results"], key=lambda x: x.discovery_date,)) == list( assert list(sorted(result["results"], key=lambda x: x.discovery_date,)) == list(
content_metadata content_metadata
) )
def test_content_metadata_add_duplicate(self, swh_storage, sample_data_model): def test_content_metadata_add_duplicate(self, swh_storage, sample_data):
"""Duplicates should be silently updated.""" """Duplicates should be silently updated."""
content = sample_data_model["content"][0] content = sample_data["content"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
content_metadata, content_metadata2 = sample_data_model["content_metadata"][:2] content_metadata, content_metadata2 = sample_data["content_metadata"][:2]
content_swhid = SWHID( content_swhid = SWHID(
object_type="content", object_id=hash_to_bytes(content.sha1_git) object_type="content", object_id=hash_to_bytes(content.sha1_git)
) )
new_content_metadata2 = attr.evolve( new_content_metadata2 = attr.evolve(
content_metadata2, format="new-format", metadata=b"new-metadata", content_metadata2, format="new-format", metadata=b"new-metadata",
) )
Show All 11 Lines def test_content_metadata_add_duplicate(self, swh_storage, sample_data):
expected_results1 = (content_metadata, new_content_metadata2) expected_results1 = (content_metadata, new_content_metadata2)
expected_results2 = (content_metadata, content_metadata2) expected_results2 = (content_metadata, content_metadata2)
assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in ( assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in (
expected_results1, # cassandra expected_results1, # cassandra
expected_results2, # postgresql expected_results2, # postgresql
) )
def test_content_metadata_get(self, swh_storage, sample_data_model): def test_content_metadata_get(self, swh_storage, sample_data):
content, content2 = sample_data_model["content"][:2] content, content2 = sample_data["content"][:2]
fetcher, fetcher2 = sample_data_model["fetcher"][:2] fetcher, fetcher2 = sample_data["fetcher"][:2]
authority, authority2 = sample_data_model["authority"][:2] authority, authority2 = sample_data["authority"][:2]
content1_metadata1, content1_metadata2, content1_metadata3 = sample_data_model[ content1_metadata1, content1_metadata2, content1_metadata3 = sample_data[
"content_metadata" "content_metadata"
][:3] ][:3]
content1_swhid = SWHID(object_type="content", object_id=content.sha1_git) content1_swhid = SWHID(object_type="content", object_id=content.sha1_git)
content2_swhid = SWHID(object_type="content", object_id=content2.sha1_git) content2_swhid = SWHID(object_type="content", object_id=content2.sha1_git)
content2_metadata = attr.evolve(content1_metadata2, id=content2_swhid) content2_metadata = attr.evolve(content1_metadata2, id=content2_swhid)
swh_storage.metadata_authority_add([authority, authority2]) swh_storage.metadata_authority_add([authority, authority2])
Show All 25 Lines def test_content_metadata_get(self, swh_storage, sample_data):
) )
result = swh_storage.object_metadata_get( result = swh_storage.object_metadata_get(
MetadataTargetType.CONTENT, content2_swhid, authority MetadataTargetType.CONTENT, content2_swhid, authority
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert [content2_metadata] == list(result["results"],) assert [content2_metadata] == list(result["results"],)
def test_content_metadata_get_after(self, swh_storage, sample_data_model): def test_content_metadata_get_after(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
content_metadata, content_metadata2 = sample_data_model["content_metadata"][:2] content_metadata, content_metadata2 = sample_data["content_metadata"][:2]
content_swhid = SWHID(object_type="content", object_id=content.sha1_git) content_swhid = SWHID(object_type="content", object_id=content.sha1_git)
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([content_metadata, content_metadata2]) swh_storage.object_metadata_add([content_metadata, content_metadata2])
Show All 21 Lines def test_content_metadata_get_after(self, swh_storage, sample_data):
MetadataTargetType.CONTENT, MetadataTargetType.CONTENT,
content_swhid, content_swhid,
authority, authority,
after=content_metadata2.discovery_date, after=content_metadata2.discovery_date,
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert result["results"] == [] assert result["results"] == []
def test_content_metadata_get_paginate(self, swh_storage, sample_data_model): def test_content_metadata_get_paginate(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
content_metadata, content_metadata2 = sample_data_model["content_metadata"][:2] content_metadata, content_metadata2 = sample_data["content_metadata"][:2]
content_swhid = SWHID(object_type="content", object_id=content.sha1_git) content_swhid = SWHID(object_type="content", object_id=content.sha1_git)
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([content_metadata, content_metadata2]) swh_storage.object_metadata_add([content_metadata, content_metadata2])
swh_storage.object_metadata_get( swh_storage.object_metadata_get(
MetadataTargetType.CONTENT, content_swhid, authority MetadataTargetType.CONTENT, content_swhid, authority
Show All 10 Lines def test_content_metadata_get_paginate(self, swh_storage, sample_data):
content_swhid, content_swhid,
authority, authority,
limit=1, limit=1,
page_token=result["next_page_token"], page_token=result["next_page_token"],
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert result["results"] == [content_metadata2] assert result["results"] == [content_metadata2]
def test_content_metadata_get_paginate_same_date( def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data):
self, swh_storage, sample_data_model content = sample_data["content"][0]
): fetcher1, fetcher2 = sample_data["fetcher"][:2]
content = sample_data_model["content"][0] authority = sample_data["authority"][0]
fetcher1, fetcher2 = sample_data_model["fetcher"][:2] content_metadata, content_metadata2 = sample_data["content_metadata"][:2]
authority = sample_data_model["authority"][0]
content_metadata, content_metadata2 = sample_data_model["content_metadata"][:2]
content_swhid = SWHID(object_type="content", object_id=content.sha1_git) content_swhid = SWHID(object_type="content", object_id=content.sha1_git)
swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) swh_storage.metadata_fetcher_add([fetcher1, fetcher2])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
new_content_metadata2 = attr.evolve( new_content_metadata2 = attr.evolve(
content_metadata2, content_metadata2,
Show All 14 Lines def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data):
content_swhid, content_swhid,
authority, authority,
limit=1, limit=1,
page_token=result["next_page_token"], page_token=result["next_page_token"],
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert result["results"] == [new_content_metadata2] assert result["results"] == [new_content_metadata2]
def test_content_metadata_get__invalid_id(self, swh_storage, sample_data_model): def test_content_metadata_get__invalid_id(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
content_metadata, content_metadata2 = sample_data_model["content_metadata"][:2] content_metadata, content_metadata2 = sample_data["content_metadata"][:2]
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([content_metadata, content_metadata2]) swh_storage.object_metadata_add([content_metadata, content_metadata2])
with pytest.raises(StorageArgumentException, match="SWHID"): with pytest.raises(StorageArgumentException, match="SWHID"):
swh_storage.object_metadata_get( swh_storage.object_metadata_get(
MetadataTargetType.CONTENT, origin.url, authority MetadataTargetType.CONTENT, origin.url, authority
) )
def test_origin_metadata_add(self, swh_storage, sample_data_model): def test_origin_metadata_add(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([origin_metadata, origin_metadata2]) swh_storage.object_metadata_add([origin_metadata, origin_metadata2])
result = swh_storage.object_metadata_get( result = swh_storage.object_metadata_get(
MetadataTargetType.ORIGIN, origin.url, authority MetadataTargetType.ORIGIN, origin.url, authority
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert list(sorted(result["results"], key=lambda x: x.discovery_date)) == [ assert list(sorted(result["results"], key=lambda x: x.discovery_date)) == [
origin_metadata, origin_metadata,
origin_metadata2, origin_metadata2,
] ]
def test_origin_metadata_add_duplicate(self, swh_storage, sample_data_model): def test_origin_metadata_add_duplicate(self, swh_storage, sample_data):
"""Duplicates should be silently updated.""" """Duplicates should be silently updated."""
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
new_origin_metadata2 = attr.evolve( new_origin_metadata2 = attr.evolve(
origin_metadata2, format="new-format", metadata=b"new-metadata", origin_metadata2, format="new-format", metadata=b"new-metadata",
) )
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
Show All 10 Lines def test_origin_metadata_add_duplicate(self, swh_storage, sample_data):
expected_results1 = (origin_metadata, new_origin_metadata2) expected_results1 = (origin_metadata, new_origin_metadata2)
expected_results2 = (origin_metadata, origin_metadata2) expected_results2 = (origin_metadata, origin_metadata2)
assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in ( assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in (
expected_results1, # cassandra expected_results1, # cassandra
expected_results2, # postgresql expected_results2, # postgresql
) )
def test_origin_metadata_get(self, swh_storage, sample_data_model): def test_origin_metadata_get(self, swh_storage, sample_data):
origin, origin2 = sample_data_model["origin"][:2] origin, origin2 = sample_data["origin"][:2]
fetcher, fetcher2 = sample_data_model["fetcher"][:2] fetcher, fetcher2 = sample_data["fetcher"][:2]
authority, authority2 = sample_data_model["authority"][:2] authority, authority2 = sample_data["authority"][:2]
origin1_metadata1, origin1_metadata2, origin1_metadata3 = sample_data_model[ origin1_metadata1, origin1_metadata2, origin1_metadata3 = sample_data[
"origin_metadata" "origin_metadata"
][:3] ][:3]
assert swh_storage.origin_add([origin, origin2]) == {"origin:add": 2} assert swh_storage.origin_add([origin, origin2]) == {"origin:add": 2}
origin2_metadata = attr.evolve(origin1_metadata2, id=origin2.url) origin2_metadata = attr.evolve(origin1_metadata2, id=origin2.url)
swh_storage.metadata_authority_add([authority, authority2]) swh_storage.metadata_authority_add([authority, authority2])
Show All 20 Lines def test_origin_metadata_get(self, swh_storage, sample_data):
) )
result = swh_storage.object_metadata_get( result = swh_storage.object_metadata_get(
MetadataTargetType.ORIGIN, origin2.url, authority MetadataTargetType.ORIGIN, origin2.url, authority
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert [origin2_metadata] == list(result["results"],) assert [origin2_metadata] == list(result["results"],)
def test_origin_metadata_get_after(self, swh_storage, sample_data_model): def test_origin_metadata_get_after(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([origin_metadata, origin_metadata2]) swh_storage.object_metadata_add([origin_metadata, origin_metadata2])
result = swh_storage.object_metadata_get( result = swh_storage.object_metadata_get(
Show All 21 Lines def test_origin_metadata_get_after(self, swh_storage, sample_data):
MetadataTargetType.ORIGIN, MetadataTargetType.ORIGIN,
origin.url, origin.url,
authority, authority,
after=origin_metadata2.discovery_date, after=origin_metadata2.discovery_date,
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert result["results"] == [] assert result["results"] == []
def test_origin_metadata_get_paginate(self, swh_storage, sample_data_model): def test_origin_metadata_get_paginate(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([origin_metadata, origin_metadata2]) swh_storage.object_metadata_add([origin_metadata, origin_metadata2])
swh_storage.object_metadata_get( swh_storage.object_metadata_get(
Show All 11 Lines def test_origin_metadata_get_paginate(self, swh_storage, sample_data):
origin.url, origin.url,
authority, authority,
limit=1, limit=1,
page_token=result["next_page_token"], page_token=result["next_page_token"],
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert result["results"] == [origin_metadata2] assert result["results"] == [origin_metadata2]
def test_origin_metadata_get_paginate_same_date( def test_origin_metadata_get_paginate_same_date(self, swh_storage, sample_data):
self, swh_storage, sample_data_model origin = sample_data["origin"][0]
): fetcher1, fetcher2 = sample_data["fetcher"][:2]
origin = sample_data_model["origin"][0] authority = sample_data["authority"][0]
fetcher1, fetcher2 = sample_data_model["fetcher"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
authority = sample_data_model["authority"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) swh_storage.metadata_fetcher_add([fetcher1, fetcher2])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
new_origin_metadata2 = attr.evolve( new_origin_metadata2 = attr.evolve(
origin_metadata2, origin_metadata2,
discovery_date=origin_metadata2.discovery_date, discovery_date=origin_metadata2.discovery_date,
Show All 13 Lines def test_origin_metadata_get_paginate_same_date(self, swh_storage, sample_data):
origin.url, origin.url,
authority, authority,
limit=1, limit=1,
page_token=result["next_page_token"], page_token=result["next_page_token"],
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert result["results"] == [new_origin_metadata2] assert result["results"] == [new_origin_metadata2]
def test_origin_metadata_add_missing_authority( def test_origin_metadata_add_missing_authority(self, swh_storage, sample_data):
self, swh_storage, sample_data_model origin = sample_data["origin"][0]
): fetcher = sample_data["fetcher"][0]
origin = sample_data_model["origin"][0] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
fetcher = sample_data_model["fetcher"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
with pytest.raises(StorageArgumentException, match="authority"): with pytest.raises(StorageArgumentException, match="authority"):
swh_storage.object_metadata_add([origin_metadata, origin_metadata2]) swh_storage.object_metadata_add([origin_metadata, origin_metadata2])
def test_origin_metadata_add_missing_fetcher(self, swh_storage, sample_data_model): def test_origin_metadata_add_missing_fetcher(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
with pytest.raises(StorageArgumentException, match="fetcher"): with pytest.raises(StorageArgumentException, match="fetcher"):
swh_storage.object_metadata_add([origin_metadata, origin_metadata2]) swh_storage.object_metadata_add([origin_metadata, origin_metadata2])
def test_origin_metadata_get__invalid_id_type(self, swh_storage, sample_data_model): def test_origin_metadata_get__invalid_id_type(self, swh_storage, sample_data):
origin = sample_data_model["origin"][0] origin = sample_data["origin"][0]
authority = sample_data_model["authority"][0] authority = sample_data["authority"][0]
fetcher = sample_data_model["fetcher"][0] fetcher = sample_data["fetcher"][0]
origin_metadata, origin_metadata2 = sample_data_model["origin_metadata"][:2] origin_metadata, origin_metadata2 = sample_data["origin_metadata"][:2]
content_metadata = sample_data_model["content_metadata"][0] content_metadata = sample_data["content_metadata"][0]
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add([fetcher]) swh_storage.metadata_fetcher_add([fetcher])
swh_storage.metadata_authority_add([authority]) swh_storage.metadata_authority_add([authority])
swh_storage.object_metadata_add([origin_metadata, origin_metadata2]) swh_storage.object_metadata_add([origin_metadata, origin_metadata2])
with pytest.raises(StorageArgumentException, match="SWHID"): with pytest.raises(StorageArgumentException, match="SWHID"):
▲ Show 20 Lines • Show All 182 Lines • ▼ Show 20 Lines def test_origin_list(self, swh_storage, swh_origins, limit):
assert i * limit >= len(swh_origins) assert i * limit >= len(swh_origins)
break break
else: else:
assert len(result["origins"]) == limit assert len(result["origins"]) == limit
expected_origins = [origin.url for origin in swh_origins] expected_origins = [origin.url for origin in swh_origins]
assert sorted(returned_origins) == sorted(expected_origins) assert sorted(returned_origins) == sorted(expected_origins)
def test_origin_count(self, swh_storage, sample_data_model): def test_origin_count(self, swh_storage, sample_data):
swh_storage.origin_add(sample_data_model["origin"]) swh_storage.origin_add(sample_data["origin"])
assert swh_storage.origin_count("github") == 3 assert swh_storage.origin_count("github") == 3
assert swh_storage.origin_count("gitlab") == 2 assert swh_storage.origin_count("gitlab") == 2
assert swh_storage.origin_count(".*user.*", regexp=True) == 5 assert swh_storage.origin_count(".*user.*", regexp=True) == 5
assert swh_storage.origin_count(".*user.*", regexp=False) == 0 assert swh_storage.origin_count(".*user.*", regexp=False) == 0
assert swh_storage.origin_count(".*user1.*", regexp=True) == 2 assert swh_storage.origin_count(".*user1.*", regexp=True) == 2
assert swh_storage.origin_count(".*user1.*", regexp=False) == 0 assert swh_storage.origin_count(".*user1.*", regexp=False) == 0
def test_origin_count_with_visit_no_visits(self, swh_storage, sample_data_model): def test_origin_count_with_visit_no_visits(self, swh_storage, sample_data):
swh_storage.origin_add(sample_data_model["origin"]) swh_storage.origin_add(sample_data["origin"])
# none of them have visits, so with_visit=True => 0 # none of them have visits, so with_visit=True => 0
assert swh_storage.origin_count("github", with_visit=True) == 0 assert swh_storage.origin_count("github", with_visit=True) == 0
assert swh_storage.origin_count("gitlab", with_visit=True) == 0 assert swh_storage.origin_count("gitlab", with_visit=True) == 0
assert swh_storage.origin_count(".*user.*", regexp=True, with_visit=True) == 0 assert swh_storage.origin_count(".*user.*", regexp=True, with_visit=True) == 0
assert swh_storage.origin_count(".*user.*", regexp=False, with_visit=True) == 0 assert swh_storage.origin_count(".*user.*", regexp=False, with_visit=True) == 0
assert swh_storage.origin_count(".*user1.*", regexp=True, with_visit=True) == 0 assert swh_storage.origin_count(".*user1.*", regexp=True, with_visit=True) == 0
assert swh_storage.origin_count(".*user1.*", regexp=False, with_visit=True) == 0 assert swh_storage.origin_count(".*user1.*", regexp=False, with_visit=True) == 0
def test_origin_count_with_visit_with_visits_no_snapshot( def test_origin_count_with_visit_with_visits_no_snapshot(
self, swh_storage, sample_data_model self, swh_storage, sample_data
): ):
swh_storage.origin_add(sample_data_model["origin"]) swh_storage.origin_add(sample_data["origin"])
origin_url = "https://github.com/user1/repo1" origin_url = "https://github.com/user1/repo1"
visit = OriginVisit(origin=origin_url, date=now(), type="git",) visit = OriginVisit(origin=origin_url, date=now(), type="git",)
swh_storage.origin_visit_add([visit]) swh_storage.origin_visit_add([visit])
assert swh_storage.origin_count("github", with_visit=False) == 3 assert swh_storage.origin_count("github", with_visit=False) == 3
# it has a visit, but no snapshot, so with_visit=True => 0 # it has a visit, but no snapshot, so with_visit=True => 0
assert swh_storage.origin_count("github", with_visit=True) == 0 assert swh_storage.origin_count("github", with_visit=True) == 0
assert swh_storage.origin_count("gitlab", with_visit=False) == 2 assert swh_storage.origin_count("gitlab", with_visit=False) == 2
# these gitlab origins have no visit # these gitlab origins have no visit
assert swh_storage.origin_count("gitlab", with_visit=True) == 0 assert swh_storage.origin_count("gitlab", with_visit=True) == 0
assert ( assert (
swh_storage.origin_count("github.*user1", regexp=True, with_visit=False) swh_storage.origin_count("github.*user1", regexp=True, with_visit=False)
== 1 == 1
) )
assert ( assert (
swh_storage.origin_count("github.*user1", regexp=True, with_visit=True) == 0 swh_storage.origin_count("github.*user1", regexp=True, with_visit=True) == 0
) )
assert swh_storage.origin_count("github", regexp=True, with_visit=True) == 0 assert swh_storage.origin_count("github", regexp=True, with_visit=True) == 0
def test_origin_count_with_visit_with_visits_and_snapshot( def test_origin_count_with_visit_with_visits_and_snapshot(
self, swh_storage, sample_data_model self, swh_storage, sample_data
): ):
snapshot = sample_data_model["snapshot"][0] snapshot = sample_data["snapshot"][0]
swh_storage.origin_add(sample_data_model["origin"]) swh_storage.origin_add(sample_data["origin"])
swh_storage.snapshot_add([snapshot]) swh_storage.snapshot_add([snapshot])
origin_url = "https://github.com/user1/repo1" origin_url = "https://github.com/user1/repo1"
visit = OriginVisit(origin=origin_url, date=now(), type="git",) visit = OriginVisit(origin=origin_url, date=now(), type="git",)
visit = swh_storage.origin_visit_add([visit])[0] visit = swh_storage.origin_visit_add([visit])[0]
swh_storage.origin_visit_status_add( swh_storage.origin_visit_status_add(
[ [
OriginVisitStatus( OriginVisitStatus(
Show All 36 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, sample_data_model): def test_content_add_objstorage_exception(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
swh_storage.objstorage.content_add = Mock( swh_storage.objstorage.content_add = Mock(
side_effect=Exception("mocked broken objstorage") side_effect=Exception("mocked broken objstorage")
) )
with pytest.raises(Exception, match="mocked broken"): with pytest.raises(Exception, match="mocked broken"):
swh_storage.content_add([content]) swh_storage.content_add([content])
missing = list(swh_storage.content_missing([content.hashes()])) missing = list(swh_storage.content_missing([content.hashes()]))
assert missing == [content.sha1] assert missing == [content.sha1]
@pytest.mark.db @pytest.mark.db
class TestStorageRaceConditions: class TestStorageRaceConditions:
@pytest.mark.xfail @pytest.mark.xfail
def test_content_add_race(self, swh_storage, sample_data_model): def test_content_add_race(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
results = queue.Queue() results = queue.Queue()
def thread(): def thread():
try: try:
with db_transaction(swh_storage) as (db, cur): with db_transaction(swh_storage) as (db, cur):
ret = swh_storage.content_add([content], db=db, cur=cur) ret = swh_storage.content_add([content], db=db, cur=cur)
results.put((threading.get_ident(), "data", ret)) results.put((threading.get_ident(), "data", ret))
Show All 24 Lines
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_with_new_cols(self, swh_storage, sample_data_model): def test_content_update_with_new_cols(self, swh_storage, sample_data):
content, content2 = sample_data_model["content"][:2] content, content2 = sample_data["content"][:2]
swh_storage.journal_writer.journal = None # TODO, not supported swh_storage.journal_writer.journal = None # TODO, not supported
with db_transaction(swh_storage) as (_, cur): with db_transaction(swh_storage) as (_, cur):
cur.execute( cur.execute(
"""alter table content """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 All 27 Lines def test_content_update_with_new_cols(self, swh_storage, sample_data):
) )
with db_transaction(swh_storage) as (_, cur): with db_transaction(swh_storage) as (_, cur):
cur.execute( cur.execute(
"""alter table content drop column test, """alter table content drop column test,
drop column test2""" drop column test2"""
) )
def test_content_add_db(self, swh_storage, sample_data_model): def test_content_add_db(self, swh_storage, sample_data):
content = sample_data_model["content"][0] content = sample_data["content"][0]
actual_result = swh_storage.content_add([content]) actual_result = swh_storage.content_add([content])
assert actual_result == { assert actual_result == {
"content:add": 1, "content:add": 1,
"content:add:bytes": content.length, "content:add:bytes": content.length,
} }
Show All 19 Lines def test_content_add_db(self, swh_storage, sample_data):
contents = [ contents = [
obj obj
for (obj_type, obj) in swh_storage.journal_writer.journal.objects for (obj_type, obj) in swh_storage.journal_writer.journal.objects
if obj_type == "content" if obj_type == "content"
] ]
assert len(contents) == 1 assert len(contents) == 1
assert contents[0] == attr.evolve(content, data=None) assert contents[0] == attr.evolve(content, data=None)
def test_content_add_metadata_db(self, swh_storage, sample_data_model): def test_content_add_metadata_db(self, swh_storage, sample_data):
content = attr.evolve(sample_data_model["content"][0], data=None, ctime=now()) content = attr.evolve(sample_data["content"][0], data=None, ctime=now())
actual_result = swh_storage.content_add_metadata([content]) actual_result = swh_storage.content_add_metadata([content])
assert actual_result == { assert actual_result == {
"content:add": 1, "content:add": 1,
} }
if hasattr(swh_storage, "objstorage"): if hasattr(swh_storage, "objstorage"):
Show All 16 Lines def test_content_add_metadata_db(self, swh_storage, sample_data):
contents = [ contents = [
obj obj
for (obj_type, obj) in swh_storage.journal_writer.journal.objects for (obj_type, obj) in swh_storage.journal_writer.journal.objects
if obj_type == "content" if obj_type == "content"
] ]
assert len(contents) == 1 assert len(contents) == 1
assert contents[0] == content assert contents[0] == content
def test_skipped_content_add_db(self, swh_storage, sample_data_model): def test_skipped_content_add_db(self, swh_storage, sample_data):
content, cont2 = sample_data_model["skipped_content"][:2] content, cont2 = sample_data["skipped_content"][:2]
content2 = attr.evolve(cont2, blake2s256=None) content2 = attr.evolve(cont2, blake2s256=None)
actual_result = swh_storage.skipped_content_add([content, content, content2]) actual_result = swh_storage.skipped_content_add([content, content, content2])
assert 2 <= actual_result.pop("skipped_content:add") <= 3 assert 2 <= actual_result.pop("skipped_content:add") <= 3
assert actual_result == {} assert actual_result == {}
with db_transaction(swh_storage) as (_, cur): with db_transaction(swh_storage) as (_, cur):
Show All 40 Lines
About · Developer information · Legal