Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/mercurial/tests/test_from_disk.py
- This file was added.
import json | |||||
marmoute: This need a module documentation. | |||||
Done Inline ActionsWhere does this test comes from ? Is this a copy of the test for the older loader ? If so, what re the difference ? marmoute: Where does this test comes from ? Is this a copy of the test for the older loader ? If so, what… | |||||
import os | |||||
from swh.loader.core.tests import BaseLoaderTest | |||||
from swh.loader.mercurial.from_disk import HgLoaderFromDisk | |||||
from swh.model.hashutil import hash_to_bytes | |||||
from swh.model.identifiers import snapshot_identifier | |||||
class TestLoader(HgLoaderFromDisk): | |||||
def parse_config_file(self, *args, **kwargs): | |||||
return { | |||||
"storage": { | |||||
"cls": "pipeline", | |||||
"steps": [{"cls": "filter"}, {"cls": "memory"},], | |||||
}, | |||||
Done Inline Actionsplease document the testcase marmoute: please document the testcase | |||||
"max_content_size": 100 * 1024 * 1024, | |||||
"save_data": False, | |||||
} | |||||
class BaseHgLoaderFromDiskTest(BaseLoaderTest): | |||||
def setUp(self, archive_name, filename=None): | |||||
super().setUp( | |||||
archive_name=archive_name, | |||||
filename=filename, | |||||
prefix_tmp_folder_name="swh.loader.mercurial.", | |||||
start_path=os.path.dirname(__file__), | |||||
Done Inline Actionswhere does this hash comes from. marmoute: where does this hash comes from. | |||||
uncompress_archive=False, | |||||
) | |||||
self.loader = TestLoader(url=self.repo_url, directory=self.destination_path,) | |||||
self.storage = self.loader.storage | |||||
self.repo = self.destination_path | |||||
def load(self): | |||||
return self.loader.load() | |||||
class HgLoaderFromDiskTests(BaseHgLoaderFromDiskTest): | |||||
def setUp(self): | |||||
super().setUp(archive_name="example/repo") | |||||
self._expected = None | |||||
def expected(self): | |||||
if self._expected is None: | |||||
path = os.path.join(os.path.dirname(self.destination_path), "expected.json") | |||||
self._expected = json.load(open(path, "r")) | |||||
return self._expected | |||||
def expected_content(self): | |||||
return [ | |||||
obj["id"] | |||||
for obj in self.expected()["objects"].values() | |||||
if obj["type"] == "blob" | |||||
] | |||||
def expected_dir_count(self): | |||||
return sum( | |||||
[1 for obj in self.expected()["objects"].values() if obj["type"] == "tree"] | |||||
) | |||||
def expected_revisions(self): | |||||
return { | |||||
obj["id"]: obj["tree"] | |||||
for obj in self.expected()["objects"].values() | |||||
if obj["type"] == "commit" | |||||
} | |||||
def expected_snapshot(self): | |||||
result = {"branches": self.expected()["branches"]} | |||||
result["id"] = snapshot_identifier( | |||||
{ | |||||
"branches": { | |||||
name.encode(): { | |||||
"target": branch["target"].encode() | |||||
if name == "HEAD" | |||||
else hash_to_bytes(branch["target"]), | |||||
"target_type": branch["target_type"], | |||||
} | |||||
for name, branch in result["branches"].items() | |||||
} | |||||
} | |||||
) | |||||
return result | |||||
def assertContentsContainSha1Git(self, expected): | |||||
self.assertEqual( | |||||
list( | |||||
self.storage.content_missing( | |||||
[{"sha1_git": hash_to_bytes(sha1_git)} for sha1_git in expected] | |||||
) | |||||
), | |||||
[], | |||||
) | |||||
def test_load(self): | |||||
result = self.load() | |||||
self.assertEqual(result["status"], "eventful", result) | |||||
self.assertContentsContainSha1Git(self.expected_content()) | |||||
self.assertCountDirectories(self.expected_dir_count()) | |||||
revisions = self.expected_revisions() | |||||
self.assertCountRevisions(len(revisions)) | |||||
self.assertRevisionsContain(revisions) | |||||
snapshot = self.expected_snapshot() | |||||
self.assertCountSnapshots(1) | |||||
self.assertSnapshotEqual(snapshot) | |||||
self.assertCountReleases(0) | |||||
Done Inline ActionsWhat is going on here, and why is it correct ? marmoute: What is going on here, and why is it correct ? | |||||
self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | |||||
self.assertEqual(self.loader.visit_status(), "full") | |||||
visit = self.storage.origin_visit_get_latest(self.repo_url) | |||||
self.assertEqual(visit["snapshot"], hash_to_bytes(snapshot["id"])) | |||||
self.assertEqual(visit["status"], "full") |
This need a module documentation.