Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/test_history_graph.py
| Show All 9 Lines | |||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.provenance.archive import ArchiveInterface | from swh.provenance.archive import ArchiveInterface | ||||
| from swh.provenance.graph import HistoryNode, build_history_graph | from swh.provenance.graph import HistoryNode, build_history_graph | ||||
| from swh.provenance.interface import ProvenanceInterface | from swh.provenance.interface import ProvenanceInterface | ||||
| from swh.provenance.model import OriginEntry, RevisionEntry | from swh.provenance.model import OriginEntry, RevisionEntry | ||||
| from swh.provenance.origin import origin_add_revision | from swh.provenance.origin import origin_add_revision | ||||
| from swh.provenance.tests.conftest import fill_storage, get_datafile, load_repo_data | from swh.provenance.tests.conftest import fill_storage, get_datafile, load_repo_data | ||||
| from swh.storage.postgresql.storage import Storage | |||||
| def history_graph_from_dict(d: Dict[str, Any]) -> HistoryNode: | def history_graph_from_dict(d: Dict[str, Any]) -> HistoryNode: | ||||
| """Takes a dictionary representing a tree of HistoryNode objects, and | """Takes a dictionary representing a tree of HistoryNode objects, and | ||||
| recursively builds the corresponding graph.""" | recursively builds the corresponding graph.""" | ||||
| node = HistoryNode( | node = HistoryNode( | ||||
| entry=RevisionEntry(hash_to_bytes(d["rev"])), | entry=RevisionEntry(hash_to_bytes(d["rev"])), | ||||
| visited=d.get("visited", False), | visited=d.get("visited", False), | ||||
| in_history=d.get("in_history", False), | in_history=d.get("in_history", False), | ||||
| ) | ) | ||||
| node.parents = set( | node.parents = set( | ||||
| history_graph_from_dict(parent) for parent in d.get("parents", []) | history_graph_from_dict(parent) for parent in d.get("parents", []) | ||||
| ) | ) | ||||
| return node | return node | ||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
| "repo, visit", | "repo, visit", | ||||
| (("with-merges", "visits-01"),), | (("with-merges", "visits-01"),), | ||||
| ) | ) | ||||
| @pytest.mark.parametrize("batch", (True, False)) | @pytest.mark.parametrize("batch", (True, False)) | ||||
| def test_history_graph( | def test_history_graph( | ||||
| provenance: ProvenanceInterface, | provenance: ProvenanceInterface, | ||||
| swh_storage: Storage, | |||||
| archive: ArchiveInterface, | archive: ArchiveInterface, | ||||
| repo: str, | repo: str, | ||||
| visit: str, | visit: str, | ||||
| batch: bool, | batch: bool, | ||||
| ) -> None: | ) -> None: | ||||
| # read data/README.md for more details on how these datasets are generated | # read data/README.md for more details on how these datasets are generated | ||||
| data = load_repo_data(repo) | data = load_repo_data(repo) | ||||
| fill_storage(swh_storage, data) | fill_storage(archive.storage, data) | ||||
| filename = f"history_graphs_{repo}_{visit}.yaml" | filename = f"history_graphs_{repo}_{visit}.yaml" | ||||
| with open(get_datafile(filename)) as file: | with open(get_datafile(filename)) as file: | ||||
| for expected in yaml.full_load(file): | for expected in yaml.full_load(file): | ||||
| entry = OriginEntry(expected["origin"], hash_to_bytes(expected["snapshot"])) | entry = OriginEntry(expected["origin"], hash_to_bytes(expected["snapshot"])) | ||||
| provenance.origin_add(entry) | provenance.origin_add(entry) | ||||
| Show All 16 Lines | |||||