Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/mercurial/tests/test_loader.py
| Show All 17 Lines | |||||
| from .common import HgLoaderMemoryStorage, HgArchiveLoaderMemoryStorage | from .common import HgLoaderMemoryStorage, HgArchiveLoaderMemoryStorage | ||||
| from ..loader import HgBundle20Loader, CloneTimeoutError | from ..loader import HgBundle20Loader, CloneTimeoutError | ||||
| class BaseHgLoaderTest(BaseLoaderTest): | class BaseHgLoaderTest(BaseLoaderTest): | ||||
| """Mixin base loader test to prepare the mercurial | """Mixin base loader test to prepare the mercurial | ||||
| repository to uncompress, load and test the results. | repository to uncompress, load and test the results. | ||||
| This sets up | |||||
| """ | """ | ||||
| def setUp(self, loader=HgLoaderMemoryStorage, | def setUp(self, archive_name='the-sandbox.tgz', filename='the-sandbox', | ||||
| archive_name='the-sandbox.tgz', filename='the-sandbox', | |||||
| uncompress_archive=True): | uncompress_archive=True): | ||||
| super().setUp(archive_name=archive_name, filename=filename, | super().setUp(archive_name=archive_name, filename=filename, | ||||
| prefix_tmp_folder_name='swh.loader.mercurial.', | prefix_tmp_folder_name='swh.loader.mercurial.', | ||||
| start_path=os.path.dirname(__file__), | start_path=os.path.dirname(__file__), | ||||
| uncompress_archive=uncompress_archive) | uncompress_archive=uncompress_archive) | ||||
| self.loader = loader() | |||||
| self.storage = self.loader.storage | |||||
| class WithoutReleaseLoaderTest(BaseHgLoaderTest): | class WithoutReleaseLoaderTest(BaseHgLoaderTest): | ||||
| """Load a mercurial repository without release | """Load a mercurial repository without release | ||||
| """ | """ | ||||
| def setUp(self, *args, **kwargs): | |||||
| super().setUp(*args, **kwargs) | |||||
| self.loader = HgLoaderMemoryStorage( | |||||
| url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| directory=self.destination_path) | |||||
| self.storage = self.loader.storage | |||||
| def test_load(self): | def test_load(self): | ||||
| """Load a repository with multiple branches results in 1 snapshot | """Load a repository with multiple branches results in 1 snapshot | ||||
| Another visit with no change in between result in uneventful visit | |||||
| """ | """ | ||||
| # when | # when | ||||
| self.loader.load( | self.loader.load() | ||||
| origin_url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| directory=self.destination_path) | |||||
| # then | # then | ||||
| self.assertCountContents(2) | self.assertCountContents(2) | ||||
| self.assertCountDirectories(3) | self.assertCountDirectories(3) | ||||
| self.assertCountReleases(0) | self.assertCountReleases(0) | ||||
| self.assertCountRevisions(58) | self.assertCountRevisions(58) | ||||
| tip_revision_develop = 'a9c4534552df370f43f0ef97146f393ef2f2a08c' | tip_revision_develop = 'a9c4534552df370f43f0ef97146f393ef2f2a08c' | ||||
| ▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Lines | def test_load(self): | ||||
| } | } | ||||
| } | } | ||||
| } | } | ||||
| self.assertSnapshotEqual(expected_snapshot) | self.assertSnapshotEqual(expected_snapshot) | ||||
| self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) | self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) | ||||
| self.assertEqual(self.loader.visit_status(), 'full') | self.assertEqual(self.loader.visit_status(), 'full') | ||||
| def test_load_status(self): | |||||
| # first visit of the mercurial repository | |||||
| self.loader.load( | |||||
| origin_url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| directory=self.destination_path) | |||||
| self.assertEqual(self.loader.load_status(), {'status': 'eventful'}) | |||||
| self.assertEqual(self.loader.visit_status(), 'full') | |||||
| # second visit with no changes in the mercurial repository | # second visit with no changes in the mercurial repository | ||||
| # since the first one | # since the first one | ||||
| self.loader.load( | self.loader.load() | ||||
| origin_url=self.repo_url, | |||||
| visit_date='2016-05-04 14:12:21+00', | |||||
| directory=self.destination_path) | |||||
| self.assertEqual(self.loader.load_status(), {'status': 'uneventful'}) | self.assertEqual(self.loader.load_status(), {'status': 'uneventful'}) | ||||
| self.assertEqual(self.loader.visit_status(), 'full') | self.assertEqual(self.loader.visit_status(), 'full') | ||||
| class CommonHgLoaderData: | class CommonHgLoaderData: | ||||
| def assert_data_ok(self): | def assert_data_ok(self): | ||||
| # then | # then | ||||
| ▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | |||||
| class WithReleaseLoaderTest(BaseHgLoaderTest, CommonHgLoaderData): | class WithReleaseLoaderTest(BaseHgLoaderTest, CommonHgLoaderData): | ||||
| """Load a mercurial repository with release | """Load a mercurial repository with release | ||||
| """ | """ | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp(archive_name='hello.tgz', filename='hello') | super().setUp(archive_name='hello.tgz', filename='hello') | ||||
| self.loader = HgLoaderMemoryStorage( | |||||
| url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| directory=self.destination_path | |||||
| ) | |||||
| self.storage = self.loader.storage | |||||
| def test_load(self): | def test_load(self): | ||||
| """Load a repository with tags results in 1 snapshot | """Load a repository with tags results in 1 snapshot | ||||
| """ | """ | ||||
| # when | # when | ||||
| self.loader.load( | self.loader.load() | ||||
| origin_url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| directory=self.destination_path) | |||||
| self.assert_data_ok() | self.assert_data_ok() | ||||
| class ArchiveLoaderTest(BaseHgLoaderTest, CommonHgLoaderData): | class ArchiveLoaderTest(BaseHgLoaderTest, CommonHgLoaderData): | ||||
| """Load a mercurial repository archive with release | """Load a mercurial repository archive with release | ||||
| """ | """ | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp(loader=HgArchiveLoaderMemoryStorage, | super().setUp(archive_name='hello.tgz', filename='hello', | ||||
| archive_name='hello.tgz', filename='hello', | |||||
| uncompress_archive=False) | uncompress_archive=False) | ||||
| self.loader = HgArchiveLoaderMemoryStorage( | |||||
| url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| archive_path=self.destination_path | |||||
| ) | |||||
| self.storage = self.loader.storage | |||||
| def test_load(self): | def test_load(self): | ||||
| """Load a mercurial repository archive with tags results in 1 snapshot | """Load a mercurial repository archive with tags results in 1 snapshot | ||||
| """ | """ | ||||
| # when | # when | ||||
| self.loader.load( | self.loader.load() | ||||
| origin_url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| archive_path=self.destination_path) | |||||
| self.assert_data_ok() | self.assert_data_ok() | ||||
| @patch('swh.loader.mercurial.archive_extract.patoolib') | @patch('swh.loader.mercurial.archive_extract.patoolib') | ||||
| def test_load_with_failure(self, mock_patoo): | def test_load_with_failure(self, mock_patoo): | ||||
| mock_patoo.side_effect = ValueError | mock_patoo.side_effect = ValueError | ||||
| # when | # when | ||||
| r = self.loader.load( | r = self.loader.load() | ||||
| origin_url=self.repo_url, | |||||
| visit_date='2016-05-03 15:16:32+00', | |||||
| archive_path=self.destination_path) | |||||
| self.assertEqual(r, {'status': 'failed'}) | self.assertEqual(r, {'status': 'failed'}) | ||||
| self.assertCountContents(0) | self.assertCountContents(0) | ||||
| self.assertCountDirectories(0) | self.assertCountDirectories(0) | ||||
| self.assertCountRevisions(0) | self.assertCountRevisions(0) | ||||
| self.assertCountReleases(0) | self.assertCountReleases(0) | ||||
| self.assertCountSnapshots(0) | self.assertCountSnapshots(0) | ||||
| class WithTransplantLoaderTest(BaseHgLoaderTest): | class WithTransplantLoaderTest(BaseHgLoaderTest): | ||||
| """Load a mercurial repository where transplant operations | """Load a mercurial repository where transplant operations | ||||
| have been used. | have been used. | ||||
| """ | """ | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp(archive_name='transplant.tgz', filename='transplant') | super().setUp(archive_name='transplant.tgz', filename='transplant') | ||||
| self.loader = HgLoaderMemoryStorage( | |||||
| url=self.repo_url, | |||||
| visit_date='2019-05-23 12:06:00+00', | |||||
| directory=self.destination_path | |||||
| ) | |||||
| self.storage = self.loader.storage | |||||
| def test_load(self): | def test_load(self): | ||||
| # load hg repository | # load hg repository | ||||
| self.loader.load( | self.loader.load() | ||||
| origin_url=self.repo_url, | |||||
| visit_date='2019-05-23 12:06:00+00', | |||||
| directory=self.destination_path) | |||||
| # collect swh revisions | # collect swh revisions | ||||
| origin_url = self.storage.origin_get([ | origin_url = self.storage.origin_get([ | ||||
| {'type': 'hg', 'url': self.repo_url}])[0]['url'] | {'type': 'hg', 'url': self.repo_url}])[0]['url'] | ||||
| visit = self.storage.origin_visit_get_latest( | visit = self.storage.origin_visit_get_latest( | ||||
| origin_url, require_snapshot=True) | origin_url, require_snapshot=True) | ||||
| revisions = [] | revisions = [] | ||||
| snapshot = snapshot_get_all_branches(self.storage, visit['snapshot']) | snapshot = snapshot_get_all_branches(self.storage, visit['snapshot']) | ||||
| ▲ Show 20 Lines • Show All 67 Lines • Show Last 20 Lines | |||||