Skip to content

Get rid of using external resource in tests #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jul 19, 2015
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 21 additions & 41 deletions tests/test_simple.py
Original file line number Diff line number Diff line change
@@ -1,41 +1,29 @@
"""Quick'n'dirty unit tests for provided fixtures and markers."""
import asyncio
import os
import urllib
import pytest


@asyncio.coroutine
def simple_http_client(url):
"""Just a simple http client, for testing."""
u = urllib.parse.urlparse(url)
port = u.port if u.port else 80
r, w = yield from asyncio.open_connection(host=u.netloc, port=port)
w.write(b'GET ' + u.path.encode() + b' HTTP/1.0\r\n')
w.write(b'Host: ' + u.netloc.encode() + b'\r\n')
w.write(b'\r\n')
yield from w.drain()

resp = yield from r.read()

w.close()
return resp
def async_coro(loop):
yield from asyncio.sleep(0, loop=loop)
return 'ok'


def test_event_loop_fixture(event_loop):
"""Test the injection of the event_loop fixture."""
assert event_loop
url = 'http://httpbin.org/get'
resp = event_loop.run_until_complete(simple_http_client(url))
assert b'HTTP/1.1 200 OK' in resp
ret = event_loop.run_until_complete(async_coro(event_loop))
assert ret == 'ok'


def test_event_loop_processpool_fixture(event_loop_process_pool):
"""Test the injection of the event_loop with a process pool fixture."""
assert event_loop_process_pool
url = 'http://httpbin.org/get'
resp = event_loop_process_pool.run_until_complete(simple_http_client(url))
assert b'HTTP/1.1 200 OK' in resp

ret = event_loop_process_pool.run_until_complete(
async_coro(event_loop_process_pool))
assert ret == 'ok'

this_pid = os.getpid()
future = event_loop_process_pool.run_in_executor(None, os.getpid)
Expand All @@ -46,45 +34,38 @@ def test_event_loop_processpool_fixture(event_loop_process_pool):
@pytest.mark.asyncio
def test_asyncio_marker():
"""Test the asyncio pytest marker."""
url = 'http://httpbin.org/get'
resp = yield from simple_http_client(url)
assert b'HTTP/1.1 200 OK' in resp
yield # sleep(0)


@pytest.mark.asyncio
def test_asyncio_marker_with_default_param(a_param=None):
"""Test the asyncio pytest marker."""
url = 'http://httpbin.org/get'
resp = yield from simple_http_client(url)
assert b'HTTP/1.1 200 OK' in resp
yield # sleep(0)


@pytest.mark.asyncio_process_pool
def test_asyncio_process_pool_marker(event_loop):
"""Test the asyncio pytest marker."""
url = 'http://httpbin.org/get'
resp = yield from simple_http_client(url)
assert b'HTTP/1.1 200 OK' in resp

this_pid = os.getpid()
pool_pid = yield from event_loop.run_in_executor(None, os.getpid)
assert this_pid != pool_pid
ret = yield from async_coro(event_loop)
assert ret == 'ok'


@pytest.mark.asyncio
def test_unused_port_fixture(unused_tcp_port):
def test_unused_port_fixture(unused_tcp_port, event_loop):
"""Test the unused TCP port fixture."""

@asyncio.coroutine
def closer(_, writer):
writer.close()

server1 = yield from asyncio.start_server(closer, host='localhost',
port=unused_tcp_port)
port=unused_tcp_port,
loop=event_loop)

with pytest.raises(IOError):
yield from asyncio.start_server(closer, host='localhost',
port=unused_tcp_port)
port=unused_tcp_port,
loop=event_loop)

server1.close()
yield from server1.wait_closed()
Expand All @@ -94,8 +75,7 @@ class Test:
"""Test that asyncio marked functions work in test methods."""

@pytest.mark.asyncio
def test_asyncio_marker_method(self):
def test_asyncio_marker_method(self, event_loop):
"""Test the asyncio pytest marker in a Test class."""
url = 'http://httpbin.org/get'
resp = yield from simple_http_client(url)
assert b'HTTP/1.1 200 OK' in resp
ret = yield from async_coro(event_loop)
assert ret == 'ok'