Page Menu
Home
Software Heritage
Search
Configure Global Search
Log In
Files
F7163805
D1744.id5976.diff
No One
Temporary
Actions
View File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Flag For Later
Size
9 KB
Subscribers
None
D1744.id5976.diff
View Options
diff --git a/requirements-test.txt b/requirements-test.txt
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -1 +1,2 @@
pytest
+requests_mock
\ No newline at end of file
diff --git a/swh/loader/base/tests/base_test_loader.py b/swh/loader/base/tests/base_test_loader.py
new file mode 100644
--- /dev/null
+++ b/swh/loader/base/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/base/tests/download_test_base.py b/swh/loader/base/tests/download_test_base.py
new file mode 100644
--- /dev/null
+++ b/swh/loader/base/tests/download_test_base.py
@@ -0,0 +1,105 @@
+# 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
+
+import requests_mock
+from abc import abstractmethod
+
+
+class TestIf_Modified_Since():
+ """
+ I am under construction
+ See the class below me it is ready. When completed
+ I will be quite like that
+ """
+
+ def test_filter_package_versions():
+ pass
+
+ def test_invalid_request():
+ pass
+
+
+class BaseTestcompareField:
+ """Base tests for :class:`compareField`
+
+ """
+
+ tarballs = None
+ ("""Valuse of the key `tarball` returned by """
+ """the :func:`convert_to_standard_format` """)
+
+ @abstractmethod
+ def get_fl(self):
+ """Creates and returns a fake loader instance
+
+ Returns:
+ Fake loader instance
+
+ """
+ pass
+
+ def test_compare_field(self):
+ """Check if the `compare_field` variable is declared
+
+ """
+ fl = self.get_fl()
+ if fl.compare_field is None:
+ assert False
+
+ def test_filter_package_versions(self):
+ """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
+ fl = self.get_fl()
+ print(type(fl))
+ print(fl)
+ filtered_tarballs = fl.filter_package_versions(self.tarballs, {})
+ assert filtered_tarballs == self.tarballs
+
+ known_versions = {}
+ # test with one known version
+ # Assuming first tarball is already known
+ key = fl.compare_field
+ known_tarball_key = self.tarballs[0][key]
+ known_versions[known_tarball_key] = 'some_revision_id'
+ filtered_tarballs = fl.filter_package_versions(self.tarballs,
+ known_versions)
+ # one tarball removed
+ assert len(filtered_tarballs) == (len(self.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 self.tarballs:
+ known_versions[tarball[key]] = 'some_revision_id'
+
+ filtered_tarballs = fl.filter_package_versions(self.tarballs,
+ known_versions)
+ assert filtered_tarballs == []
+
+ @requests_mock.Mocker()
+ def test_invalid_request(self, http_mocker):
+ """It tests for scenario when the tarball url are invalid.
+
+ In this scenario all the tarball url gives 404 error.
+
+ """
+ for tarball in self.tarballs:
+ http_mocker.get(tarball['url'], text='Not Found', status_code=404)
+
+ fl = self.get_fl()
+ generator = fl.prepare_package_versions(self.tarballs)
+ package_versions = [data for data in generator]
+ assert package_versions == []
File Metadata
Details
Attached
Mime Type
text/plain
Expires
Thu, Jan 30, 3:53 PM (1 h, 44 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3225460
Attached To
D1744: swh.loader.base: Add tests
Event Timeline
Log In to Comment