Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/strategies.py
| # Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import random | import random | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| from datetime import datetime | from datetime import datetime | ||||
| 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, | ||||
| integers, 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.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 | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 205 Lines • ▼ Show 20 Lines | def origin_with_release(): | ||||
| for origin in tests_data['origins']: | for origin in tests_data['origins']: | ||||
| snapshot = tests_data['storage'].snapshot_get_latest(origin['url']) | snapshot = tests_data['storage'].snapshot_get_latest(origin['url']) | ||||
| 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(): | |||||
| """ | |||||
| Hypothesis strategy returning a random origin id not ingested | |||||
| into the test archive. | |||||
| """ | |||||
| 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'] | 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) | ||||
| ▲ Show 20 Lines • Show All 269 Lines • Show Last 20 Lines | |||||