Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 489 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
self.complete_snapshot = { | self.complete_snapshot = { | ||||
'id': hash_to_bytes('6e65b86363953b780d92b0a928f3e8fcdd10db36'), | 'id': hash_to_bytes('6e65b86363953b780d92b0a928f3e8fcdd10db36'), | ||||
'branches': { | 'branches': { | ||||
b'directory': { | b'directory': { | ||||
'target': hash_to_bytes( | 'target': hash_to_bytes( | ||||
'1bd0e65f7d2ff14ae994de17a1e7fe65111dcad8'), | '1bd0e65f7d2ff14ae994de17a1e7fe65111dcad8'), | ||||
'target_type': 'directory', | 'target_type': 'directory', | ||||
}, | }, | ||||
b'directory2': { | |||||
'target': hash_to_bytes( | |||||
'1bd0e65f7d2ff14ae994de17a1e7fe65111dcad8'), | |||||
'target_type': 'directory', | |||||
}, | |||||
b'content': { | b'content': { | ||||
'target': hash_to_bytes( | 'target': hash_to_bytes( | ||||
'fe95a46679d128ff167b7c55df5d02356c5a1ae1'), | 'fe95a46679d128ff167b7c55df5d02356c5a1ae1'), | ||||
'target_type': 'content', | 'target_type': 'content', | ||||
}, | }, | ||||
b'alias': { | b'alias': { | ||||
'target': b'revision', | 'target': b'revision', | ||||
'target_type': 'alias', | 'target_type': 'alias', | ||||
▲ Show 20 Lines • Show All 1,413 Lines • ▼ Show 20 Lines | def test_snapshot_add_count_branches(self): | ||||
self.assertEqual(actual_result, {'snapshot:add': 1}) | self.assertEqual(actual_result, {'snapshot:add': 1}) | ||||
snp_id = self.complete_snapshot['id'] | snp_id = self.complete_snapshot['id'] | ||||
snp_size = self.storage.snapshot_count_branches(snp_id) | snp_size = self.storage.snapshot_count_branches(snp_id) | ||||
expected_snp_size = { | expected_snp_size = { | ||||
'alias': 1, | 'alias': 1, | ||||
'content': 1, | 'content': 1, | ||||
'directory': 1, | 'directory': 2, | ||||
'release': 1, | 'release': 1, | ||||
'revision': 1, | 'revision': 1, | ||||
'snapshot': 1, | 'snapshot': 1, | ||||
None: 1 | None: 1 | ||||
} | } | ||||
self.assertEqual(snp_size, expected_snp_size) | self.assertEqual(snp_size, expected_snp_size) | ||||
def test_snapshot_add_get_paginated(self): | def test_snapshot_add_get_paginated(self): | ||||
origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
origin_visit1 = self.storage.origin_visit_add(origin_id, | origin_visit1 = self.storage.origin_visit_add(origin_id, | ||||
self.date_visit1) | self.date_visit1) | ||||
visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
self.storage.snapshot_add(origin_id, visit_id, self.complete_snapshot) | self.storage.snapshot_add(origin_id, visit_id, self.complete_snapshot) | ||||
snp_id = self.complete_snapshot['id'] | snp_id = self.complete_snapshot['id'] | ||||
branches = self.complete_snapshot['branches'] | branches = self.complete_snapshot['branches'] | ||||
branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
# Test branch_from | |||||
snapshot = self.storage.snapshot_get_branches(snp_id, | snapshot = self.storage.snapshot_get_branches(snp_id, | ||||
branches_from=b'release') | branches_from=b'release') | ||||
rel_idx = branch_names.index(b'release') | rel_idx = branch_names.index(b'release') | ||||
expected_snapshot = { | expected_snapshot = { | ||||
'id': snp_id, | 'id': snp_id, | ||||
'branches': { | 'branches': { | ||||
name: branches[name] | name: branches[name] | ||||
for name in branch_names[rel_idx:] | for name in branch_names[rel_idx:] | ||||
}, | }, | ||||
'next_branch': None, | 'next_branch': None, | ||||
} | } | ||||
self.assertEqual(snapshot, expected_snapshot) | self.assertEqual(snapshot, expected_snapshot) | ||||
# Test branches_count | |||||
snapshot = self.storage.snapshot_get_branches(snp_id, | snapshot = self.storage.snapshot_get_branches(snp_id, | ||||
branches_count=1) | branches_count=1) | ||||
expected_snapshot = { | expected_snapshot = { | ||||
'id': snp_id, | 'id': snp_id, | ||||
'branches': { | 'branches': { | ||||
branch_names[0]: branches[branch_names[0]], | branch_names[0]: branches[branch_names[0]], | ||||
}, | }, | ||||
'next_branch': b'content', | 'next_branch': b'content', | ||||
} | } | ||||
self.assertEqual(snapshot, expected_snapshot) | self.assertEqual(snapshot, expected_snapshot) | ||||
# test branch_from + branches_count | |||||
snapshot = self.storage.snapshot_get_branches( | snapshot = self.storage.snapshot_get_branches( | ||||
snp_id, branches_from=b'directory', branches_count=3) | snp_id, branches_from=b'directory', branches_count=3) | ||||
dir_idx = branch_names.index(b'directory') | dir_idx = branch_names.index(b'directory') | ||||
expected_snapshot = { | expected_snapshot = { | ||||
'id': snp_id, | 'id': snp_id, | ||||
'branches': { | 'branches': { | ||||
name: branches[name] | name: branches[name] | ||||
Show All 40 Lines | def test_snapshot_add_get_filtered(self): | ||||
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, | ||||
} | } | ||||
self.assertEqual(snapshot, expected_snapshot) | self.assertEqual(snapshot, expected_snapshot) | ||||
def test_snapshot_add_get_filtered_and_paginated(self): | |||||
origin_id = self.storage.origin_add_one(self.origin) | |||||
origin_visit1 = self.storage.origin_visit_add(origin_id, | |||||
self.date_visit1) | |||||
visit_id = origin_visit1['visit'] | |||||
self.storage.snapshot_add(origin_id, visit_id, self.complete_snapshot) | |||||
snp_id = self.complete_snapshot['id'] | |||||
branches = self.complete_snapshot['branches'] | |||||
branch_names = list(sorted(branches)) | |||||
# Test branch_from | |||||
snapshot = self.storage.snapshot_get_branches( | |||||
snp_id, target_types=['directory', 'release'], | |||||
branches_from=b'directory2') | |||||
expected_snapshot = { | |||||
'id': snp_id, | |||||
'branches': { | |||||
name: branches[name] | |||||
for name in (b'directory2', b'release') | |||||
}, | |||||
'next_branch': None, | |||||
} | |||||
self.assertEqual(snapshot, expected_snapshot) | |||||
# Test branches_count | |||||
snapshot = self.storage.snapshot_get_branches( | |||||
snp_id, target_types=['directory', 'release'], | |||||
branches_count=1) | |||||
expected_snapshot = { | |||||
'id': snp_id, | |||||
'branches': { | |||||
b'directory': branches[b'directory'] | |||||
}, | |||||
'next_branch': b'directory2', | |||||
} | |||||
self.assertEqual(snapshot, expected_snapshot) | |||||
# Test branches_count | |||||
snapshot = self.storage.snapshot_get_branches( | |||||
snp_id, target_types=['directory', 'release'], | |||||
branches_count=2) | |||||
expected_snapshot = { | |||||
'id': snp_id, | |||||
'branches': { | |||||
name: branches[name] | |||||
for name in (b'directory', b'directory2') | |||||
}, | |||||
'next_branch': b'release', | |||||
} | |||||
self.assertEqual(snapshot, expected_snapshot) | |||||
# test branch_from + branches_count | |||||
snapshot = self.storage.snapshot_get_branches( | |||||
snp_id, target_types=['directory', 'release'], | |||||
branches_from=b'directory2', branches_count=1) | |||||
dir_idx = branch_names.index(b'directory2') | |||||
expected_snapshot = { | |||||
'id': snp_id, | |||||
'branches': { | |||||
branch_names[dir_idx]: branches[branch_names[dir_idx]], | |||||
}, | |||||
'next_branch': b'release', | |||||
} | |||||
self.assertEqual(snapshot, expected_snapshot) | |||||
def test_snapshot_add_get(self): | def test_snapshot_add_get(self): | ||||
origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
origin_visit1 = self.storage.origin_visit_add(origin_id, | origin_visit1 = self.storage.origin_visit_add(origin_id, | ||||
self.date_visit1) | self.date_visit1) | ||||
visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
self.storage.snapshot_add(origin_id, visit_id, self.snapshot) | self.storage.snapshot_add(origin_id, visit_id, self.snapshot) | ||||
▲ Show 20 Lines • Show All 1,322 Lines • Show Last 20 Lines |