Differential D8179 Diff 29534 swh/web/tests/conftest.py

Changeset View

Standalone View

swh/web/tests/conftest.py

# Copyright (C) 2018-2022 The Software Heritage developers		# Copyright (C) 2018-2022 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 Affero General Public License version 3, or any later version		# License: GNU Affero 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 collections import defaultdict		from collections import defaultdict
from datetime import timedelta		from datetime import timedelta
import functools		import functools
import json		import json
import os		import os
import random		import random
import shutil		import shutil
from subprocess import PIPE, run
import sys		import sys
import time		import time
from typing import Any, Dict, List, Optional		from typing import Any, Dict, List, Optional

from _pytest.python import Function		from _pytest.python import Function
from hypothesis import HealthCheck, settings		from hypothesis import HealthCheck, settings
import pytest		import pytest

Show All 32 Lines	from swh.web.tests.data import (
random_sha1,		random_sha1,
random_sha1_bytes,		random_sha1_bytes,
random_sha256,		random_sha256,
)		)
from swh.web.tests.utils import create_django_permission		from swh.web.tests.utils import create_django_permission

os.environ["LC_ALL"] = "C.UTF-8"		os.environ["LC_ALL"] = "C.UTF-8"

# Used to skip some tests
ctags_json_missing = (
shutil.which("ctags") is None
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

# Register some hypothesis profiles		# Register some hypothesis profiles
settings.register_profile("default", settings())		settings.register_profile("default", settings())

# we use getattr here to keep mypy happy regardless hypothesis version		# we use getattr here to keep mypy happy regardless hypothesis version
function_scoped_fixture_check = (		function_scoped_fixture_check = (
[getattr(HealthCheck, "function_scoped_fixture")]		[getattr(HealthCheck, "function_scoped_fixture")]
▲ Show 20 Lines • Show All 364 Lines • ▼ Show 20 Lines	def content_utf8_detected_as_binary():
"""Fixture returning a random textual content detected as binary		"""Fixture returning a random textual content detected as binary
by libmagic while they are valid UTF-8 encoded files.		by libmagic while they are valid UTF-8 encoded files.
"""		"""

return random.choice(_content_utf8_detected_as_binary())		return random.choice(_content_utf8_detected_as_binary())


@pytest.fixture(scope="function")		@pytest.fixture(scope="function")
def contents_with_ctags():
"""
Fixture returning contents ingested into the test archive.
Those contents are ctags compatible, that is running ctags on those lay results.
"""
return {
"sha1s": [
"0ab37c02043ebff946c1937523f60aadd0844351",
"15554cf7608dde6bfefac7e3d525596343a85b6f",
"2ce837f1489bdfb8faf3ebcc7e72421b5bea83bd",
"30acd0b47fc25e159e27a980102ddb1c4bea0b95",
"4f81f05aaea3efb981f9d90144f746d6b682285b",
"5153aa4b6e4455a62525bc4de38ed0ff6e7dd682",
"59d08bafa6a749110dfb65ba43a61963d5a5bf9f",
"7568285b2d7f31ae483ae71617bd3db873deaa2c",
"7ed3ee8e94ac52ba983dd7690bdc9ab7618247b4",
"8ed7ef2e7ff9ed845e10259d08e4145f1b3b5b03",
"9b3557f1ab4111c8607a4f2ea3c1e53c6992916c",
"9c20da07ed14dc4fcd3ca2b055af99b2598d8bdd",
"c20ceebd6ec6f7a19b5c3aebc512a12fbdc9234b",
"e89e55a12def4cd54d5bff58378a3b5119878eb7",
"e8c0654fe2d75ecd7e0b01bee8a8fc60a130097e",
"eb6595e559a1d34a2b41e8d4835e0e4f98a5d2b5",
],
"symbol_name": "ABS",
}


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


@functools.lru_cache(maxsize=None)		@functools.lru_cache(maxsize=None)
def _directory_with_entry_type(type_):		def _directory_with_entry_type(type_):
tests_data = get_tests_data()		tests_data = get_tests_data()
▲ Show 20 Lines • Show All 540 Lines • ▼ Show 20 Lines	class _IndexerData:
a json serializable format in order to ease tests implementation.		a json serializable format in order to ease tests implementation.

"""		"""

def __init__(self, tests_data):		def __init__(self, tests_data):
self.idx_storage = tests_data["idx_storage"]		self.idx_storage = tests_data["idx_storage"]
self.mimetype_indexer = tests_data["mimetype_indexer"]		self.mimetype_indexer = tests_data["mimetype_indexer"]
self.license_indexer = tests_data["license_indexer"]		self.license_indexer = tests_data["license_indexer"]
self.ctags_indexer = tests_data["ctags_indexer"]

def content_add_mimetype(self, cnt_id):		def content_add_mimetype(self, cnt_id):
self.mimetype_indexer.run([hash_to_bytes(cnt_id)])		self.mimetype_indexer.run([hash_to_bytes(cnt_id)])

def content_get_mimetype(self, cnt_id):		def content_get_mimetype(self, cnt_id):
mimetype = self.idx_storage.content_mimetype_get([hash_to_bytes(cnt_id)])[		mimetype = self.idx_storage.content_mimetype_get([hash_to_bytes(cnt_id)])[
0		0
].to_dict()		].to_dict()
return converters.from_filetype(mimetype)		return converters.from_filetype(mimetype)

def content_add_license(self, cnt_id):		def content_add_license(self, cnt_id):
self.license_indexer.run([hash_to_bytes(cnt_id)])		self.license_indexer.run([hash_to_bytes(cnt_id)])

def content_get_license(self, cnt_id):		def content_get_license(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)		cnt_id_bytes = hash_to_bytes(cnt_id)
licenses = self.idx_storage.content_fossology_license_get([cnt_id_bytes])		licenses = self.idx_storage.content_fossology_license_get([cnt_id_bytes])
for license in licenses:		for license in licenses:
yield converters.from_swh(license.to_dict(), hashess={"id"})		yield converters.from_swh(license.to_dict(), hashess={"id"})

def content_add_ctags(self, cnt_id):
self.ctags_indexer.run([hash_to_bytes(cnt_id)])

def content_get_ctags(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
ctags = self.idx_storage.content_ctags_get([cnt_id_bytes])
for ctag in ctags:
yield converters.from_swh(ctag, hashess={"id"})


@pytest.fixture		@pytest.fixture
def keycloak_oidc(keycloak_oidc, mocker):		def keycloak_oidc(keycloak_oidc, mocker):
keycloak_config = get_config()["keycloak"]		keycloak_config = get_config()["keycloak"]

keycloak_oidc.server_url = keycloak_config["server_url"]		keycloak_oidc.server_url = keycloak_config["server_url"]
keycloak_oidc.realm_name = keycloak_config["realm_name"]		keycloak_oidc.realm_name = keycloak_config["realm_name"]
keycloak_oidc.client_id = OIDC_SWH_WEB_CLIENT_ID		keycloak_oidc.client_id = OIDC_SWH_WEB_CLIENT_ID
▲ Show 20 Lines • Show All 181 Lines • Show Last 20 Lines