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.storage.algos.dir_iterators import dir_iterator | |||||
| 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 134 Lines • ▼ Show 20 Lines | |||||
| def unknown_content(): | def unknown_content(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random content not ingested | Hypothesis strategy returning a random content not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return new_content().filter( | return new_content().filter( | ||||
| lambda c: next(get_tests_data()['storage'].content_get( | lambda c: next(get_tests_data()['storage'].content_get( | ||||
| [hash_to_bytes(c['sha1'])])) is None) | [hash_to_bytes(c['sha1'])])) is None) | ||||
| def unknown_contents(): | def unknown_contents(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning random contents not ingested | Hypothesis strategy returning random contents not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return lists(unknown_content(), min_size=2, max_size=8) | return lists(unknown_content(), min_size=2, max_size=8) | ||||
| ▲ Show 20 Lines • Show All 73 Lines • ▼ Show 20 Lines | |||||
| 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().map(lambda origin: origin.to_dict()).filter( | ||||
| lambda origin: get_tests_data()['storage'].origin_get( | lambda origin: get_tests_data()['storage'].origin_get( | ||||
| [origin])[0] is None) | [origin])[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 139 Lines • ▼ Show 20 Lines | 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. | ||||
| """ | """ | ||||
| return sha1().filter( | return sha1().filter( | ||||
| lambda s: get_tests_data()['storage'].snapshot_get( | lambda s: get_tests_data()['storage'].snapshot_get( | ||||
| hash_to_bytes(s)) is None) | hash_to_bytes(s)) is None) | ||||
| def _get_origin_dfs_revisions_walker(): | def _get_snapshot_head_revision(snapshot): | ||||
| tests_data = get_tests_data() | |||||
| storage = tests_data['storage'] | |||||
| origin = random.choice(tests_data['origins'][:-1]) | |||||
| snapshot = storage.snapshot_get_latest(origin['url']) | |||||
| if snapshot['branches'][b'HEAD']['target_type'] == 'alias': | if snapshot['branches'][b'HEAD']['target_type'] == 'alias': | ||||
| target = snapshot['branches'][b'HEAD']['target'] | target = snapshot['branches'][b'HEAD']['target'] | ||||
| head = snapshot['branches'][target]['target'] | head = snapshot['branches'][target]['target'] | ||||
| else: | else: | ||||
| head = snapshot['branches'][b'HEAD']['target'] | head = snapshot['branches'][b'HEAD']['target'] | ||||
| return head | |||||
| def _get_origin_dfs_revisions_walker(): | |||||
| tests_data = get_tests_data() | |||||
| storage = tests_data['storage'] | |||||
| origin = random.choice(tests_data['origins'][:-2]) | |||||
| snapshot = storage.snapshot_get_latest(origin['url']) | |||||
| head = _get_snapshot_head_revision(snapshot) | |||||
| return get_revisions_walker('dfs', storage, head) | return get_revisions_walker('dfs', storage, head) | ||||
| def ancestor_revisions(): | def ancestor_revisions(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a pair of revisions ingested into the | Hypothesis strategy returning a pair of revisions ingested into the | ||||
| test archive with an ancestor relation. | test archive with an ancestor relation. | ||||
| """ | """ | ||||
| ▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines | def revision_with_submodules(): | ||||
| Hypothesis strategy returning a revision that is known to | Hypothesis strategy returning a revision that is known to | ||||
| point to a directory with revision entries (aka git submodule) | point to a directory with revision entries (aka git submodule) | ||||
| """ | """ | ||||
| return just({ | return just({ | ||||
| 'rev_sha1_git': 'ffcb69001f3f6745dfd5b48f72ab6addb560e234', | 'rev_sha1_git': 'ffcb69001f3f6745dfd5b48f72ab6addb560e234', | ||||
| 'rev_dir_sha1_git': 'd92a21446387fa28410e5a74379c934298f39ae2', | 'rev_dir_sha1_git': 'd92a21446387fa28410e5a74379c934298f39ae2', | ||||
| 'rev_dir_rev_path': 'libtess2' | 'rev_dir_rev_path': 'libtess2' | ||||
| }) | }) | ||||
| def origin_with_directory_named_directory(): | |||||
| tests_data = get_tests_data() | |||||
| storage = tests_data['storage'] | |||||
| origins = [] | |||||
| for origin in tests_data['origins']: | |||||
| snapshot = storage.snapshot_get_latest(origin['url']) | |||||
| head = _get_snapshot_head_revision(snapshot) | |||||
| head_data = list(storage.revision_get([head]))[0] | |||||
| for entry in dir_iterator(storage, head_data['directory']): | |||||
| if entry['type'] == 'dir' and entry['name'] == b'directory': | |||||
| origins.append({ | |||||
| 'origin': origin['url'], | |||||
| 'path': entry['path'].decode('utf-8') | |||||
| }) | |||||
| return sampled_from(origins) | |||||