Skip to content
Snippets Groups Projects
Commit b4193e23 authored by John Chilton's avatar John Chilton
Browse files

Improve tests.

 - Docstrings for AMQP tests.
 - Skip curl tests if pycurl is unavailable.
 - Time the AMQP test in case this is what is causing Travis to hang.
parent 38953f3c
No related branches found
No related tags found
No related merge requests found
"""Tests for ``pulsar.client.amqp_exchange``."""
import time import time
import threading import threading
from .test_utils import skip_unless_module
from pulsar.client import amqp_exchange from pulsar.client import amqp_exchange
from .test_utils import (
skip_unless_module,
timed,
)
TEST_CONNECTION = "memory://test_amqp" TEST_CONNECTION = "memory://test_amqp"
@skip_unless_module("kombu") @skip_unless_module("kombu")
@timed(15)
def test_amqp(): def test_amqp():
"""Test the client PulsarExchange abstraction with an in-memory connection."""
manager1_exchange = amqp_exchange.PulsarExchange(TEST_CONNECTION, "manager_test") manager1_exchange = amqp_exchange.PulsarExchange(TEST_CONNECTION, "manager_test")
manager3_exchange = amqp_exchange.PulsarExchange(TEST_CONNECTION, "manager3_test") manager3_exchange = amqp_exchange.PulsarExchange(TEST_CONNECTION, "manager3_test")
manager2_exchange = amqp_exchange.PulsarExchange(TEST_CONNECTION, "manager2_test") manager2_exchange = amqp_exchange.PulsarExchange(TEST_CONNECTION, "manager2_test")
...@@ -56,3 +63,6 @@ class TestThread(threading.Thread): ...@@ -56,3 +63,6 @@ class TestThread(threading.Thread):
raise AssertionError(msg) raise AssertionError(msg)
self.join(2) self.join(2)
__all__ = ["test_amqp"]
import os import os
from tempfile import NamedTemporaryFile
from pulsar.client.transport.standard import Urllib2Transport from pulsar.client.transport.standard import Urllib2Transport
from pulsar.client.transport.curl import PycurlTransport from pulsar.client.transport.curl import PycurlTransport
from pulsar.client.transport.curl import post_file from pulsar.client.transport.curl import post_file
from pulsar.client.transport.curl import get_file from pulsar.client.transport.curl import get_file
from pulsar.client.transport import get_transport from pulsar.client.transport import get_transport
from tempfile import NamedTemporaryFile
from .test_utils import files_server from .test_utils import files_server
from .test_utils import skip_unless_module
def test_urllib_transports(): def test_urllib_transports():
_test_transport(Urllib2Transport()) _test_transport(Urllib2Transport())
@skip_unless_module("pycurl")
def test_pycurl_transport(): def test_pycurl_transport():
_test_transport(PycurlTransport()) _test_transport(PycurlTransport())
...@@ -37,6 +40,7 @@ def _test_transport(transport): ...@@ -37,6 +40,7 @@ def _test_transport(transport):
assert open(output_path, 'r').read().find("Test123") >= 0 assert open(output_path, 'r').read().find("Test123") >= 0
@skip_unless_module("pycurl")
def test_curl_put_get(): def test_curl_put_get():
with files_server() as (server, directory): with files_server() as (server, directory):
server_url = server.application_url server_url = server.application_url
...@@ -74,6 +78,7 @@ def test_curl_status_code(): ...@@ -74,6 +78,7 @@ def test_curl_status_code():
assert exception_raised assert exception_raised
@skip_unless_module("pycurl")
def test_curl_problems(): def test_curl_problems():
with files_server() as (server, directory): with files_server() as (server, directory):
server_url = server.application_url server_url = server.application_url
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment