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)) |