Created
June 18, 2019 04:57
-
-
Save setazer/dc2dca70d02963b5cdc427d992f38e94 to your computer and use it in GitHub Desktop.
Код с комментариями из видео" Продвинутое использование py.test, Андрей Светлов, Python Core Developer" https://www.youtube.com/watch?v=7KgihdKTWY4
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import pytest, random | |
# Константные фикстуры | |
@pytest.fixture # фикстура выполняется 1 раз (при каждом обращении - результат фиксирован) | |
def rnd(): | |
return random.random() | |
@pytest.fixture | |
def fixture_a(rnd): # имена аргументов должны совпадать с фикстурами | |
return rnd | |
@pytest.fixture | |
def fixture_b(rnd): | |
return rnd | |
def test_mono(fixture_a, fixture_b): | |
assert fixture_a == fixture_b | |
# Динамические фикстуры | |
@pytest.fixture | |
def rnd_gen(): | |
return random.Random(123456) | |
@pytest.fixture # когда нужно чтобы фикстура возвращала новое значение после подсчёта | |
def make_rnd(rnd_gen): | |
def maker(): | |
return rnd_gen.random() | |
return maker | |
@pytest.fixture | |
def fixture_a(make_rnd): | |
return make_rnd() | |
@pytest.fixture | |
def fixture_b(make_rnd): | |
return make_rnd() | |
def test_multi(fixture_a, fixture_b): | |
assert fixture_a == fixture_b | |
# Очистка ресурсов (фиксированное открытие) | |
@pytest.yield_fixture | |
def opened_file(): | |
f = open("filename") | |
try: | |
yield f | |
finally: | |
f.close() | |
def test_fixed_file(opened_file): | |
assert opened_file.read() == "file content" | |
# Фабрика для открытия файлов (множественное открытие) | |
@pytest.yield_fixture | |
def open_file(): | |
f = None | |
def opener(filename): | |
nonlocal f | |
assert f is None | |
f = open(filename) | |
return f | |
yield opener | |
if f is not None: | |
f.close() | |
def test_mul_files(open_file): | |
assert open_file("file_a.txt").read() == "Content A" | |
assert open_file("file_b.txt").read() == "Content B" | |
# Работа с базами (yield_fixture и контекст-менеджеры) | |
@pytest.yield_fixture | |
def redis(): | |
with Redis() as redis: | |
yield redis | |
@pytest.yield_fixture | |
def db(): | |
with sqlite3.connect(':memory:') as db: | |
yield db | |
def test_bases(db, redis): | |
db.execute() | |
redis.set() | |
# Транзакции | |
@pytest.yield_fixture | |
def transaction(db): | |
cursor=db.cursor() | |
try: | |
yield cursor | |
db.commit() | |
except: | |
db.rollback() | |
def test_transactions(transaction): | |
transaction.execute('SOME SQL') | |
raise RuntimeError | |
transaction.execute('SOME MORE SQL') | |
# conftest.py - файл для настройки pytest | |
# Начало файла conftest.py | |
import pytest | |
# загружает файлы redis_fixtures.py и db_fixtures.py - юзать вместо импорта! | |
pytest_plugins = ['redis_fixtures', 'db_fixtures'] | |
@pytest.fixture | |
def fixture_a(): | |
return "value" | |
# Конец файла conftest.py | |
# У фикстур могут быть разные области видимости | |
# function / class / module / session | |
# Начало примера с докером | |
import docker as libdocker | |
# Фикстура будет посчитана один раз на протяжении всей сессии тестирования | |
# Для запуска двух процессов одновременно | |
@pytest.fixture(scope='session') | |
def session_id(): | |
return str(uuid.uuid4()) # Для | |
@pytest.fixture(scope='session') | |
def unused_port(): | |
def f(): | |
with socket.socket(socket.AP_INET, socket.SOCK_STREAM) as s: | |
s.bind(('127.0.0.1',0)) | |
return s.getsocketname()[1] | |
return f() | |
@pytest.fixture(scope='session') | |
def docker(): | |
return libdocker.Client(version='auto') | |
# Запуск контейнера | |
@pytest.yield_fixture(scope='session') | |
def redis_server(unused_port, session_id, docker): | |
docker.pull('redis') | |
port = unused_port() | |
container = docker.create_container(image='redis', | |
name='test-redis-{}'.format(session_id), | |
ports=[6379], detach=True, | |
host_config=docker.create_host_config( | |
port_bindings={6739: port})) | |
docker.start(container=container['Id']) | |
ping_redis(port) | |
container['redis_port'] = port | |
yield container | |
docker.kill(container=container['Id']) | |
docker.remove_container(container['Id']) | |
# Пинг redis'а (пока не поднимется) | |
def ping_redis(port): | |
timeout=0.001 | |
for i in range(100): | |
try: | |
client = redis.StrictRedis(host='127.0.0.1', port=port, db=0) | |
except redis.Connection.Error: | |
time.sleep(timeout) | |
timeout *= 2 | |
else: | |
client.close() | |
else: | |
raise RuntimeError("Cannot connect to redis") | |
# Клиент redis'а | |
@pytest.yield_fixture | |
def redis_client(redis_server): | |
client = redis.StrictRedis(host='127.0.0.1', | |
port=redis_server['redis_port'], | |
db=0) | |
yield client | |
client.flushdb() | |
client.close() | |
def test_docker(redis_client): | |
redis_client.set(b'key', b'value') | |
assert redis_client.get(b'key') == b'value' | |
# Конец примера с докером | |
# Проверка версии питона | |
# Проблема в том что синтаксис должен быть правильным для запускаемой версии питона | |
# Те же async/await отсутствуют на старых версиях, поэтому тест не компилируется. Решение - далее. | |
@pytest.mark.skipif(sys.version_info <(3,4,1), reason="Python<3.4.1 doesnt support __del__ calls from GC") | |
def test___del__(): | |
pass | |
# Исключение теста из выполнения | |
# Начинающееся с pytest_ - плагины | |
# Тесты лежащие в подпапке py35 не выполняются - не выходит ошибок интерпретатора из-за нового кода | |
def pytest_ignore_collect(path, config): | |
if 'py35' in str(path): | |
if sys.version_info < (3,5,0): | |
return True | |
# Добавление параметров командной строки для тестов | |
# Cинтаксис parse.addoption схож с argparse - добавляет новый параметр для ожидания в командной строке | |
def pytest_addoption(parser): | |
parser.addoption('--gc-collect', action='store_true', | |
default=False, help="Perform GC collection after every test") | |
@pytest.mark.trylast | |
def pytest_runtest_teardown(item, nextitem): # Запускается после каждого теста | |
if item.config.getoption('--gc-collect'): | |
gc.collect() | |
return nextitem | |
# Параметризация фикстур | |
# Тест выполняется для каждого элемента из params (можно повторять значения) | |
@pytest.yield_fixture(scope='session', params=['2.8','3.0']) | |
def redis_server(unused_port, session_id, docker, request): | |
redis_version = request.param # Параметр (текущий элемент из params) передается в request.param | |
# . . . | |
pass | |
# Продвинутая параметризация, "генерация фикстур" | |
def pytest_addoption(parser): | |
parser.addoption('--redis_version', action='append', default=[], | |
, help=("Redis server versions. " | |
"May be used several times. " | |
"Available values: 2.8, 3.0, 3.2, all")) | |
# Динамическая парметризация функций, при передаче параметров ком.строки | |
# тестирует по указанным версиям или только по 3.2 в противном случае | |
def pytest_generate_test(metafunc): | |
if 'redis_version' in metafunc.fixturenames: | |
tags = set(metafunc.config.option.redis_version) # Значения из командной строки | |
if not tags: | |
tags = ['3.2'] | |
elif 'all' in tags: | |
tags = ['2.8','3.0','3.2'] | |
else: | |
tags = list(tags) | |
metafunc.parametrize("redis_version",tags,scope='session') | |
# Использование версии: | |
@pytest.yield_fixture(scope='session') | |
def redis_server(unused_port, session_id, docker, redis_version): | |
image = 'redis:{}'.format(redis_version) | |
def pytest_addoption(parser): | |
parser.addoption('--run-slow',action='store_true',default=False,help="Run slow tests") | |
def pytest_runtest_setup(item): # Управление запуском тестов | |
if ('slowtest' in item.keywords and (not item.config.getoption('--run-slow'))): | |
pytest.skip('Need --run-slow to run') # Пропускаем тест при отсутствии параметра командной строки --run-slow | |
@pytest.mark.slowtest | |
def test_smth(): | |
pass | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment