Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 2,933 Lines • ▼ Show 20 Lines | def test_snapshot_missing(self, swh_storage, sample_data_model): | ||||
snapshot, missing_snapshot = sample_data_model["snapshot"][:2] | snapshot, missing_snapshot = sample_data_model["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): | def test_stat_counters(self, swh_storage, sample_data_model): | ||||
origin = sample_data_model["origin"][0] | |||||
snapshot = sample_data_model["snapshot"][0] | |||||
revision = sample_data_model["revision"][0] | |||||
release = sample_data_model["release"][0] | |||||
directory = sample_data_model["directory"][0] | |||||
content = sample_data_model["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) | ||||
for key in expected_keys: | for key in expected_keys: | ||||
assert counters[key] == 0 | assert counters[key] == 0 | ||||
# Add a content. Only the content counter should increase. | # Add a content. Only the content counter should increase. | ||||
swh_storage.content_add([data.cont]) | swh_storage.content_add([content]) | ||||
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) | ||||
for key in expected_keys: | for key in expected_keys: | ||||
if key != "content": | if key != "content": | ||||
assert counters[key] == 0 | assert counters[key] == 0 | ||||
assert counters["content"] == 1 | assert counters["content"] == 1 | ||||
# Add other objects. Check their counter increased as well. | # Add other objects. Check their counter increased as well. | ||||
swh_storage.origin_add([data.origin2]) | swh_storage.origin_add([origin]) | ||||
origin_url = data.origin2["url"] | |||||
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] | ||||
swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=origin_visit1.visit, | visit=origin_visit1.visit, | ||||
date=now(), | date=now(), | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.snapshot["id"], | snapshot=snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
swh_storage.directory_add([data.dir]) | swh_storage.directory_add([directory]) | ||||
swh_storage.revision_add([data.revision]) | swh_storage.revision_add([revision]) | ||||
swh_storage.release_add([data.release]) | swh_storage.release_add([release]) | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
counters = swh_storage.stat_counters() | counters = swh_storage.stat_counters() | ||||
assert counters["content"] == 1 | assert counters["content"] == 1 | ||||
assert counters["directory"] == 1 | assert counters["directory"] == 1 | ||||
assert counters["snapshot"] == 1 | assert counters["snapshot"] == 1 | ||||
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 | ||||
ardumont: "bravo encore une fois" ;) | |||||
if "person" in counters: | if "person" in counters: | ||||
assert counters["person"] == 3 | assert counters["person"] == 3 | ||||
def test_content_find_ctime(self, swh_storage): | def test_content_find_ctime(self, swh_storage): | ||||
cont = data.cont.copy() | cont = data.cont.copy() | ||||
del cont["data"] | del cont["data"] | ||||
ctime = now() | ctime = now() | ||||
cont["ctime"] = ctime | cont["ctime"] = ctime | ||||
▲ Show 20 Lines • Show All 1,274 Lines • Show Last 20 Lines |
"bravo encore une fois" ;)