Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/mercurial/tests/test_loader.py
| Show All 23 Lines | def setUp(self, loader=HgLoaderMemoryStorage, | ||||
| 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.loader = loader() | ||||
| self.storage = self.loader.storage | self.storage = self.loader.storage | ||||
| class LoaderTestT1(BaseHgLoaderTest): | class WithoutReleaseLoaderTest(BaseHgLoaderTest): | ||||
| """Load a mercurial repository without release | """Load a mercurial repository without release | ||||
| """ | """ | ||||
| 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 | ||||
| """ | """ | ||||
| # when | # when | ||||
| ▲ Show 20 Lines • Show All 144 Lines • ▼ Show 20 Lines | def assert_data_ok(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') | ||||
| class LoaderTest2(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') | ||||
| 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 | ||||
| Show All 40 Lines | def test_load_with_failure(self, mock_patoo): | ||||
| archive_path=self.destination_path) | 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): | |||||
| """Load a mercurial repository where transplant operations | |||||
| have been used. | |||||
| """ | |||||
| def setUp(self): | |||||
| super().setUp(archive_name='transplant.tgz', filename='transplant') | |||||
| def test_load(self): | |||||
| # load hg repository | |||||
| self.loader.load( | |||||
| origin_url=self.repo_url, | |||||
| visit_date='2019-05-23 12:06:00+00', | |||||
| directory=self.destination_path) | |||||
| # collect swh revisions | |||||
| origin_id = self.storage.origin_get([ | |||||
| {'type': 'hg', 'url': self.repo_url}])[0]['id'] | |||||
| snapshot = self.storage.snapshot_get_latest(origin_id) | |||||
| revisions = [] | |||||
| for branch in snapshot['branches'].values(): | |||||
| if branch['target_type'] != 'revision': | |||||
| continue | |||||
| revisions.append(branch['target']) | |||||
| # extract original changesets info and the transplant sources | |||||
| hg_changesets = set() | |||||
| transplant_sources = set() | |||||
| for rev in self.storage.revision_log(revisions): | |||||
| hg_changesets.add(rev['metadata']['node']) | |||||
| for k, v in rev['metadata']['extra_headers']: | |||||
| if k == 'transplant_source': | |||||
| transplant_sources.add(v) | |||||
| # check extracted data are valid | |||||
| self.assertTrue(len(hg_changesets) > 0) | |||||
| self.assertTrue(len(transplant_sources) > 0) | |||||
| self.assertTrue(transplant_sources.issubset(hg_changesets)) | |||||