Differential D8110 Diff 29301 swh/model/hypothesis_strategies.py

Changeset View

Standalone View

swh/model/hypothesis_strategies.py

	Show First 20 Lines • Show All 114 Lines • ▼ Show 20 Lines
	def persons_d(draw):			def persons_d(draw):
	fullname = draw(binary())			fullname = draw(binary())
	email = draw(optional(binary()))			email = draw(optional(binary()))
	name = draw(optional(binary()))			name = draw(optional(binary()))
	assume(not (len(fullname) == 32 and email is None and name is None))			assume(not (len(fullname) == 32 and email is None and name is None))
	return dict(fullname=fullname, name=name, email=email)			return dict(fullname=fullname, name=name, email=email)


	def persons():			def persons(**kwargs):
	return persons_d().map(Person.from_dict)			return persons_d(**kwargs).map(Person.from_dict)


	def timestamps_d():			def timestamps_d(**kwargs):
	max_seconds = datetime.datetime.max.replace(			max_seconds = datetime.datetime.max.replace(
	tzinfo=datetime.timezone.utc			tzinfo=datetime.timezone.utc
	).timestamp()			).timestamp()
	min_seconds = datetime.datetime.min.replace(			min_seconds = datetime.datetime.min.replace(
	tzinfo=datetime.timezone.utc			tzinfo=datetime.timezone.utc
	).timestamp()			).timestamp()
	return builds(			defaults = dict(
	dict,
	seconds=integers(min_seconds, max_seconds),			seconds=integers(min_seconds, max_seconds),
	microseconds=integers(0, 1000000),			microseconds=integers(0, 1000000),
	)			)
				return builds(dict, {defaults, **kwargs})


	def timestamps():			def timestamps():
	return timestamps_d().map(Timestamp.from_dict)			return timestamps_d().map(Timestamp.from_dict)


	@composite			@composite
	def timestamps_with_timezone_d(			def timestamps_with_timezone_d(
	draw,			draw,
				*,
	timestamp=timestamps_d(),			timestamp=timestamps_d(),
	offset=integers(min_value=-14 * 60, max_value=14 * 60),			offset=integers(min_value=-14 * 60, max_value=14 * 60),
	negative_utc=booleans(),			negative_utc=booleans(),
	):			):
	timestamp = draw(timestamp)			timestamp = draw(timestamp)
	offset = draw(offset)			offset = draw(offset)
	negative_utc = draw(negative_utc)			negative_utc = draw(negative_utc)
	assume(not (negative_utc and offset))			assume(not (negative_utc and offset))
	return dict(timestamp=timestamp, offset=offset, negative_utc=negative_utc)			return dict(timestamp=timestamp, offset=offset, negative_utc=negative_utc)


	timestamps_with_timezone = timestamps_with_timezone_d().map(			timestamps_with_timezone = timestamps_with_timezone_d().map(
	TimestampWithTimezone.from_dict			TimestampWithTimezone.from_dict
	)			)


	def origins_d():			def origins_d(*, url=iris()):
	return builds(dict, url=iris())			return builds(dict, url=url)


	def origins():			def origins(**kwargs):
	return origins_d().map(Origin.from_dict)			return origins_d(**kwargs).map(Origin.from_dict)


	def origin_visits_d():			def origin_visits_d(**kwargs):
	return builds(			defaults = dict(
	dict,
	visit=integers(1, 1000),			visit=integers(1, 1000),
	origin=iris(),			origin=iris(),
	date=aware_datetimes(),			date=aware_datetimes(),
	type=pgsql_text(),			type=pgsql_text(),
	)			)
				return builds(dict, {defaults, **kwargs})


	def origin_visits():			def origin_visits(**kwargs):
	return origin_visits_d().map(OriginVisit.from_dict)			return origin_visits_d(**kwargs).map(OriginVisit.from_dict)


	def metadata_dicts():			def metadata_dicts():
	return dictionaries(pgsql_text(), pgsql_text())			return dictionaries(pgsql_text(), pgsql_text())


	def origin_visit_statuses_d():			def origin_visit_statuses_d(**kwargs):
	return builds(			defaults = dict(
	dict,
	visit=integers(1, 1000),			visit=integers(1, 1000),
	origin=iris(),			origin=iris(),
	type=optional(sampled_from(["git", "svn", "pypi", "debian"])),			type=optional(sampled_from(["git", "svn", "pypi", "debian"])),
	status=sampled_from(			status=sampled_from(
	["created", "ongoing", "full", "partial", "not_found", "failed"]			["created", "ongoing", "full", "partial", "not_found", "failed"]
	),			),
	date=aware_datetimes(),			date=aware_datetimes(),
	snapshot=optional(sha1_git()),			snapshot=optional(sha1_git()),
	metadata=optional(metadata_dicts()),			metadata=optional(metadata_dicts()),
	)			)
				return builds(dict, {defaults, **kwargs})


	def origin_visit_statuses():			def origin_visit_statuses(**kwargs):
	return origin_visit_statuses_d().map(OriginVisitStatus.from_dict)			return origin_visit_statuses_d(**kwargs).map(OriginVisitStatus.from_dict)


	@composite			@composite
	def releases_d(draw):			def releases_d(draw, **kwargs):
	target_type = sampled_from([x.value for x in ObjectType])			defaults = dict(
	name = binary()			target_type=sampled_from([x.value for x in ObjectType]),
	message = optional(binary())			name=binary(),
	synthetic = booleans()			message=optional(binary()),
	target = sha1_git()			synthetic=booleans(),
	metadata = optional(revision_metadata())			target=sha1_git(),
				metadata=optional(revision_metadata()),
				raw_manifest=optional(binary()),
				)

	d = draw(			d = draw(
	one_of(			one_of(
	# None author/date:			# None author/date:
	builds(			builds(dict, author=none(), date=none(), {defaults, **kwargs}),
	dict,
	name=name,
	message=message,
	synthetic=synthetic,
	author=none(),
	date=none(),
	target=target,
	target_type=target_type,
	metadata=metadata,
	),
	# non-None author/date:			# non-None author/date:
	builds(			builds(
	dict,			dict,
	name=name,
	message=message,
	synthetic=synthetic,
	date=timestamps_with_timezone_d(),			date=timestamps_with_timezone_d(),
	author=persons_d(),			author=persons_d(),
	target=target,			{defaults, **kwargs},
	target_type=target_type,
	metadata=metadata,
	),			),
	# it is also possible for date to be None but not author, but let's not			# it is also possible for date to be None but not author, but let's not
	# overwhelm hypothesis with this edge case			# overwhelm hypothesis with this edge case
	)			)
	)			)

	raw_manifest = draw(optional(binary()))			if d["raw_manifest"] is None:
	if raw_manifest:			del d["raw_manifest"]
	d["raw_manifest"] = raw_manifest
	return d			return d


	def releases():			def releases(**kwargs):
	return releases_d().map(Release.from_dict)			return releases_d(**kwargs).map(Release.from_dict)


	revision_metadata = metadata_dicts			revision_metadata = metadata_dicts


	def extra_headers():			def extra_headers():
	return lists(			return lists(
	tuples(binary(min_size=0, max_size=50), binary(min_size=0, max_size=500))			tuples(binary(min_size=0, max_size=50), binary(min_size=0, max_size=500))
	).map(tuple)			).map(tuple)


	@composite			@composite
	def revisions_d(draw):			def revisions_d(draw, **kwargs):
				defaults = dict(
				message=optional(binary()),
				synthetic=booleans(),
				parents=tuples(sha1_git()),
				directory=sha1_git(),
				type=sampled_from([x.value for x in RevisionType]),
				metadata=optional(revision_metadata()),
				extra_headers=extra_headers(),
				raw_manifest=optional(binary()),
				)
	d = draw(			d = draw(
	one_of(			one_of(
	# None author/committer/date/committer_date			# None author/committer/date/committer_date
	builds(			builds(
	dict,			dict,
	message=optional(binary()),
	synthetic=booleans(),
	author=none(),			author=none(),
	committer=none(),			committer=none(),
	date=none(),			date=none(),
	committer_date=none(),			committer_date=none(),
	parents=tuples(sha1_git()),			{defaults, **kwargs},
	directory=sha1_git(),
	type=sampled_from([x.value for x in RevisionType]),
	metadata=optional(revision_metadata()),
	extra_headers=extra_headers(),
	),			),
	# non-None author/committer/date/committer_date			# non-None author/committer/date/committer_date
	builds(			builds(
	dict,			dict,
	message=optional(binary()),
	synthetic=booleans(),
	author=persons_d(),			author=persons_d(),
	committer=persons_d(),			committer=persons_d(),
	date=timestamps_with_timezone_d(),			date=timestamps_with_timezone_d(),
	committer_date=timestamps_with_timezone_d(),			committer_date=timestamps_with_timezone_d(),
	parents=tuples(sha1_git()),			{defaults, **kwargs},
	directory=sha1_git(),
	type=sampled_from([x.value for x in RevisionType]),
	metadata=optional(revision_metadata()),
	extra_headers=extra_headers(),
	),			),
	# There are many other combinations, but let's not overwhelm hypothesis			# There are many other combinations, but let's not overwhelm hypothesis
	# with these edge cases			# with these edge cases
	)			)
	)			)
	# TODO: metadata['extra_headers'] can have binary keys and values			# TODO: metadata['extra_headers'] can have binary keys and values

	raw_manifest = draw(optional(binary()))			if d["raw_manifest"] is None:
	if raw_manifest:			del d["raw_manifest"]
	d["raw_manifest"] = raw_manifest
	return d			return d


	def revisions():			def revisions(**kwargs):
	return revisions_d().map(Revision.from_dict)			return revisions_d(**kwargs).map(Revision.from_dict)


	def directory_entries_d():			def directory_entries_d(**kwargs):
				defaults = dict(
				name=binaries_without_bytes(b"/"),
				target=sha1_git(),
				)
	return one_of(			return one_of(
	builds(			builds(
	dict,			dict,
	name=binaries_without_bytes(b"/"),
	target=sha1_git(),
	type=just("file"),			type=just("file"),
	perms=one_of(			perms=one_of(
	integers(min_value=0o100000, max_value=0o100777), # regular file			integers(min_value=0o100000, max_value=0o100777), # regular file
	integers(min_value=0o120000, max_value=0o120777), # symlink			integers(min_value=0o120000, max_value=0o120777), # symlink
	),			),
				{defaults, **kwargs},
	),			),
	builds(			builds(
	dict,			dict,
	name=binaries_without_bytes(b"/"),
	target=sha1_git(),
	type=just("dir"),			type=just("dir"),
	perms=integers(			perms=integers(
	min_value=DentryPerms.directory,			min_value=DentryPerms.directory,
	max_value=DentryPerms.directory + 0o777,			max_value=DentryPerms.directory + 0o777,
	),			),
				{defaults, **kwargs},
	),			),
	builds(			builds(
	dict,			dict,
	name=binaries_without_bytes(b"/"),
	target=sha1_git(),
	type=just("rev"),			type=just("rev"),
	perms=integers(			perms=integers(
	min_value=DentryPerms.revision,			min_value=DentryPerms.revision,
	max_value=DentryPerms.revision + 0o777,			max_value=DentryPerms.revision + 0o777,
	),			),
				{defaults, **kwargs},
	),			),
	)			)


	def directory_entries():			def directory_entries(**kwargs):
	return directory_entries_d().map(DirectoryEntry)			return directory_entries_d(**kwargs).map(DirectoryEntry)


	@composite			@composite
	def directories_d(draw):			def directories_d(draw, raw_manifest=optional(binary())):
	d = draw(builds(dict, entries=tuples(directory_entries_d())))			d = draw(builds(dict, entries=tuples(directory_entries_d())))

	raw_manifest = draw(optional(binary()))			d["raw_manifest"] = draw(raw_manifest)
	if raw_manifest:			if d["raw_manifest"] is None:
	d["raw_manifest"] = raw_manifest			del d["raw_manifest"]
	return d			return d


	def directories():			def directories(**kwargs):
	return directories_d().map(Directory.from_dict)			return directories_d(**kwargs).map(Directory.from_dict)


	def contents_d():			def contents_d():
	return one_of(present_contents_d(), skipped_contents_d())			return one_of(present_contents_d(), skipped_contents_d())


	def contents():			def contents():
	return one_of(present_contents(), skipped_contents())			return one_of(present_contents(), skipped_contents())


	def present_contents_d():			def present_contents_d(**kwargs):
	return builds(			defaults = dict(
	dict,
	data=binary(max_size=4096),			data=binary(max_size=4096),
	ctime=optional(aware_datetimes()),			ctime=optional(aware_datetimes()),
	status=one_of(just("visible"), just("hidden")),			status=one_of(just("visible"), just("hidden")),
	)			)
				return builds(dict, {defaults, **kwargs})


	def present_contents():			def present_contents(**kwargs):
	return present_contents_d().map(lambda d: Content.from_data(**d))			return present_contents_d().map(lambda d: Content.from_data(**d))


	@composite			@composite
	def skipped_contents_d(draw):			def skipped_contents_d(
				draw, reason=pgsql_text(), status=just("absent"), ctime=optional(aware_datetimes())
				):
	result = BaseContent._hash_data(draw(binary(max_size=4096)))			result = BaseContent._hash_data(draw(binary(max_size=4096)))
	result.pop("data")			result.pop("data")
	nullify_attrs = draw(			nullify_attrs = draw(
	sets(sampled_from(["sha1", "sha1_git", "sha256", "blake2s256"]))			sets(sampled_from(["sha1", "sha1_git", "sha256", "blake2s256"]))
	)			)
	for k in nullify_attrs:			for k in nullify_attrs:
	result[k] = None			result[k] = None
	result["reason"] = draw(pgsql_text())			result["reason"] = draw(reason)
	result["status"] = "absent"			result["status"] = draw(status)
	result["ctime"] = draw(optional(aware_datetimes()))			result["ctime"] = draw(ctime)
	return result			return result


	def skipped_contents():			def skipped_contents(**kwargs):
	return skipped_contents_d().map(SkippedContent.from_dict)			return skipped_contents_d().map(SkippedContent.from_dict)


	def branch_names():			def branch_names():
	return binary(min_size=1)			return binary(min_size=1)


	def branch_targets_object_d():			def branch_targets_object_d():
	▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Lines


	def snapshots(*, min_size=0, max_size=100, only_objects=False):			def snapshots(*, min_size=0, max_size=100, only_objects=False):
	return snapshots_d(			return snapshots_d(
	min_size=min_size, max_size=max_size, only_objects=only_objects			min_size=min_size, max_size=max_size, only_objects=only_objects
	).map(Snapshot.from_dict)			).map(Snapshot.from_dict)


	def metadata_authorities():			def metadata_authorities(url=iris()):
	return builds(MetadataAuthority, url=iris(), metadata=just(None))			return builds(MetadataAuthority, url=url, metadata=just(None))


	def metadata_fetchers():			def metadata_fetchers(**kwargs):
	return builds(			defaults = dict(
	MetadataFetcher,
	name=text(min_size=1, alphabet=string.printable),			name=text(min_size=1, alphabet=string.printable),
	version=text(			version=text(
	min_size=1,			min_size=1,
	alphabet=string.ascii_letters + string.digits + string.punctuation,			alphabet=string.ascii_letters + string.digits + string.punctuation,
	),			),
				)
				return builds(
				MetadataFetcher,
	metadata=just(None),			metadata=just(None),
				{defaults, **kwargs},
	)			)


	def raw_extrinsic_metadata():			def raw_extrinsic_metadata(**kwargs):
	return builds(			defaults = dict(
	RawExtrinsicMetadata,
	target=extended_swhids(),			target=extended_swhids(),
	discovery_date=aware_datetimes(),			discovery_date=aware_datetimes(),
	authority=metadata_authorities(),			authority=metadata_authorities(),
	fetcher=metadata_fetchers(),			fetcher=metadata_fetchers(),
	format=text(min_size=1, alphabet=string.printable),			format=text(min_size=1, alphabet=string.printable),
	)			)
				return builds(RawExtrinsicMetadata, {defaults, **kwargs})


	def raw_extrinsic_metadata_d():			def raw_extrinsic_metadata_d(**kwargs):
	return raw_extrinsic_metadata().map(RawExtrinsicMetadata.to_dict)			return raw_extrinsic_metadata(**kwargs).map(RawExtrinsicMetadata.to_dict)


	def objects(blacklist_types=("origin_visit_status",), split_content=False):			def objects(blacklist_types=("origin_visit_status",), split_content=False):
	"""generates a random couple (type, obj)			"""generates a random couple (type, obj)

	which obj is an instance of the Model class corresponding to obj_type.			which obj is an instance of the Model class corresponding to obj_type.

	`blacklist_types` is a list of obj_type to exclude from the strategy.			`blacklist_types` is a list of obj_type to exclude from the strategy.
	▲ Show 20 Lines • Show All 59 Lines • Show Last 20 Lines