Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/core/tests/__init__.py
| Show All 10 Lines | |||||
| from unittest import TestCase | from unittest import TestCase | ||||
| from swh.model import hashutil | from swh.model import hashutil | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| class BaseLoaderStorageTest: | class BaseLoaderStorageTest: | ||||
| def _assertCountOk(self, type, expected_length, msg=None): | def _assertCountEqual(self, type, expected_length, msg=None): | ||||
| """Check typed 'type' state to have the same expected length. | """Check typed 'type' state to have the same expected length. | ||||
| """ | """ | ||||
| self.storage.refresh_stat_counters() | self.storage.refresh_stat_counters() | ||||
| self.assertEqual(self.storage.stat_counters()[type], | self.assertEqual(self.storage.stat_counters()[type], | ||||
| expected_length, msg=msg) | expected_length, msg=msg) | ||||
| def assertCountContents(self, len_expected_contents, msg=None): | def assertCountContents(self, len_expected_contents, msg=None): | ||||
| self._assertCountOk('content', len_expected_contents, msg=msg) | self._assertCountEqual('content', len_expected_contents, msg=msg) | ||||
| def assertCountDirectories(self, len_expected_directories, msg=None): | def assertCountDirectories(self, len_expected_directories, msg=None): | ||||
| self._assertCountOk('directory', len_expected_directories, msg=msg) | self._assertCountEqual('directory', len_expected_directories, | ||||
| msg=msg) | |||||
| def assertCountReleases(self, len_expected_releases, msg=None): | def assertCountReleases(self, len_expected_releases, msg=None): | ||||
| self._assertCountOk('release', len_expected_releases, msg=msg) | self._assertCountEqual('release', len_expected_releases, msg=msg) | ||||
| def assertCountRevisions(self, len_expected_revisions, msg=None): | def assertCountRevisions(self, len_expected_revisions, msg=None): | ||||
| self._assertCountOk('revision', len_expected_revisions, msg=msg) | self._assertCountEqual('revision', len_expected_revisions, msg=msg) | ||||
| def assertCountSnapshots(self, len_expected_snapshot, msg=None): | def assertCountSnapshots(self, len_expected_snapshot, msg=None): | ||||
| self._assertCountOk('snapshot', len_expected_snapshot, msg=msg) | self._assertCountEqual('snapshot', len_expected_snapshot, msg=msg) | ||||
| def assertContentsOk(self, expected_contents): | def assertContentsContain(self, expected_contents): | ||||
anlambert: s/Checks/Check/ | |||||
| self._assertCountOk('content', len(expected_contents)) | """Check the provided content are a subset of the stored ones. | ||||
| Args: | |||||
| expected_contents ([sha1]): List of content ids""" | |||||
| self._assertCountEqual('content', len(expected_contents)) | |||||
| missing = list(self.storage.content_missing( | missing = list(self.storage.content_missing( | ||||
| {'sha1': hash_to_bytes(content_hash)} | {'sha1': hash_to_bytes(content_hash)} | ||||
| for content_hash in expected_contents)) | for content_hash in expected_contents)) | ||||
| self.assertEqual(missing, []) | self.assertEqual(missing, []) | ||||
| def assertDirectoriesOk(self, expected_directories): | def assertDirectoriesContain(self, expected_directories): | ||||
| self._assertCountOk('directory', len(expected_directories)) | """Check the provided directories are a subset of the stored ones. | ||||
Not Done Inline Actionss/Checks/Check/ anlambert: s/Checks/Check/ | |||||
| Args: | |||||
| expected_directories ([sha1]): List of directory ids.""" | |||||
| self._assertCountEqual('directory', len(expected_directories)) | |||||
| missing = list(self.storage.directory_missing( | missing = list(self.storage.directory_missing( | ||||
| dir_['id'] for dir_ in expected_directories)) | dir_['id'] for dir_ in expected_directories)) | ||||
| self.assertEqual(missing, []) | self.assertEqual(missing, []) | ||||
| def assertReleasesOk(self, expected_releases): | def assertReleasesContain(self, expected_releases): | ||||
| """Check the loader's releases match the expected releases. | """Check the provided releases are a subset of the stored ones. | ||||
| Args: | Args: | ||||
| releases ([dict]): List of dictionaries representing swh releases. | releases ([dict]): List of dictionaries representing swh releases. | ||||
| """ | """ | ||||
| self._assertCountOk('release', len(expected_releases)) | self._assertCountEqual('release', len(expected_releases)) | ||||
| missing = list(self.storage.releases_missing( | missing = list(self.storage.releases_missing( | ||||
| rel['id'] for rel in expected_releases)) | rel['id'] for rel in expected_releases)) | ||||
| self.assertEqual(missing, []) | self.assertEqual(missing, []) | ||||
| def assertRevisionsOk(self, expected_revisions): | def assertRevisionsContain(self, expected_revisions): | ||||
| """Check the loader's revisions match the expected revisions. | """Check the provided revisions are a subset of the stored ones. | ||||
Not Done Inline Actionss/Checks/Check/ anlambert: s/Checks/Check/ | |||||
| Expects self.loader to be instantiated and ready to be | Expects self.loader to be instantiated and ready to be | ||||
| inspected (meaning the loading took place). | inspected (meaning the loading took place). | ||||
| Args: | Args: | ||||
| expected_revisions (dict): Dict with key revision id, | expected_revisions (dict): Dict with key revision id, | ||||
| value the targeted directory id. | value the targeted directory id. | ||||
| """ | """ | ||||
| self._assertCountOk('revision', len(expected_revisions)) | self._assertCountEqual('revision', len(expected_revisions)) | ||||
| revs = list(self.storage.revision_get( | revs = list(self.storage.revision_get( | ||||
| hashutil.hash_to_bytes(rev_id) for rev_id in expected_revisions)) | hashutil.hash_to_bytes(rev_id) for rev_id in expected_revisions)) | ||||
| self.assertNotIn(None, revs) | self.assertNotIn(None, revs) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| {rev['id']: rev['directory'] for rev in revs}, | {rev['id']: rev['directory'] for rev in revs}, | ||||
| {hash_to_bytes(rev_id): hash_to_bytes(rev_dir) | {hash_to_bytes(rev_id): hash_to_bytes(rev_dir) | ||||
| for (rev_id, rev_dir) in expected_revisions.items()}) | for (rev_id, rev_dir) in expected_revisions.items()}) | ||||
| def assertSnapshotOk(self, expected_snapshot, expected_branches=[]): | def assertSnapshotEqual(self, expected_snapshot, expected_branches=[]): | ||||
| """Check for snapshot match. | """Check for snapshot match. | ||||
| Provide the hashes as hexadecimal, the conversion is done | Provide the hashes as hexadecimal, the conversion is done | ||||
| within the method. | within the method. | ||||
| Args: | Args: | ||||
| expected_snapshot (str/dict): Either the snapshot | expected_snapshot (str/dict): Either the snapshot | ||||
| identifier or the full | identifier or the full | ||||
| snapshot | snapshot | ||||
| expected_branches (dict): expected branches or nothing is | expected_branches (dict): expected branches or nothing is | ||||
| the full snapshot is provided | the full snapshot is provided | ||||
| """ | """ | ||||
| if isinstance(expected_snapshot, dict) and not expected_branches: | if isinstance(expected_snapshot, dict) and not expected_branches: | ||||
| expected_snapshot_id = expected_snapshot['id'] | expected_snapshot_id = expected_snapshot['id'] | ||||
| expected_branches = expected_snapshot['branches'] | expected_branches = expected_snapshot['branches'] | ||||
| else: | else: | ||||
| expected_snapshot_id = expected_snapshot | expected_snapshot_id = expected_snapshot | ||||
| self._assertCountOk('snapshot', 1) | self._assertCountEqual('snapshot', 1) | ||||
| snap = self.storage.snapshot_get(hash_to_bytes(expected_snapshot_id)) | snap = self.storage.snapshot_get(hash_to_bytes(expected_snapshot_id)) | ||||
| self.assertIsNotNone(snap) | self.assertIsNotNone(snap) | ||||
| def decode_target(target): | def decode_target(target): | ||||
| if not target: | if not target: | ||||
| return target | return target | ||||
| target_type = target['target_type'] | target_type = target['target_type'] | ||||
| Show All 11 Lines | def assertSnapshotEqual(self, expected_snapshot, expected_branches=[]): | ||||
| branches = { | branches = { | ||||
| branch.decode('utf-8'): decode_target(target) | branch.decode('utf-8'): decode_target(target) | ||||
| for branch, target in snap['branches'].items() | for branch, target in snap['branches'].items() | ||||
| } | } | ||||
| self.assertEqual(expected_branches, branches) | self.assertEqual(expected_branches, branches) | ||||
| def assertOriginMetadataContains(self, origin_type, origin_url, | def assertOriginMetadataContains(self, origin_type, origin_url, | ||||
| expected_origin_metadata): | expected_origin_metadata): | ||||
| """Checks the storage contains this metadata for the given origin. | """Check the storage contains this metadata for the given origin. | ||||
Not Done Inline Actionss/Checks/Check/ anlambert: s/Checks/Check/ | |||||
| Args: | Args: | ||||
| origin_type (str): type of origin ('deposit', 'git', 'svn', ...) | origin_type (str): type of origin ('deposit', 'git', 'svn', ...) | ||||
| origin_url (str): URL of the origin | origin_url (str): URL of the origin | ||||
| expected_origin_metadata (dict): | expected_origin_metadata (dict): | ||||
| Extrinsic metadata of the origin | Extrinsic metadata of the origin | ||||
| <https://forge.softwareheritage.org/T1344> | <https://forge.softwareheritage.org/T1344> | ||||
| ▲ Show 20 Lines • Show All 73 Lines • Show Last 20 Lines | |||||
s/Checks/Check/