Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/strategies.py
| Show All 10 Lines | |||||
| from hypothesis import settings, assume | from hypothesis import settings, assume | ||||
| from hypothesis.strategies import ( | from hypothesis.strategies import ( | ||||
| just, sampled_from, lists, composite, datetimes, | just, sampled_from, lists, composite, datetimes, | ||||
| binary, text, characters | binary, text, characters | ||||
| ) | ) | ||||
| from swh.model.hashutil import hash_to_hex, hash_to_bytes | from swh.model.hashutil import hash_to_hex, hash_to_bytes | ||||
| from swh.model.identifiers import directory_identifier | from swh.model.identifiers import directory_identifier | ||||
| from swh.model.model import Person, Revision, TimestampWithTimezone | |||||
| from swh.storage.algos.revisions_walker import get_revisions_walker | from swh.storage.algos.revisions_walker import get_revisions_walker | ||||
| from swh.model.hypothesis_strategies import ( | from swh.model.hypothesis_strategies import ( | ||||
| origins as new_origin_strategy, snapshots as new_snapshot | origins as new_origin_strategy, snapshots as new_snapshot | ||||
| ) | ) | ||||
| from swh.web.tests.data import get_tests_data | from swh.web.tests.data import get_tests_data | ||||
| # Module dedicated to the generation of input data for tests through | # Module dedicated to the generation of input data for tests through | ||||
| # the use of hypothesis. | # the use of hypothesis. | ||||
| ▲ Show 20 Lines • Show All 223 Lines • ▼ Show 20 Lines | def origin_with_releases(): | ||||
| return sampled_from(ret) | return sampled_from(ret) | ||||
| def new_origin(): | def new_origin(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random origin not ingested | Hypothesis strategy returning a random origin not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return new_origin_strategy().map(lambda origin: origin.to_dict()).filter( | return new_origin_strategy().filter( | ||||
| lambda origin: get_tests_data()['storage'].origin_get( | lambda origin: get_tests_data()['storage'].origin_get( | ||||
| [origin])[0] is None) | [origin.to_dict()])[0] is None) | ||||
| def new_origins(nb_origins=None): | def new_origins(nb_origins=None): | ||||
| """ | """ | ||||
| Hypothesis strategy returning random origins not ingested | Hypothesis strategy returning random origins not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| min_size = nb_origins if nb_origins is not None else 2 | min_size = nb_origins if nb_origins is not None else 2 | ||||
| ▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines | |||||
| @composite | @composite | ||||
| def new_person(draw): | def new_person(draw): | ||||
| """ | """ | ||||
| Hypothesis strategy returning random raw swh person data. | Hypothesis strategy returning random raw swh person data. | ||||
| """ | """ | ||||
| name = draw(text(min_size=5, max_size=30, | name = draw(text(min_size=5, max_size=30, | ||||
| alphabet=characters(min_codepoint=0, max_codepoint=255))) | alphabet=characters(min_codepoint=0, max_codepoint=255))) | ||||
| email = '%s@company.org' % name | email = '%s@company.org' % name | ||||
| return { | return Person( | ||||
| 'name': name.encode(), | name=name.encode(), | ||||
| 'email': email.encode(), | email=email.encode(), | ||||
| 'fullname': ('%s <%s>' % (name, email)).encode() | fullname=('%s <%s>' % (name, email)).encode() | ||||
| } | ) | ||||
| @composite | @composite | ||||
| def new_swh_date(draw): | def new_swh_date(draw): | ||||
| """ | """ | ||||
| Hypothesis strategy returning random raw swh date data. | Hypothesis strategy returning random raw swh date data. | ||||
| """ | """ | ||||
| timestamp = draw( | timestamp = draw( | ||||
| datetimes(min_value=datetime(2015, 1, 1, 0, 0), | datetimes(min_value=datetime(2015, 1, 1, 0, 0), | ||||
| max_value=datetime(2018, 12, 31, 0, 0)).map( | max_value=datetime(2018, 12, 31, 0, 0)).map( | ||||
| lambda d: int(d.timestamp()))) | lambda d: int(d.timestamp()))) | ||||
| return { | return { | ||||
| 'timestamp': timestamp, | 'timestamp': timestamp, | ||||
| 'offset': 0, | 'offset': 0, | ||||
| 'negative_utc': False, | 'negative_utc': False, | ||||
| } | } | ||||
| @composite | @composite | ||||
| def new_revision(draw): | def new_revision(draw): | ||||
| """ | """ | ||||
| Hypothesis strategy returning random raw swh revision data | Hypothesis strategy returning random raw swh revision data | ||||
| not ingested into the test archive. | not ingested into the test archive. | ||||
| """ | """ | ||||
| return { | return Revision( | ||||
| 'id': draw(unknown_revision().map(hash_to_bytes)), | directory=draw(sha1().map(hash_to_bytes)), | ||||
| 'directory': draw(sha1().map(hash_to_bytes)), | author=draw(new_person()), | ||||
| 'author': draw(new_person()), | committer=draw(new_person()), | ||||
| 'committer': draw(new_person()), | message=draw( | ||||
| 'message': draw( | |||||
| text(min_size=20, max_size=100).map(lambda t: t.encode())), | text(min_size=20, max_size=100).map(lambda t: t.encode())), | ||||
| 'date': draw(new_swh_date()), | date=TimestampWithTimezone.from_datetime(draw(new_swh_date())), | ||||
| 'committer_date': draw(new_swh_date()), | committer_date=TimestampWithTimezone.from_datetime( | ||||
| 'synthetic': False, | draw(new_swh_date())), | ||||
| 'type': 'git', | synthetic=False, | ||||
| 'parents': [], | type='git', | ||||
| 'metadata': [], | ) | ||||
| } | |||||
| def revisions(min_size=2, max_size=8): | def revisions(min_size=2, max_size=8): | ||||
| """ | """ | ||||
| Hypothesis strategy returning random revisions ingested | Hypothesis strategy returning random revisions ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return lists(revision(), min_size=min_size, max_size=max_size) | return lists(revision(), min_size=min_size, max_size=max_size) | ||||
| Show All 13 Lines | def snapshot(): | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return _known_swh_object('snapshots') | return _known_swh_object('snapshots') | ||||
| def new_snapshots(nb_snapshots=None): | def new_snapshots(nb_snapshots=None): | ||||
| min_size = nb_snapshots if nb_snapshots else 2 | min_size = nb_snapshots if nb_snapshots else 2 | ||||
| max_size = nb_snapshots if nb_snapshots else 8 | max_size = nb_snapshots if nb_snapshots else 8 | ||||
| return lists(new_snapshot(min_size=2, max_size=10, only_objects=True) | return lists(new_snapshot(min_size=2, max_size=10, only_objects=True), | ||||
| .map(lambda snp: snp.to_dict()), | |||||
| min_size=min_size, max_size=max_size) | min_size=min_size, max_size=max_size) | ||||
| def unknown_snapshot(): | def unknown_snapshot(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random revision not ingested | Hypothesis strategy returning a random revision not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| ▲ Show 20 Lines • Show All 123 Lines • Show Last 20 Lines | |||||