Differential D3728 Diff 13161 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 180 Lines • ▼ Show 20 Lines	def test_content_add(self, swh_storage, sample_data):
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,
}		}

assert list(swh_storage.content_get([cont.sha1])) == [		assert swh_storage.content_get_data(cont.sha1) == cont.data
{"sha1": cont.sha1, "data": cont.data}
]

expected_cont = attr.evolve(cont, data=None)		expected_cont = attr.evolve(cont, data=None)

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"
]		]
Show All 18 Lines	def test_content_add_from_lazy_content(self, swh_storage, sample_data):

assert actual_result == {		assert actual_result == {
"content:add": 1,		"content:add": 1,
"content:add:bytes": cont.length,		"content:add:bytes": cont.length,
}		}

# the fact that we retrieve the content object from the storage with		# the fact that we retrieve the content object from the storage with
# the correct 'data' field ensures it has been 'called'		# the correct 'data' field ensures it has been 'called'
assert list(swh_storage.content_get([cont.sha1])) == [		assert swh_storage.content_get_data(cont.sha1) == cont.data
{"sha1": cont.sha1, "data": cont.data}
]

expected_cont = attr.evolve(lazy_content, data=None, ctime=None)		expected_cont = attr.evolve(lazy_content, data=None, ctime=None)
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:
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):		def test_content_get_data_missing(self, swh_storage, sample_data):
cont, cont2 = sample_data.contents[:2]		cont, cont2 = sample_data.contents[: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]))		actual_content_data = swh_storage.content_get_data(cont2.sha1)
assert results == [None]		assert actual_content_data is 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]))		actual_content_data = swh_storage.content_get_data(cont.sha1)
assert results == [{"sha1": cont.sha1, "data": cont.data}, None]		assert actual_content_data == cont.data
		actual_content_data = swh_storage.content_get_data(cont2.sha1)
# Check content_get does not discard found countent when it finds		assert actual_content_data is None
# a missing content.
results = list(swh_storage.content_get([cont2.sha1, cont.sha1]))
assert results == [None, {"sha1": cont.sha1, "data": cont.data}]

def test_content_add_different_input(self, swh_storage, sample_data):		def test_content_add_different_input(self, swh_storage, sample_data):
cont, cont2 = sample_data.contents[:2]		cont, cont2 = sample_data.contents[: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,
▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines	def test_content_add_collision(self, swh_storage, sample_data):
cont1.hashes(),		cont1.hashes(),
cont1b.hashes(),		cont1b.hashes(),
]		]

def test_content_add_duplicate(self, swh_storage, sample_data):		def test_content_add_duplicate(self, swh_storage, sample_data):
cont = sample_data.content		cont = sample_data.content
swh_storage.content_add([cont, cont])		swh_storage.content_add([cont, cont])

assert list(swh_storage.content_get([cont.sha1])) == [		assert swh_storage.content_get_data(cont.sha1) == cont.data
{"sha1": cont.sha1, "data": cont.data}
]

def test_content_update(self, swh_storage, sample_data):		def test_content_update(self, swh_storage, sample_data):
cont1 = sample_data.content		cont1 = sample_data.content

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])
▲ Show 20 Lines • Show All 3,490 Lines • ▼ Show 20 Lines	def test_origin_metadata_get__invalid_id_type(self, swh_storage, sample_data):

with pytest.raises(StorageArgumentException, match="SWHID"):		with pytest.raises(StorageArgumentException, match="SWHID"):
swh_storage.raw_extrinsic_metadata_get(		swh_storage.raw_extrinsic_metadata_get(
MetadataTargetType.ORIGIN, content_metadata.id, authority,		MetadataTargetType.ORIGIN, content_metadata.id, authority,
)		)


class TestStorageGeneratedData:		class TestStorageGeneratedData:
def test_generate_content_get(self, swh_storage, swh_contents):		def test_generate_content_get_data(self, swh_storage, swh_contents):
contents_with_data = [c.to_dict() for c in swh_contents if c.status != "absent"]		contents_with_data = [c for c in swh_contents if c.status != "absent"]
# input the list of sha1s we want from storage
get_sha1s = [c["sha1"] for c in contents_with_data]

# retrieve contents		# retrieve contents
actual_contents = list(swh_storage.content_get(get_sha1s))		for content in contents_with_data:
assert None not in actual_contents		actual_content_data = swh_storage.content_get_data(content.sha1)
assert_contents_ok(contents_with_data, actual_contents)		assert actual_content_data is not None
		assert actual_content_data == content.data

def test_generate_content_get_metadata(self, swh_storage, swh_contents):		def test_generate_content_get_metadata(self, swh_storage, swh_contents):
# input the list of sha1s we want from storage		# input the list of sha1s we want from storage
expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"]		expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"]
get_sha1s = [c["sha1"] for c in expected_contents]		get_sha1s = [c["sha1"] for c in expected_contents]

# retrieve contents		# retrieve contents
meta_contents = swh_storage.content_get_metadata(get_sha1s)		meta_contents = swh_storage.content_get_metadata(get_sha1s)
▲ Show 20 Lines • Show All 365 Lines • Show Last 20 Lines