Andrew Svetlov
http://asvetlov.blogspot.com
andrew.svetlov@gmail.com
http://asvetlov.github.io/pytest-moscow
“Eveybody is using py.test anyway...”
Guido van Rossum
import pytest
def test_a():
assert 1 != 2
class TestB:
def test_b(self):
assert 'a'.upper() == 'A'
def test_c(self):
with pytest.raises(ZeroDivisionError):
1/0
import random
@pytest.fixture
def rnd():
return random.random()
def test_rnd(rnd):
print(rnd)
@pytest.fixture
def rnd_gen():
return random.Random(123456)
@pytest.fixture
def rnd(rnd_gen):
return rnd_gen.random()
@pytest.fixture
def fixture_a(rnd):
return rnd
@pytest.fixture
def fixture_b(rnd):
return rnd
def test(fixture_a, fixture_b):
assert fixture_a == fixture_b # ???
@pytest.fixture
def make_rnd(rnd_gen):
def maker()
return rnd_gen.random()
return maker
@pytest.fixture
def fixture_a(rnd):
return rnd()
@pytest.fixture
def fixture_b(rnd):
return rnd()
def test(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_a(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_a(open_file):
assert open_file("file_a.txt").read() == "Content A"
assert open_file("file_b.txt").read() == "Content B"
class TestA(unittest.TestCase):
def setUp(self):
self.redis = redis.Redis()
self.db = db.connect(':memory:')
def tearDown(self):
self.redis.close()
self.db.close()
def test_a(self):
self.db.execute(...)
self.redis.set(...)
class RedisMixin:
def setUp(self):
self.redis = Redis()
super().setUp()
def tearDown(self):
self.redis.close()
super().tearDown()
class DBMixin:
def setUp(self):
self.db = sqlite3.connect(':memory:')
super().setUp()
def tearDown(self):
self.db.close()
super().tearDown()
class TestA(RedisMixin, DBMixin, unittest.TestCase):
def setUp(self):
self.val = 'value'
super().setUp()
def tearDown(self):
...
super().tearDown()
def test_a(self):
self.db.execute(...)
self.redis.set(...)
@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_a(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_b(transaction):
transaction.execute("INSERT INTO tbl(name) VALUES ('John')")
raise RuntimeError("Error text")
transaction.execute("INSERT INTO tbl(name) VALUES ('Mary')")
No __init__.py in tests folder
import pytest
pytest_plugins = ['redis_fixtures', 'db_fixtures']
@pytest.fixture
def fixture_a():
return 'value'
No fixture imports in conftest.py
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.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('127.0.0.1', 0))
return s.getsockname()[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={6379: port}))
docker.start(container=container['Id'])
ping_redis(port) # N.B.
container['redis_port'] = port
yield container
docker.kill(container=container['Id'])
docker.remove_container(container['Id'])
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.ConnectionError:
time.sleep(timeout)
timeout *= 2
else:
client.close()
else:
raise RuntimeError("Cannot connect to 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_redis(redis_client):
redis_client.set(b'key', b'value')
assert redis_client.get(b'key') == b'value'
@pytest.mark.skipif(sys.version_info < (3, 4, 1),
reason="Python<3.4.1 doesnt support "
"__del__ calls from GC")
def test___del__():
...
def pytest_ignore_collect(path, config):
if 'py35' in str(path):
if sys.version_info < (3, 5, 0):
return True
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
$ py.test --gc-collect
@pytest.yield_fixture(scope='session', params=['2.8', '3.0'])
def redis_server(unused_port, session_id, docker, request):
redis_version = request.param
image = 'redis:{}'.format(redis_version)
docker.pull(image)
container = docker.create_container(
image=image,
name='test-redis-{}-{}'.format(redis_version, session_id),
...)
...
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"))
def pytest_generate_tests(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)
docker.pull(image)
container = docker.create_container(
image=image,
name='test-redis-{}-{}'.format(redis_version, session_id),
...)
...
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')
@pytest.mark.slowtest
def test_xxx():
...
py.test --run-slow
Andrew Svetlov
http://asvetlov.blogspot.com
andrew.svetlov@gmail.com