strategies.py
No OneTemporary
Actions

Size

7 KB

Subscribers

None

strategies.py
View Options

	# Copyright (C) 2018-2021 The Software Heritage developers
	# See the AUTHORS file at the top-level directory of this distribution
	# License: GNU Affero General Public License version 3, or any later version
	# See top-level LICENSE file for more information

	from datetime import datetime
	import random

	from hypothesis import assume, settings
	from hypothesis.extra.dateutil import timezones
	from hypothesis.strategies import (
	binary,
	characters,
	composite,
	datetimes,
	lists,
	sampled_from,
	text,
	)

	from swh.model.hashutil import hash_to_bytes, hash_to_hex
	from swh.model.hypothesis_strategies import origins as new_origin_strategy
	from swh.model.hypothesis_strategies import snapshots as new_snapshot
	from swh.model.model import Person, Revision, RevisionType, TimestampWithTimezone
	from swh.model.swhids import ObjectType
	from swh.web.tests.data import get_tests_data

	# Module dedicated to the generation of input data for tests through
	# the use of hypothesis.
	# Some of these data are sampled from a test archive created and populated
	# in the swh.web.tests.data module.

	# Set the swh-web hypothesis profile if none has been explicitly set
	hypothesis_default_settings = settings.get_profile("default")
	if repr(settings()) == repr(hypothesis_default_settings):
	settings.load_profile("swh-web")


	# The following strategies exploit the hypothesis capabilities


	def _known_swh_object(object_type):
	return sampled_from(get_tests_data()[object_type])


	def sha1():
	"""
	Hypothesis strategy returning a valid hexadecimal sha1 value.
	"""
	return binary(min_size=20, max_size=20).map(hash_to_hex)


	def invalid_sha1():
	"""
	Hypothesis strategy returning an invalid sha1 representation.
	"""
	return binary(min_size=50, max_size=50).map(hash_to_hex)


	def sha256():
	"""
	Hypothesis strategy returning a valid hexadecimal sha256 value.
	"""
	return binary(min_size=32, max_size=32).map(hash_to_hex)


	@composite
	def new_content(draw):
	blake2s256_hex = draw(sha256())
	sha1_hex = draw(sha1())
	sha1_git_hex = draw(sha1())
	sha256_hex = draw(sha256())

	assume(sha1_hex != sha1_git_hex)
	assume(blake2s256_hex != sha256_hex)

	return {
	"blake2S256": blake2s256_hex,
	"sha1": sha1_hex,
	"sha1_git": sha1_git_hex,
	"sha256": sha256_hex,
	}


	def unknown_content():
	"""
	Hypothesis strategy returning a random content not ingested
	into the test archive.
	"""
	return new_content().filter(
	lambda c: get_tests_data()["storage"].content_get_data(hash_to_bytes(c["sha1"]))
	is None
	)


	def unknown_contents():
	"""
	Hypothesis strategy returning random contents not ingested
	into the test archive.
	"""
	return lists(unknown_content(), min_size=2, max_size=8)


	def unknown_directory():
	"""
	Hypothesis strategy returning a random directory not ingested
	into the test archive.
	"""
	return sha1().filter(
	lambda s: len(
	list(get_tests_data()["storage"].directory_missing([hash_to_bytes(s)]))
	)
	> 0
	)


	def new_origin():
	"""
	Hypothesis strategy returning a random origin not ingested
	into the test archive.
	"""
	return new_origin_strategy()


	def new_origins(nb_origins=None):
	"""
	Hypothesis strategy returning random origins not ingested
	into the test archive.
	"""
	min_size = nb_origins if nb_origins is not None else 2
	max_size = nb_origins if nb_origins is not None else 8
	size = random.randint(min_size, max_size)
	return lists(
	new_origin(),
	min_size=size,
	max_size=size,
	unique_by=lambda o: tuple(sorted(o.items())),
	)


	def visit_dates(nb_dates=None):
	"""
	Hypothesis strategy returning a list of visit dates.
	"""
	min_size = nb_dates if nb_dates else 2
	max_size = nb_dates if nb_dates else 8
	return lists(
	datetimes(
	min_value=datetime(2015, 1, 1, 0, 0),
	max_value=datetime(2018, 12, 31, 0, 0),
	timezones=timezones(),
	),
	min_size=min_size,
	max_size=max_size,
	unique=True,
	).map(sorted)


	def unknown_release():
	"""
	Hypothesis strategy returning a random revision not ingested
	into the test archive.
	"""
	return sha1().filter(
	lambda s: get_tests_data()["storage"].release_get([s])[0] is None
	)


	def unknown_revision():
	"""
	Hypothesis strategy returning a random revision not ingested
	into the test archive.
	"""
	return sha1().filter(
	lambda s: get_tests_data()["storage"].revision_get([hash_to_bytes(s)])[0]
	is None
	)


	@composite
	def new_person(draw):
	"""
	Hypothesis strategy returning random raw swh person data.
	"""
	name = draw(
	text(
	min_size=5,
	max_size=30,
	alphabet=characters(min_codepoint=0, max_codepoint=255),
	)
	)
	email = "%s@company.org" % name
	return Person(
	name=name.encode(),
	email=email.encode(),
	fullname=("%s <%s>" % (name, email)).encode(),
	)


	@composite
	def new_swh_date(draw):
	"""
	Hypothesis strategy returning random raw swh date data.
	"""
	timestamp = draw(
	datetimes(
	min_value=datetime(2015, 1, 1, 0, 0), max_value=datetime(2018, 12, 31, 0, 0)
	).map(lambda d: int(d.timestamp()))
	)
	return {
	"timestamp": timestamp,
	"offset": 0,
	"negative_utc": False,
	}


	@composite
	def new_revision(draw):
	"""
	Hypothesis strategy returning random raw swh revision data
	not ingested into the test archive.
	"""
	return Revision(
	directory=draw(sha1().map(hash_to_bytes)),
	author=draw(new_person()),
	committer=draw(new_person()),
	message=draw(text(min_size=20, max_size=100).map(lambda t: t.encode())),
	date=TimestampWithTimezone.from_datetime(draw(new_swh_date())),
	committer_date=TimestampWithTimezone.from_datetime(draw(new_swh_date())),
	synthetic=False,
	type=RevisionType.GIT,
	)


	def unknown_revisions(min_size=2, max_size=8):
	"""
	Hypothesis strategy returning random revisions not ingested
	into the test archive.
	"""
	return lists(unknown_revision(), min_size=min_size, max_size=max_size)


	def snapshot():
	"""
	Hypothesis strategy returning a random snapshot ingested
	into the test archive.
	"""
	return _known_swh_object("snapshots")


	def new_snapshots(nb_snapshots=None):
	min_size = nb_snapshots if nb_snapshots else 2
	max_size = nb_snapshots if nb_snapshots else 8
	return lists(
	new_snapshot(min_size=2, max_size=10, only_objects=True),
	min_size=min_size,
	max_size=max_size,
	)


	def unknown_snapshot():
	"""
	Hypothesis strategy returning a random revision not ingested
	into the test archive.
	"""
	return sha1().filter(
	lambda s: get_tests_data()["storage"].snapshot_get_branches(hash_to_bytes(s))
	is None
	)


	def swhid():
	"""
	Hypothesis strategy returning a qualified SWHID for any object
	ingested into the test archive.
	"""
	return _known_swh_object("swhids")


	def snapshot_swhid():
	"""
	Hypothesis strategy returning a qualified SWHID for a snapshot object
	ingested into the test archive.
	"""
	return swhid().filter(lambda swhid: swhid.object_type == ObjectType.SNAPSHOT)

File Metadata

Mime Type: text/x-python
Expires: Fri, Jul 4, 12:41 PM (2 w, 2 d ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 3446542

strategies.pyNo OneTemporaryActions

strategies.pyView Options

File Metadata

Event Timeline

strategies.py
No OneTemporary
Actions

strategies.py
View Options