Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/conftest.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 | ||||
| import glob | import glob | ||||
| from os import path | from os import path | ||||
| import re | import re | ||||
| from typing import Iterable, Iterator, List | from typing import Iterable, Iterator, List, Optional | ||||
| import pytest | import pytest | ||||
| from typing_extensions import TypedDict | from typing_extensions import TypedDict | ||||
| from swh.core.api.serializers import msgpack_loads | from swh.core.api.serializers import msgpack_loads | ||||
| from swh.core.db import BaseDb | from swh.core.db import BaseDb | ||||
| from swh.core.db.pytest_plugin import postgresql_fact | from swh.core.db.pytest_plugin import postgresql_fact | ||||
| from swh.core.utils import numfile_sortkey as sortkey | from swh.core.utils import numfile_sortkey as sortkey | ||||
| ▲ Show 20 Lines • Show All 173 Lines • ▼ Show 20 Lines | storage.directory_add( | ||||
| ) | ) | ||||
| for dir in data["directory"] | for dir in data["directory"] | ||||
| ] | ] | ||||
| ) | ) | ||||
| storage.revision_add(Revision.from_dict(revision) for revision in data["revision"]) | storage.revision_add(Revision.from_dict(revision) for revision in data["revision"]) | ||||
| class SynthRelation(TypedDict): | class SynthRelation(TypedDict): | ||||
| prefix: Optional[str] | |||||
| path: str | path: str | ||||
| src: bytes | src: bytes | ||||
| dst: bytes | dst: bytes | ||||
| rel_ts: float | rel_ts: float | ||||
| class SynthRevision(TypedDict): | class SynthRevision(TypedDict): | ||||
| sha1: bytes | sha1: bytes | ||||
| ▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines | def _mk_synth_rev(synth_rev) -> SynthRevision: | ||||
| rev = SynthRevision( | rev = SynthRevision( | ||||
| sha1=bytes.fromhex(synth_rev[0]["sha1"]), | sha1=bytes.fromhex(synth_rev[0]["sha1"]), | ||||
| date=float(synth_rev[0]["ts"]), | date=float(synth_rev[0]["ts"]), | ||||
| msg=synth_rev[0]["revname"], | msg=synth_rev[0]["revname"], | ||||
| R_C=[], | R_C=[], | ||||
| R_D=[], | R_D=[], | ||||
| D_C=[], | D_C=[], | ||||
| ) | ) | ||||
| current_path = None | |||||
| # path of the last R-D relation we parsed, used a prefix for next D-C | |||||
| # relations | |||||
| for row in synth_rev[1:]: | for row in synth_rev[1:]: | ||||
| if row["reltype"] == "R---C": | if row["reltype"] == "R---C": | ||||
| assert row["type"] == "C" | assert row["type"] == "C" | ||||
| rev["R_C"].append( | rev["R_C"].append( | ||||
| SynthRelation( | SynthRelation( | ||||
| prefix=None, | |||||
| path=row["path"], | path=row["path"], | ||||
| src=rev["sha1"], | src=rev["sha1"], | ||||
| dst=bytes.fromhex(row["sha1"]), | dst=bytes.fromhex(row["sha1"]), | ||||
| rel_ts=float(row["ts"]), | rel_ts=float(row["ts"]), | ||||
| ) | ) | ||||
| ) | ) | ||||
| current_path = None | |||||
| elif row["reltype"] == "R-D": | elif row["reltype"] == "R-D": | ||||
| assert row["type"] == "D" | assert row["type"] == "D" | ||||
| rev["R_D"].append( | rev["R_D"].append( | ||||
| SynthRelation( | SynthRelation( | ||||
| prefix=None, | |||||
| path=row["path"], | path=row["path"], | ||||
| src=rev["sha1"], | src=rev["sha1"], | ||||
| dst=bytes.fromhex(row["sha1"]), | dst=bytes.fromhex(row["sha1"]), | ||||
| rel_ts=float(row["ts"]), | rel_ts=float(row["ts"]), | ||||
| ) | ) | ||||
| ) | ) | ||||
| current_path = row["path"] | |||||
| elif row["reltype"] == "D-C": | elif row["reltype"] == "D-C": | ||||
| assert row["type"] == "C" | assert row["type"] == "C" | ||||
| rev["D_C"].append( | rev["D_C"].append( | ||||
| SynthRelation( | SynthRelation( | ||||
| prefix=current_path, | |||||
| path=row["path"], | path=row["path"], | ||||
| src=rev["R_D"][-1]["dst"], | src=rev["R_D"][-1]["dst"], | ||||
| dst=bytes.fromhex(row["sha1"]), | dst=bytes.fromhex(row["sha1"]), | ||||
| rel_ts=float(row["ts"]), | rel_ts=float(row["ts"]), | ||||
| ) | ) | ||||
| ) | ) | ||||
| return rev | return rev | ||||