Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/strategies.py
| Show All 26 Lines | |||||
| # Some of these data are sampled from a test archive created and populated | # Some of these data are sampled from a test archive created and populated | ||||
| # in the swh.web.tests.data module. | # in the swh.web.tests.data module. | ||||
| # Set the swh-web hypothesis profile if none has been explicitly set | # Set the swh-web hypothesis profile if none has been explicitly set | ||||
| hypothesis_default_settings = settings.get_profile('default') | hypothesis_default_settings = settings.get_profile('default') | ||||
| if repr(settings()) == repr(hypothesis_default_settings): | if repr(settings()) == repr(hypothesis_default_settings): | ||||
| settings.load_profile('swh-web') | settings.load_profile('swh-web') | ||||
| # Import tests data | |||||
| tests_data = get_tests_data() | |||||
| storage = tests_data['storage'] | |||||
| # The following strategies exploit the hypothesis capabilities | # The following strategies exploit the hypothesis capabilities | ||||
| _generated_checksums = set() | |||||
| def _filter_checksum(cs): | def _filter_checksum(cs): | ||||
| generated_checksums = get_tests_data()['generated_checksums'] | |||||
| if not int.from_bytes(cs, byteorder='little') or \ | if not int.from_bytes(cs, byteorder='little') or \ | ||||
| cs in _generated_checksums: | cs in generated_checksums: | ||||
| return False | return False | ||||
| _generated_checksums.add(cs) | generated_checksums.add(cs) | ||||
| return True | return True | ||||
| def _known_swh_object(object_type): | def _known_swh_object(object_type): | ||||
| return sampled_from(tests_data[object_type]) | return sampled_from(get_tests_data()[object_type]) | ||||
| def sha1(): | def sha1(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a valid hexadecimal sha1 value. | Hypothesis strategy returning a valid hexadecimal sha1 value. | ||||
| """ | """ | ||||
| return binary( | return binary( | ||||
| min_size=20, max_size=20).filter(_filter_checksum).map(hash_to_hex) | min_size=20, max_size=20).filter(_filter_checksum).map(hash_to_hex) | ||||
| ▲ Show 20 Lines • Show All 84 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(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) | ||||
| def directory(): | def directory(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random directory ingested | Hypothesis strategy returning a random directory ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return _known_swh_object('directories') | return _known_swh_object('directories') | ||||
| def directory_with_subdirs(): | def directory_with_subdirs(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random directory containing | Hypothesis strategy returning a random directory containing | ||||
| sub directories ingested into the test archive. | sub directories ingested into the test archive. | ||||
| """ | """ | ||||
| storage = get_tests_data()['storage'] | |||||
| return directory().filter( | return directory().filter( | ||||
| lambda d: any([e['type'] == 'dir' | lambda d: any([e['type'] == 'dir' | ||||
| for e in list(storage.directory_ls(hash_to_bytes(d)))])) | for e in list(storage.directory_ls(hash_to_bytes(d)))])) | ||||
| def empty_directory(): | def empty_directory(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning the empty directory ingested | Hypothesis strategy returning the empty directory ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return just(directory_identifier({'entries': []})) | return just(directory_identifier({'entries': []})) | ||||
| def unknown_directory(): | def unknown_directory(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random directory not ingested | Hypothesis strategy returning a random directory not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| storage = get_tests_data()['storage'] | |||||
| return sha1().filter( | return sha1().filter( | ||||
| lambda s: len(list(storage.directory_missing([hash_to_bytes(s)]))) > 0) | lambda s: len(list(storage.directory_missing([hash_to_bytes(s)]))) > 0) | ||||
| def origin(): | def origin(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random origin ingested | Hypothesis strategy returning a random origin ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return _known_swh_object('origins') | return _known_swh_object('origins') | ||||
| def origin_with_multiple_visits(): | def origin_with_multiple_visits(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random origin ingested | Hypothesis strategy returning a random origin ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| ret = [] | ret = [] | ||||
| tests_data = get_tests_data() | |||||
| for origin in tests_data['origins']: | for origin in tests_data['origins']: | ||||
| visits = list(storage.origin_visit_get(origin['id'])) | visits = list(tests_data['storage'].origin_visit_get(origin['id'])) | ||||
| if len(visits) > 1: | if len(visits) > 1: | ||||
| ret.append(origin) | ret.append(origin) | ||||
| return sampled_from(ret) | return sampled_from(ret) | ||||
| def origin_with_release(): | def origin_with_release(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random origin ingested | Hypothesis strategy returning a random origin ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| ret = [] | ret = [] | ||||
| tests_data = get_tests_data() | |||||
| for origin in tests_data['origins']: | for origin in tests_data['origins']: | ||||
| snapshot = storage.snapshot_get_latest(origin['id']) | snapshot = tests_data['storage'].snapshot_get_latest(origin['id']) | ||||
| if any([b['target_type'] == 'release' | if any([b['target_type'] == 'release' | ||||
| for b in snapshot['branches'].values()]): | for b in snapshot['branches'].values()]): | ||||
| ret.append(origin) | ret.append(origin) | ||||
| return sampled_from(ret) | return sampled_from(ret) | ||||
| def unknown_origin_id(): | def unknown_origin_id(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random origin id not ingested | Hypothesis strategy returning a random origin id not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return integers(min_value=1000000) | return integers(min_value=1000000) | ||||
| 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. | ||||
| """ | """ | ||||
| storage = get_tests_data()['storage'] | |||||
| return new_origin_strategy().map(lambda origin: origin.to_dict()).filter( | return new_origin_strategy().map(lambda origin: origin.to_dict()).filter( | ||||
| lambda origin: storage.origin_get([origin])[0] is None) | lambda origin: storage.origin_get([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. | ||||
| Show All 25 Lines | |||||
| def unknown_release(): | def unknown_release(): | ||||
| """ | """ | ||||
| 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: next(storage.release_get([s])) is None) | lambda s: next(get_tests_data()['storage'].release_get([s])) is None) | ||||
| def revision(): | def revision(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random revision ingested | Hypothesis strategy returning a random revision ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return _known_swh_object('revisions') | return _known_swh_object('revisions') | ||||
| def unknown_revision(): | def unknown_revision(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random revision not ingested | Hypothesis strategy returning a random revision not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| storage = get_tests_data()['storage'] | |||||
| return sha1().filter( | return sha1().filter( | ||||
| lambda s: next(storage.revision_get([hash_to_bytes(s)])) is None) | lambda s: next(storage.revision_get([hash_to_bytes(s)])) is None) | ||||
| @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. | ||||
| ▲ Show 20 Lines • Show All 78 Lines • ▼ Show 20 Lines | return lists(new_snapshot(min_size=2, max_size=10, only_objects=True) | ||||
| 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. | ||||
| """ | """ | ||||
| storage = get_tests_data()['storage'] | |||||
| return sha1().filter( | return sha1().filter( | ||||
| lambda s: storage.snapshot_get(hash_to_bytes(s)) is None) | lambda s: storage.snapshot_get(hash_to_bytes(s)) is None) | ||||
| def person(): | def person(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random person ingested | Hypothesis strategy returning a random person ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return _known_swh_object('persons') | return _known_swh_object('persons') | ||||
| def unknown_person(): | def unknown_person(): | ||||
| """ | """ | ||||
| Hypothesis strategy returning a random person not ingested | Hypothesis strategy returning a random person not ingested | ||||
| into the test archive. | into the test archive. | ||||
| """ | """ | ||||
| return integers(min_value=1000000) | return integers(min_value=1000000) | ||||
| def _get_origin_dfs_revisions_walker(): | def _get_origin_dfs_revisions_walker(): | ||||
| tests_data = get_tests_data() | |||||
| storage = tests_data['storage'] | |||||
| origin = random.choice(tests_data['origins'][:-1]) | origin = random.choice(tests_data['origins'][:-1]) | ||||
| snapshot = storage.snapshot_get_latest(origin['id']) | snapshot = storage.snapshot_get_latest(origin['id']) | ||||
anlambert: Just a nitpick, maybe use a variable to store the tests data here instead of calling the… | |||||
| head = snapshot['branches'][b'HEAD']['target'] | head = snapshot['branches'][b'HEAD']['target'] | ||||
| 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 101 Lines • Show Last 20 Lines | |||||
Just a nitpick, maybe use a variable to store the tests data here instead of calling the get_tests_data function multiple times