Differential D6463 Diff 23490 swh/web/tests/conftest.py

Changeset View

Standalone View

swh/web/tests/conftest.py

Show All 34 Lines
from swh.storage.algos.revisions_walker import get_revisions_walker		from swh.storage.algos.revisions_walker import get_revisions_walker
from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest		from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest
from swh.web.auth.utils import OIDC_SWH_WEB_CLIENT_ID		from swh.web.auth.utils import OIDC_SWH_WEB_CLIENT_ID
from swh.web.common import converters		from swh.web.common import converters
from swh.web.common.origin_save import get_scheduler_load_task_types		from swh.web.common.origin_save import get_scheduler_load_task_types
from swh.web.common.typing import OriginVisitInfo		from swh.web.common.typing import OriginVisitInfo
from swh.web.common.utils import browsers_supported_image_mimes		from swh.web.common.utils import browsers_supported_image_mimes
from swh.web.config import get_config		from swh.web.config import get_config
from swh.web.tests.data import get_tests_data, override_storages		from swh.web.tests.data import (
		get_tests_data,
		override_storages,
		random_content,
		random_sha1,
		random_sha256,
		)

# Used to skip some tests		# Used to skip some tests
ctags_json_missing = (		ctags_json_missing = (
shutil.which("ctags") is None		shutil.which("ctags") is None
or b"+json" not in run(["ctags", "--version"], stdout=PIPE).stdout		or b"+json" not in run(["ctags", "--version"], stdout=PIPE).stdout
)		)

fossology_missing = shutil.which("nomossa") is None		fossology_missing = shutil.which("nomossa") is None
▲ Show 20 Lines • Show All 109 Lines • ▼ Show 20 Lines	def tests_data():
# Update swh-web configuration to use the in-memory storages		# Update swh-web configuration to use the in-memory storages
# instantiated in the tests.data module		# instantiated in the tests.data module
override_storages(		override_storages(
data["storage"], data["idx_storage"], data["search"], data["counters"]		data["storage"], data["idx_storage"], data["search"], data["counters"]
)		)
return data		return data


		@pytest.fixture(scope="function")
		def sha1():
		"""Fixture returning a valid hexadecimal sha1 value.
		"""
		return random_sha1()


		@pytest.fixture(scope="function")
		def invalid_sha1():
		"""Fixture returning an invalid sha1 representation.
		"""
		return hash_to_hex(bytes(random.randint(0, 255) for _ in range(50)))


		@pytest.fixture(scope="function")
		def sha256():
		"""Fixture returning a valid hexadecimal sha256 value.
		"""
		return random_sha256()


def _known_swh_objects(tests_data, object_type):		def _known_swh_objects(tests_data, object_type):
return tests_data[object_type]		return tests_data[object_type]


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def content(tests_data):		def content(tests_data):
"""Fixture returning a random content ingested into the test archive.		"""Fixture returning a random content ingested into the test archive.
"""		"""
return random.choice(_known_swh_objects(tests_data, "contents"))		return random.choice(_known_swh_objects(tests_data, "contents"))


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def contents(tests_data):		def contents(tests_data):
"""Fixture returning random contents ingested into the test archive.		"""Fixture returning random contents ingested into the test archive.
"""		"""
return random.choices(		return random.choices(
_known_swh_objects(tests_data, "contents"), k=random.randint(2, 8)		_known_swh_objects(tests_data, "contents"), k=random.randint(2, 8)
)		)


		def _new_content(tests_data):
		while True:
		new_content = random_content()
		sha1_bytes = hash_to_bytes(new_content["sha1"])
		if tests_data["storage"].content_get_data(sha1_bytes) is None:
		return new_content


		@pytest.fixture(scope="function")
		def unknown_content(tests_data):
		"""Fixture returning a random content not ingested into the test archive.
		"""
		return _new_content(tests_data)


		@pytest.fixture(scope="function")
		def unknown_contents(tests_data):
		"""Fixture returning random contents not ingested into the test archive.
		"""
		new_contents = []
		new_content_ids = set()
		nb_contents = random.randint(2, 8)
		while len(new_contents) != nb_contents:
		new_content = _new_content(tests_data)
		if new_content["sha1"] not in new_content_ids:
		new_contents.append(new_content)
		new_content_ids.add(new_content["sha1"])
		return list(new_contents)


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def empty_content():		def empty_content():
"""Fixture returning the empty content ingested into the test archive.		"""Fixture returning the empty content ingested into the test archive.
"""		"""
empty_content = Content.from_data(data=b"").to_dict()		empty_content = Content.from_data(data=b"").to_dict()
for algo in DEFAULT_ALGORITHMS:		for algo in DEFAULT_ALGORITHMS:
empty_content[algo] = hash_to_hex(empty_content[algo])		empty_content[algo] = hash_to_hex(empty_content[algo])
return empty_content		return empty_content
▲ Show 20 Lines • Show All 180 Lines • ▼ Show 20 Lines
@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def directory_with_files(tests_data):		def directory_with_files(tests_data):
"""Fixture returning a random directory containing at least one regular file.		"""Fixture returning a random directory containing at least one regular file.
"""		"""
return _directory_with_entry_type(tests_data, "file")		return _directory_with_entry_type(tests_data, "file")


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
		def unknown_directory(tests_data):
		"""Fixture returning a random directory not ingested into the test archive.
		"""
		while True:
		new_directory = random_sha1()
		sha1_bytes = hash_to_bytes(new_directory)
		if list(tests_data["storage"].directory_missing([sha1_bytes])):
		return new_directory


		@pytest.fixture(scope="function")
def empty_directory():		def empty_directory():
"""Fixture returning the empty directory ingested into the test archive.		"""Fixture returning the empty directory ingested into the test archive.
"""		"""
return Directory(entries=()).id.hex()		return Directory(entries=()).id.hex()


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def revision(tests_data):		def revision(tests_data):
Show All 17 Lines	def revisions_list(tests_data):
"""		"""

def gen_revisions_list(size):		def gen_revisions_list(size):
return random.choices(_known_swh_objects(tests_data, "revisions"), k=size,)		return random.choices(_known_swh_objects(tests_data, "revisions"), k=size,)

return gen_revisions_list		return gen_revisions_list


		@pytest.fixture(scope="function")
		def unknown_revision(tests_data):
		"""Fixture returning a random revision not ingested into the test archive.
		"""
		while True:
		new_revision = random_sha1()
		sha1_bytes = hash_to_bytes(new_revision)
		if tests_data["storage"].revision_get([sha1_bytes])[0] is None:
		return new_revision


def _get_origin_dfs_revisions_walker(tests_data):		def _get_origin_dfs_revisions_walker(tests_data):
storage = tests_data["storage"]		storage = tests_data["storage"]
origin = random.choice(tests_data["origins"][:-1])		origin = random.choice(tests_data["origins"][:-1])
snapshot = snapshot_get_latest(storage, origin["url"])		snapshot = snapshot_get_latest(storage, origin["url"])
if snapshot.branches[b"HEAD"].target_type.value == "alias":		if snapshot.branches[b"HEAD"].target_type.value == "alias":
target = snapshot.branches[b"HEAD"].target		target = snapshot.branches[b"HEAD"].target
head = snapshot.branches[target].target		head = snapshot.branches[target].target
else:		else:
▲ Show 20 Lines • Show All 90 Lines • ▼ Show 20 Lines	def releases(tests_data):
"""Fixture returning random releases ingested into the test archive.		"""Fixture returning random releases ingested into the test archive.
"""		"""
return random.choices(		return random.choices(
_known_swh_objects(tests_data, "releases"), k=random.randint(2, 8)		_known_swh_objects(tests_data, "releases"), k=random.randint(2, 8)
)		)


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
		def unknown_release(tests_data):
		"""Fixture returning a random release not ingested into the test archive.
		"""
		while True:
		new_release = random_sha1()
		sha1_bytes = hash_to_bytes(new_release)
		if tests_data["storage"].release_get([sha1_bytes])[0] is None:
		return new_release


		@pytest.fixture(scope="function")
def snapshot(tests_data):		def snapshot(tests_data):
"""Fixture returning a random snapshot ingested into the test archive.		"""Fixture returning a random snapshot ingested into the test archive.
"""		"""
return random.choice(_known_swh_objects(tests_data, "snapshots"))		return random.choice(_known_swh_objects(tests_data, "snapshots"))


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
		def unknown_snapshot(tests_data):
		"""Fixture returning a random snapshot not ingested into the test archive.
		"""
		while True:
		new_snapshot = random_sha1()
		sha1_bytes = hash_to_bytes(new_snapshot)
		if tests_data["storage"].snapshot_get_branches(sha1_bytes) is None:
		return new_snapshot


		@pytest.fixture(scope="function")
def origin(tests_data):		def origin(tests_data):
"""Fixture returning a random origin ingested into the test archive.		"""Fixture returning a random origin ingested into the test archive.
"""		"""
return random.choice(_known_swh_objects(tests_data, "origins"))		return random.choice(_known_swh_objects(tests_data, "origins"))


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def origin_with_multiple_visits(tests_data):		def origin_with_multiple_visits(tests_data):
▲ Show 20 Lines • Show All 465 Lines • Show Last 20 Lines