Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/test_isochrone_graph.py
# Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from copy import deepcopy | from copy import deepcopy | ||||
from datetime import datetime, timezone | from datetime import datetime, timezone | ||||
from typing import Any, Dict | |||||
import pytest | import pytest | ||||
import yaml | import yaml | ||||
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.graph import IsochroneNode, build_isochrone_graph | from swh.provenance.graph import IsochroneNode, build_isochrone_graph | ||||
from swh.provenance.model import DirectoryEntry, RevisionEntry | from swh.provenance.model import DirectoryEntry, RevisionEntry | ||||
from swh.provenance.provenance import ProvenanceInterface | |||||
from swh.provenance.revision import revision_add | from swh.provenance.revision import revision_add | ||||
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.provenance.tests.test_provenance_db import ts2dt | from swh.provenance.tests.test_provenance_db import ts2dt | ||||
from swh.storage.interface import StorageInterface | |||||
def isochrone_graph_from_dict(d, depth=0) -> IsochroneNode: | def isochrone_graph_from_dict(d: Dict[str, Any], depth: int = 0) -> IsochroneNode: | ||||
"""Takes a dictionary representing a tree of IsochroneNode objects, and | """Takes a dictionary representing a tree of IsochroneNode objects, and | ||||
recursively builds the corresponding graph.""" | recursively builds the corresponding graph.""" | ||||
d = deepcopy(d) | d = deepcopy(d) | ||||
d["entry"]["id"] = hash_to_bytes(d["entry"]["id"]) | d["entry"]["id"] = hash_to_bytes(d["entry"]["id"]) | ||||
d["entry"]["name"] = bytes(d["entry"]["name"], encoding="utf-8") | d["entry"]["name"] = bytes(d["entry"]["name"], encoding="utf-8") | ||||
dbdate = d.get("dbdate", None) | dbdate = d.get("dbdate", None) | ||||
Show All 24 Lines | ( | ||||
("cmdbts2", False, 1), | ("cmdbts2", False, 1), | ||||
("cmdbts2", True, 2), | ("cmdbts2", True, 2), | ||||
("cmdbts2", False, 2), | ("cmdbts2", False, 2), | ||||
("out-of-order", True, 1), | ("out-of-order", True, 1), | ||||
), | ), | ||||
) | ) | ||||
@pytest.mark.parametrize("batch", (True, False)) | @pytest.mark.parametrize("batch", (True, False)) | ||||
def test_isochrone_graph( | def test_isochrone_graph( | ||||
provenance, swh_storage, archive, repo, lower, mindepth, batch | provenance: ProvenanceInterface, | ||||
): | swh_storage: StorageInterface, | ||||
archive: ArchiveInterface, | |||||
repo: str, | |||||
lower: bool, | |||||
mindepth: int, | |||||
batch: bool, | |||||
) -> 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(swh_storage, data) | ||||
revisions = {rev["id"]: rev for rev in data["revision"]} | revisions = {rev["id"]: rev for rev in data["revision"]} | ||||
filename = f"graphs_{repo}_{'lower' if lower else 'upper'}_{mindepth}.yaml" | filename = f"graphs_{repo}_{'lower' if lower else 'upper'}_{mindepth}.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): | ||||
print("# Processing revision", expected["rev"]) | print("# Processing revision", expected["rev"]) | ||||
revision = revisions[hash_to_bytes(expected["rev"])] | revision = revisions[hash_to_bytes(expected["rev"])] | ||||
entry = RevisionEntry( | entry = RevisionEntry( | ||||
id=revision["id"], | id=revision["id"], | ||||
date=ts2dt(revision["date"]), | date=ts2dt(revision["date"]), | ||||
root=revision["directory"], | root=revision["directory"], | ||||
) | ) | ||||
expected_graph = isochrone_graph_from_dict(expected["graph"]) | expected_graph = isochrone_graph_from_dict(expected["graph"]) | ||||
print("Expected graph:", expected_graph) | print("Expected graph:", expected_graph) | ||||
# Create graph for current revision and check it has the expected structure. | # Create graph for current revision and check it has the expected structure. | ||||
assert entry.root is not None | |||||
computed_graph = build_isochrone_graph( | computed_graph = build_isochrone_graph( | ||||
archive, | archive, | ||||
provenance, | provenance, | ||||
entry, | entry, | ||||
DirectoryEntry(entry.root), | DirectoryEntry(entry.root), | ||||
) | ) | ||||
print("Computed graph:", computed_graph) | print("Computed graph:", computed_graph) | ||||
assert computed_graph == expected_graph | assert computed_graph == expected_graph | ||||
Show All 11 Lines |