Differential D4807 Diff 18625 swh/model/hypothesis_strategies.py

Changeset View

Standalone View

swh/model/hypothesis_strategies.py

Show All 22 Lines	from hypothesis.strategies import (
one_of,		one_of,
sampled_from,		sampled_from,
sets,		sets,
text,		text,
tuples,		tuples,
)		)

from .from_disk import DentryPerms		from .from_disk import DentryPerms
from .identifiers import identifier_to_bytes, snapshot_identifier		from .identifiers import (
		EXTENDED_SWHID_TYPES,
		SWHID_TYPES,
		CoreSWHID,
		ExtendedSWHID,
		identifier_to_bytes,
		snapshot_identifier,
		)
from .model import (		from .model import (
BaseContent,		BaseContent,
Content,		Content,
Directory,		Directory,
DirectoryEntry,		DirectoryEntry,
		ExtID,
ObjectType,		ObjectType,
Origin,		Origin,
OriginVisit,		OriginVisit,
OriginVisitStatus,		OriginVisitStatus,
Person,		Person,
Release,		Release,
Revision,		Revision,
RevisionType,		RevisionType,
▲ Show 20 Lines • Show All 93 Lines • ▼ Show 20 Lines	):
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 core_SWHID_d():
		return builds(
		dict,
		namespace=just("swh"),
		scheme_version=just(1),
		object_type=sampled_from(SWHID_TYPES),
		object_id=sha1_git(),
		)


		def core_SWHID():
		return core_SWHID_d().map(lambda d: CoreSWHID(**d))


		def extended_SWHID_d():
		return builds(
		dict,
		namespace=just("swh"),
		scheme_version=just(1),
		object_type=sampled_from(EXTENDED_SWHID_TYPES),
		object_id=sha1_git(),
		)


		def extended_SWHID():
		return extended_SWHID_d().map(lambda d: ExtendedSWHID(**d))


def origins_d():		def origins_d():
return builds(dict, url=urls())		return builds(dict, url=urls())


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


▲ Show 20 Lines • Show All 131 Lines • ▼ Show 20 Lines
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 extids_d():
		return builds(
		dict,
		target=extended_SWHID().map(str),
		extid_type=sampled_from(["hg", "git", "bzr"]),
		extid=sha1(),
		)


		def extids():
		return extids_d().map(ExtID.from_dict)


def present_contents_d():		def present_contents_d():
return builds(		return builds(
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")),
)		)

▲ Show 20 Lines • Show All 118 Lines • ▼ Show 20 Lines	def objects(blacklist_types=("origin_visit_status",), split_content=False):
strategies = [		strategies = [
("origin", origins),		("origin", origins),
("origin_visit", origin_visits),		("origin_visit", origin_visits),
("origin_visit_status", origin_visit_statuses),		("origin_visit_status", origin_visit_statuses),
("snapshot", snapshots),		("snapshot", snapshots),
("release", releases),		("release", releases),
("revision", revisions),		("revision", revisions),
("directory", directories),		("directory", directories),
		("extid", extids),
]		]
if split_content:		if split_content:
strategies.append(("content", present_contents))		strategies.append(("content", present_contents))
strategies.append(("skipped_content", skipped_contents))		strategies.append(("skipped_content", skipped_contents))
else:		else:
strategies.append(("content", contents))		strategies.append(("content", contents))
args = [		args = [
obj_gen().map(lambda x, obj_type=obj_type: (obj_type, x))		obj_gen().map(lambda x, obj_type=obj_type: (obj_type, x))
Show All 17 Lines	def object_dicts(blacklist_types=("origin_visit_status",), split_content=False):
strategies = [		strategies = [
("origin", origins_d),		("origin", origins_d),
("origin_visit", origin_visits_d),		("origin_visit", origin_visits_d),
("origin_visit_status", origin_visit_statuses_d),		("origin_visit_status", origin_visit_statuses_d),
("snapshot", snapshots_d),		("snapshot", snapshots_d),
("release", releases_d),		("release", releases_d),
("revision", revisions_d),		("revision", revisions_d),
("directory", directories_d),		("directory", directories_d),
		("extid", extids_d),
]		]
if split_content:		if split_content:
strategies.append(("content", present_contents_d))		strategies.append(("content", present_contents_d))
strategies.append(("skipped_content", skipped_contents_d))		strategies.append(("skipped_content", skipped_contents_d))
else:		else:
strategies.append(("content", contents_d))		strategies.append(("content", contents_d))
args = [		args = [
obj_gen().map(lambda x, obj_type=obj_type: (obj_type, x))		obj_gen().map(lambda x, obj_type=obj_type: (obj_type, x))
for (obj_type, obj_gen) in strategies		for (obj_type, obj_gen) in strategies
if obj_type not in blacklist_types		if obj_type not in blacklist_types
]		]
return one_of(*args)		return one_of(*args)