diff --git a/requirements-test.txt b/requirements-test.txt --- a/requirements-test.txt +++ b/requirements-test.txt @@ -1 +1,2 @@ pytest +requests_mock diff --git a/swh/loader/package/tests/base_test_loader.py b/swh/loader/package/tests/base_test_loader.py new file mode 100644 --- /dev/null +++ b/swh/loader/package/tests/base_test_loader.py @@ -0,0 +1,129 @@ +import os + + +class BaseTestLoader: + + _expected_new_contents_first_visit = [] + + _expected_new_contents_second_visit = [] + + _expected_new_directories_first_visit = [] + + _expected_new_directories_second_visit = [] + + _expected_new_revisions_first_visit = {} + + _expected_new_revisions_second_visit = {} + + _expected_new_snapshot_first_visit = '' + + _expected_branches_first_visit = {} + + _expected_new_snapshot_second_visit = '' + + _expected_branches_second_visit = {} + + def first_visit_test(self): + """In this scenario no visit as taken place prior to this visit. + + """ + self.assertCountContents(len(self._expected_new_contents_first_visit)) + self.assertCountDirectories(len(self._expected_new_directories_first_visit)) # noqa + + self.assertCountReleases(0, 'No release is created by the loader') + self.assertCountSnapshots(1, 'Only 1 snapshot targeting all revisions') + + self.assertContentsContain(self._expected_new_contents_first_visit) + self.assertDirectoriesContain(self._expected_new_directories_first_visit) # noqa + self.assertRevisionsContain(self._expected_new_revisions_first_visit) + self.assertSnapshotEqual(self._expected_new_snapshot_first_visit, + self._expected_branches_first_visit) + + self.assertEqual(self.loader.counters['contents'], + len(self._expected_new_contents_first_visit)) + self.assertEqual(self.loader.counters['directories'], + len(self._expected_new_directories_first_visit)) + self.assertEqual(self.loader.counters['revisions'], + len(self._expected_new_revisions_first_visit)) + self.assertEqual(self.loader.counters['releases'], 0) + + self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) + self.assertEqual(self.loader.visit_status(), 'full') + self.assertFalse(os.path.exists(self.loader.temp_directory)) + + def first_visit_again_test(self): + """This scenario makes use of the incremental nature of the loader. + + In this test there is no change from the first visit. If nothing + changes in between visits, the snapshot for the visit must stay + the same as the first visit. + + """ + # previously loaded objects should still be here + self.assertCountContents(len(self._expected_new_contents_first_visit)) + self.assertCountDirectories(len(self._expected_new_directories_first_visit)) # noqa + self.assertCountRevisions(len(self._expected_new_revisions_first_visit)) # noqa + self.assertCountReleases(0) + self.assertCountSnapshots(1) + self.assertSnapshotEqual(self._expected_new_snapshot_first_visit, + self._expected_branches_first_visit) + + # no objects should have been loaded in that visit + counters_reset = dict.fromkeys(self.loader.counters.keys(), 0) + self.assertEqual(self.loader.counters, counters_reset) + + self.assertEqual(self.loader.load_status(), {'status': 'uneventful'}) + self.assertEqual(self.loader.visit_status(), 'full') + self.assertFalse(os.path.exists(self.loader.temp_directory)) + + def second_visit_test(self): + """In this scenario, a visit has already taken place. + An existing snapshot exists. + + This time, a new release has been uploaded. The old releases did not + change. + + The visit results in a new snapshot. + + The new snapshot shares the same history as prior visit's + snapshot. It holds a new branch targeting the new revision. + + """ + expected_nb_contents = sum([len(self._expected_new_contents_first_visit), # noqa + len(self._expected_new_contents_second_visit)]) # noqa + + expected_nb_directories = sum([len(self._expected_new_directories_first_visit), # noqa + len(self._expected_new_directories_second_visit)]) # noqa + + expected_nb_revisions = sum([len(self._expected_new_revisions_first_visit), # noqa + len(self._expected_new_revisions_second_visit)]) # noqa + + self.assertCountContents(expected_nb_contents) + self.assertCountDirectories(expected_nb_directories) + self.assertCountRevisions(expected_nb_revisions) + self.assertCountReleases(0) + self.assertCountSnapshots(2) + + self.assertContentsContain(self._expected_new_contents_first_visit) + self.assertContentsContain(self._expected_new_contents_second_visit) + self.assertDirectoriesContain(self._expected_new_directories_first_visit) # noqa + self.assertDirectoriesContain(self._expected_new_directories_second_visit) # noqa + self.assertRevisionsContain(self._expected_new_revisions_first_visit) + self.assertRevisionsContain(self._expected_new_revisions_second_visit) + + self.assertSnapshotEqual(self._expected_new_snapshot_first_visit, + self._expected_branches_first_visit) + self.assertSnapshotEqual(self._expected_new_snapshot_second_visit, + self._expected_branches_second_visit) + + self.assertEqual(self.loader.counters['contents'], + len(self._expected_new_contents_second_visit)) + self.assertEqual(self.loader.counters['directories'], + len(self._expected_new_directories_second_visit)) + self.assertEqual(self.loader.counters['revisions'], + len(self._expected_new_revisions_second_visit)) + self.assertEqual(self.loader.counters['releases'], 0) + + self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) + self.assertEqual(self.loader.visit_status(), 'full') + self.assertFalse(os.path.exists(self.loader.temp_directory)) diff --git a/swh/loader/package/tests/download_test_base.py b/swh/loader/package/tests/download_test_base.py new file mode 100644 --- /dev/null +++ b/swh/loader/package/tests/download_test_base.py @@ -0,0 +1,65 @@ +# Copyright (C) 2019 The Software Heritage developers +# See the AUTHORS file at the top-level directory of this distribution +# License: GNU General Public License version 3, or any later version +# See top-level LICENSE file for more information + + +def test_compare_field(fl): + """Check if the `compare_field` variable is declared + + """ + if fl.compare_field is None: + assert False + + +def test_filter_package_versions(fl, tarballs): + """Tests :func:`filter_package_versions` for different conditions. + + This test checks :func:`filter_package_versions` for three different + scenario + * No tarball is already known + * Only one tarball in already known + * All the tarballs are known + + """ + # test for no known version + filtered_tarballs = fl.filter_package_versions(tarballs, {}) + assert filtered_tarballs == tarballs + + known_versions = {} + # test with one known version + # Assuming first tarball is already known + key = fl.compare_field + known_tarball_key = tarballs[0][key] + known_versions[known_tarball_key] = 'some_revision_id' + filtered_tarballs = fl.filter_package_versions(tarballs, + known_versions) + # one tarball removed + assert len(filtered_tarballs) == (len(tarballs) - 1) + + # Check if the know tarball is present in filtered tarballs + for tarball in filtered_tarballs: + if known_tarball_key in tarball: + assert False + + # if all versions are known + for tarball in tarballs: + known_versions[tarball[key]] = 'some_revision_id' + + filtered_tarballs = fl.filter_package_versions(tarballs, + known_versions) + assert filtered_tarballs == [] + + +def test_invalid_request(fl, tarballs, requests_mock): + """It tests for scenario when the tarball url are invalid. + + In this scenario all the tarball url gives 404 error. + + """ + for tarball in tarballs: + requests_mock.get(tarball['url'], text='Not Found', status_code=404) + + generator = fl.prepare_package_versions(tarballs) + package_versions = [data for data in generator] + assert package_versions == []