Page MenuHomeSoftware Heritage
Files F8395237

No OneTemporary
Actions

View Options

diff --git a/swh/loader/pypi/tests/test_loader.py b/swh/loader/pypi/tests/test_loader.py
index e2054d1..4aca3f5 100644
--- a/swh/loader/pypi/tests/test_loader.py
+++ b/swh/loader/pypi/tests/test_loader.py
@@ -1,258 +1,258 @@
# Copyright (C) 2016-2018 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 json
import shutil
import tempfile
from nose.plugins.attrib import attr
from nose.tools import istest
from unittest import TestCase
from swh.model import hashutil
from swh.loader.pypi.client import PyPIProject
from swh.loader.pypi.loader import PyPILoader
from .common import PyPIClientWithCache, RESOURCES_PATH, LoaderNoStorage
class TestPyPILoader(LoaderNoStorage, PyPILoader):
"""Real PyPILoader for test purposes (storage and pypi interactions
inhibited)
"""
def __init__(self, project_name):
project_metadata_file = '%s/%s.json' % (RESOURCES_PATH, project_name)
project_metadata_url = 'https://pypi.org/pypi/%s/json' % project_name
with open(project_metadata_file) as f:
data = json.load(f)
temp_dir = tempfile.mkdtemp(
dir='/tmp/', prefix='swh.loader.pypi.tests-')
# Will use the pypi with cache
client = PyPIClientWithCache(
temp_directory=temp_dir, cache_dir=RESOURCES_PATH)
super().__init__(client=client)
self.project = PyPIProject(
client=client,
project=project_name,
project_metadata_url=project_metadata_url,
data=data)
def prepare(self, project_name, origin_url,
origin_metadata_url=None):
self.project_name = project_name
self.origin_url = origin_url
self.origin_metadata_url = origin_metadata_url
self.visit = 1 # first visit
@attr('fs')
class BaseLoaderITest(TestCase):
"""Loader Test Mixin to prepare the pypi to 'load' in a test context.
In this setup, the loader uses the cache to load data so no
network interaction (no storage, no pypi).
"""
def setUp(self, project_name='0805nexter',
dummy_pypi_instance='https://dummy.org'):
self.tmp_root_path = tempfile.mkdtemp()
self.loader = PyPILoaderNoSnapshot(project_name=project_name)
self._project = project_name
self._origin_url = '%s/pypi/%s/' % (dummy_pypi_instance, project_name)
self._project_metadata_url = '%s/pypi/%s/json' % (
dummy_pypi_instance, project_name)
def tearDown(self):
shutil.rmtree(self.tmp_root_path)
def assertContentsOk(self, expected_contents):
contents = self.loader.all_contents
self.assertEquals(len(contents), len(expected_contents))
for content in contents:
content_id = hashutil.hash_to_hex(content['sha1'])
self.assertIn(content_id, expected_contents)
def assertDirectoriesOk(self, expected_directories):
directories = self.loader.all_directories
self.assertEquals(len(directories), len(expected_directories))
for _dir in directories:
_dir_id = hashutil.hash_to_hex(_dir['id'])
self.assertIn(_dir_id, expected_directories)
def assertSnapshotOk(self, expected_snapshot, expected_revisions):
snapshots = self.loader.all_snapshots
self.assertEqual(len(snapshots), 1)
snap = snapshots[0]
snap_id = hashutil.hash_to_hex(snap['id'])
self.assertEqual(snap_id, expected_snapshot)
branches = snap['branches']
self.assertEqual(len(expected_revisions), len(branches))
for branch, target in branches.items():
rev_id = hashutil.hash_to_hex(target['target'])
self.assertIn(rev_id, expected_revisions)
self.assertEqual('revision', target['target_type'])
def assertRevisionsOk(self, expected_revisions): # noqa: N802
"""Check the loader's revisions match the expected revisions.
Expects self.loader to be instantiated and ready to be
inspected (meaning the loading took place).
Args:
expected_revisions (dict): Dict with key revision id,
value the targeted directory id.
"""
# The last revision being the one used later to start back from
for rev in self.loader.all_revisions:
rev_id = hashutil.hash_to_hex(rev['id'])
directory_id = hashutil.hash_to_hex(rev['directory'])
self.assertEquals(expected_revisions[rev_id], directory_id)
# Define loaders with no storage
# They'll just accumulate the data in place
# Only for testing purposes.
class PyPILoaderNoSnapshot(TestPyPILoader):
"""Same as TestPyPILoader with no prior snapshot seen
"""
def _last_snapshot(self):
return None
class LoaderITest(BaseLoaderITest):
def setUp(self, project_name='0805nexter',
dummy_pypi_instance='https://dummy.org'):
super().setUp(project_name, dummy_pypi_instance)
self.loader = PyPILoaderNoSnapshot(project_name=project_name)
@istest
def load(self):
"""Load a pypi origin
"""
# when
self.loader.load(
self._project, self._origin_url, self._project_metadata_url)
# then
self.assertEquals(len(self.loader.all_contents), 6,
'3 contents per release artifact files (2)')
self.assertEquals(len(self.loader.all_directories), 4)
self.assertEquals(len(self.loader.all_revisions), 2,
'2 releases so 2 revisions should be created')
self.assertEquals(len(self.loader.all_releases), 0,
'No release is created in the pypi loader')
self.assertEquals(len(self.loader.all_snapshots), 1,
'Only 1 snapshot targetting all revisions')
expected_contents = [
'a61e24cdfdab3bb7817f6be85d37a3e666b34566',
'938c33483285fd8ad57f15497f538320df82aeb8',
'a27576d60e08c94a05006d2e6d540c0fdb5f38c8',
'405859113963cb7a797642b45f171d6360425d16',
'e5686aa568fdb1d19d7f1329267082fe40482d31',
'83ecf6ec1114fd260ca7a833a2d165e71258c338',
]
self.assertContentsOk(expected_contents)
expected_directories = [
'05219ba38bc542d4345d5638af1ed56c7d43ca7d',
'cf019eb456cf6f78d8c4674596f1c9a97ece8f44',
'b178b66bd22383d5f16f4f5c923d39ca798861b4',
'c3a58f8b57433a4b56caaa5033ae2e0931405338',
]
self.assertDirectoriesOk(expected_directories)
# {revision hash: directory hash}
expected_revisions = {
'4c99891f93b81450385777235a37b5e966dd1571': '05219ba38bc542d4345d5638af1ed56c7d43ca7d', # noqa
'e445da4da22b31bfebb6ffc4383dbf839a074d21': 'b178b66bd22383d5f16f4f5c923d39ca798861b4', # noqa
}
self.assertRevisionsOk(expected_revisions)
self.assertSnapshotOk('f456b03e8bf1920d64b00df234b1efedc25b6c93',
expected_revisions)
class PyPILoaderWithSnapshot(TestPyPILoader):
"""Same as TestPyPILoader with no prior snapshot seen
"""
def _last_snapshot(self):
return {
- 'id': b'\xf4V\xb0>\x8b\xf1\x92\rd\xb0\r\xf24\xb1\xef\xed\xc2[l\x93',
+ 'id': b'\xf4V\xb0>\x8b\xf1\x92\rd\xb0\r\xf24\xb1\xef\xed\xc2[l\x93', # noqa
'branches': {
b'0805nexter-1.1.0.zip': {
'target': b'L\x99\x89\x1f\x93\xb8\x14P'
b'8Ww#Z7\xb5\xe9f\xdd\x15q',
'target_type': 'revision'
},
b'0805nexter-1.2.0.zip': {
'target': b'\xe4E\xdaM\xa2+1\xbf'
b'\xeb\xb6\xff\xc48=\xbf\x83'
b'\x9a\x07M!',
'target_type': 'revision'
},
},
}
def _known_releases(self, last_snapshot):
yield from [
(
'0805nexter-1.1.0.zip',
'52cd128ad3afe539478abc7440d4b043384295fbe6b0958a237cb6d926465035' # noqa
),
(
'0805nexter-1.2.0.zip',
'49785c6ae39ea511b3c253d7621c0b1b6228be2f965aca8a491e6b84126d0709' # noqa
)
]
class LoaderWithOriginAlreadySeenITest(BaseLoaderITest):
def setUp(self, project_name='0805nexter',
dummy_pypi_instance='https://dummy.org'):
super().setUp(project_name, dummy_pypi_instance)
self.loader = PyPILoaderWithSnapshot(project_name=project_name)
@istest
def load(self):
"""Load a pypi origin already injected will result with only 1 snapshot
"""
# when
self.loader.load(
self._project, self._origin_url, self._project_metadata_url)
# then
self.assertEquals(len(self.loader.all_contents), 0)
self.assertEquals(len(self.loader.all_directories), 0)
self.assertEquals(len(self.loader.all_revisions), 0)
self.assertEquals(len(self.loader.all_releases), 0)
self.assertEquals(len(self.loader.all_snapshots), 1)
self.assertContentsOk([])
self.assertDirectoriesOk([])
self.assertRevisionsOk(expected_revisions={})
expected_revisions = {
'4c99891f93b81450385777235a37b5e966dd1571': '05219ba38bc542d4345d5638af1ed56c7d43ca7d', # noqa
'e445da4da22b31bfebb6ffc4383dbf839a074d21': 'b178b66bd22383d5f16f4f5c923d39ca798861b4', # noqa
}
self.assertSnapshotOk('f456b03e8bf1920d64b00df234b1efedc25b6c93',
expected_revisions)

File Metadata

Mime Type
text/x-diff
Expires
Jun 4 2025, 7:36 PM (10 w, 7 h ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3399256

Event Timeline

About · Developer information · Legal