Differential D5891 Diff 21165 swh/provenance/tests/conftest.py

Changeset 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

from os import path		from os import path
import re		import re
from typing import Iterable, Iterator, List, Optional		from typing import Iterable, Iterator, List, Optional

		import msgpack
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.db import BaseDb		from swh.core.db import BaseDb
from swh.model.model import Content, Directory, DirectoryEntry, Revision		from swh.journal.serializers import msgpack_ext_hook
from swh.model.tests.swh_model_data import TEST_OBJECTS		from swh.model.tests.swh_model_data import TEST_OBJECTS
from swh.provenance.postgresql.archive import ArchivePostgreSQL		from swh.provenance.postgresql.archive import ArchivePostgreSQL
from swh.provenance.storage.archive import ArchiveStorage		from swh.provenance.storage.archive import ArchiveStorage
		from swh.storage.replay import process_replay_objects


@pytest.fixture(params=["with-path", "without-path"])		@pytest.fixture(params=["with-path", "without-path"])
def provenance(request, postgresql):		def provenance(request, postgresql):
"""return a working and initialized provenance db"""		"""return a working and initialized provenance db"""
from swh.core.cli.db import populate_database_for_package		from swh.core.cli.db import populate_database_for_package

flavor = request.param		flavor = request.param
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Lines	else:
yield ArchiveStorage(swh_storage_with_objects)		yield ArchiveStorage(swh_storage_with_objects)


def get_datafile(fname):		def get_datafile(fname):
return path.join(path.dirname(__file__), "data", fname)		return path.join(path.dirname(__file__), "data", fname)


def load_repo_data(repo):		def load_repo_data(repo):
data = {"revision": [], "directory": [], "content": []}		data = {}
with open(get_datafile(f"{repo}.msgpack"), "rb") as fobj:		with open(get_datafile(f"{repo}.msgpack"), "rb") as fobj:
for etype, value in msgpack_loads(fobj.read()):		unpacker = msgpack.Unpacker(
data[etype].append(value)		fobj,
		raw=False,
		ext_hook=msgpack_ext_hook,
		strict_map_key=False,
		timestamp=3, # convert Timestamp in datetime objects (tz UTC)
		)
		for objtype, objd in unpacker:
		data.setdefault(objtype, []).append(objd)
return data		return data


def filter_dict(d, keys):		def filter_dict(d, keys):
return {k: v for (k, v) in d.items() if k in keys}		return {k: v for (k, v) in d.items() if k in keys}


def fill_storage(storage, data):		def fill_storage(storage, data):
storage.content_add_metadata(		process_replay_objects(data, storage=storage)
Content.from_dict(content) for content in data["content"]
)
storage.directory_add(
[
Directory(
entries=tuple(
[
DirectoryEntry.from_dict(
filter_dict(entry, ("name", "type", "target", "perms"))
)
for entry in dir["entries"]
]
)
)
for dir in data["directory"]
]
)
storage.revision_add(Revision.from_dict(revision) for revision in data["revision"])


class SynthRelation(TypedDict):		class SynthRelation(TypedDict):
prefix: Optional[str]		prefix: Optional[str]
path: str		path: str
src: bytes		src: bytes
dst: bytes		dst: bytes
rel_ts: float		rel_ts: float
▲ Show 20 Lines • Show All 114 Lines • Show Last 20 Lines