diff --git a/.travis.yml b/.travis.yml index 68411a7..65cfb07 100644 --- a/.travis.yml +++ b/.travis.yml @@ -2,18 +2,24 @@ language: python python: - "2.7" -install: - - "sudo apt-get install cups cups-pdf" - - "python setup.py install" - - "pip install coverage coveralls pytest mock" +env: + - NO_TORNADO=1 + - NO_TORNADO=0 + +before_install: + - "sudo apt-get install -qq cups cups-pdf" - "curl http://www.cups.org/software/ipptool/ipptool-20130731-linux-ubuntu-x86_64.tar.gz | tar xvzf -" - "echo \"[main]\nipptool_path = \"$TRAVIS_BUILD_DIR\"/ipptool-20130731/ipptool\" > ~/.pyipptool.cfg" - "echo \"$USER:travis\" | sudo chpasswd" + - "sudo pip install -U pip setuptools" + +install: + - "pip install -e . pytest-cov coveralls pytest mock tornado pkipplib" -script: "coverage run --source=pyipptool setup.py test" +script: "py.test --cov pyipptool --ignore ipptool-20130731" after_success: - coveralls + "if [[ NO_TORNADO == '1' ]]; then coveralls; fi" notifications: email: false diff --git a/pyipptool/core.py b/pyipptool/core.py index f3910c6..39c018e 100644 --- a/pyipptool/core.py +++ b/pyipptool/core.py @@ -1,3 +1,4 @@ +import functools import os import plistlib import subprocess @@ -34,15 +35,28 @@ cancel_subscription_form, ) +try: + if os.getenv('NO_TORNADO', '').lower() in ('1', 'yes', 'true', 't'): + raise ImportError + from tornado.gen import coroutine, Return, Task + from tornado.ioloop import TimeoutError +except ImportError: + HAS_TORNADO = False -class TimeoutError(Exception): - pass + def coroutine(f): + return f + + class TimeoutError(Exception): + pass +else: + HAS_TORNADO = True class IPPToolWrapper(object): - def __init__(self, config): + def __init__(self, config, io_loop=None): self.config = config + self.io_loop = io_loop def authenticate_uri(self, uri): if 'login' in self.config and 'password' in self.config: @@ -70,6 +84,42 @@ def timeout_handler(self, process, future): time.sleep(.1) def _call_ipptool(self, uri, request): + if HAS_TORNADO: + return self._async_call_ipptool(uri, request) + return self._sync_call_ipptool(uri, request) + + @coroutine + def cleanup_fd(self, name): + try: + os.unlink(name) + except OSError: + pass + + @coroutine + def _async_call_ipptool(self, uri, request): + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + temp_file.write(request) + from tornado.process import Subprocess + process = Subprocess([self.config['ipptool_path'], + self.authenticate_uri(uri), '-X', + temp_file.name], + stdin=subprocess.PIPE, + stdout=Subprocess.STREAM, + stderr=Subprocess.STREAM, + io_loop=self.io_loop) + future = [] + self.io_loop.add_timeout(self.io_loop.time() + self.config['timeout'], + functools.partial(self.timeout_handler, + process.proc, future)) + stdout, stderr = yield [Task(process.stdout.read_until_close), + Task(process.stderr.read_until_close)] + if future: + raise TimeoutError + yield self.cleanup_fd(temp_file.name) + + raise Return(plistlib.readPlistFromString(stdout)) + + def _sync_call_ipptool(self, uri, request): with tempfile.NamedTemporaryFile(delete=False) as temp_file: temp_file.write(request) process = subprocess.Popen([self.config['ipptool_path'], diff --git a/setup.py b/setup.py index 6dff958..5d79e6d 100644 --- a/setup.py +++ b/setup.py @@ -37,9 +37,10 @@ def read_that_file(path): long_description=description, license=' Apache Software License', packages=('pyipptool',), - install_requires=('deform',), + install_requires=('deform>=2.0a2',), + extra_requires={'Tornado': ('tornado', 'futures')}, tests_require=('mock', 'pytest', 'coverage', - 'pytest-cov', 'coveralls'), + 'pytest-cov', 'coveralls', 'pkipplib', 'tornado'), include_package_data=True, test_suite='tests', cmdclass = {'test': PyTest}, diff --git a/tests/test_async_subprocess.py b/tests/test_async_subprocess.py new file mode 100644 index 0000000..a1e6676 --- /dev/null +++ b/tests/test_async_subprocess.py @@ -0,0 +1,156 @@ +import BaseHTTPServer +import os +import SocketServer +import socket +import threading +import time + +from pkipplib import pkipplib +import pytest +import tornado.testing + + +NO_TORNADO = os.getenv('NO_TORNADO', '').lower() in ('1', 'yes', 'true', 't') + + +@pytest.mark.skipif(NO_TORNADO, reason='requires tornado') +class AsyncSubprocessTestCase(tornado.testing.AsyncTestCase): + + @tornado.testing.gen_test + def test_async_call(self): + from pyipptool import wrapper + from pyipptool.forms import get_subscriptions_form + + class Handler(BaseHTTPServer.BaseHTTPRequestHandler): + """ + HTTP Handler that will make ipptool waiting + """ + protocol_version = 'HTTP/1.1' + + def do_POST(self): + # return a real IPP Response thanks to pkipplib + ipp_request = pkipplib.IPPRequest( + self.rfile.read( + int(self.headers.getheader('content-length')))) + ipp_request.parse() + try: + self.send_response(200) + self.send_header('Content-type', 'application/ipp') + self.end_headers() + ipp_response = pkipplib.IPPRequest( + operation_id=pkipplib.IPP_OK, + request_id=ipp_request.request_id) + ipp_response.operation['attributes-charset'] =\ + ('charset', 'utf-8') + ipp_response.operation['attributes-natural-language'] =\ + ('naturalLanguage', 'en-us') + self.wfile.write(ipp_response.dump()) + finally: + assassin = threading.Thread(target=self.server.shutdown) + assassin.daemon = True + assassin.start() + + PORT = 6789 + while True: + try: + httpd = SocketServer.TCPServer(("", PORT), Handler) + except socket.error as exe: + if exe.errno in (48, 98): + PORT += 1 + else: + raise + else: + break + httpd.allow_reuse_address = True + + thread = threading.Thread(target=httpd.serve_forever) + thread.daemon = True + thread.start() + + request = get_subscriptions_form.render( + {'header': + {'operation_attributes': + {'printer_uri': 'http://localhost:%s/printers/fake' % PORT}}} + ) + + try: + wrapper.io_loop = self.io_loop + response = yield wrapper._call_ipptool( + 'http://localhost:%s/' % PORT, request) + finally: + wrapper.io_loop = None + try: + del response['Tests'][0]['RequestId'] + except KeyError: + self.fail(response) + assert response == {'Successful': True, + 'Tests': + [{'Name': 'Get Subscriptions', + 'Operation': 'Get-Subscriptions', + 'Version': '1.1', + 'RequestAttributes': + [{'attributes-charset': 'utf-8', + 'attributes-natural-language': 'en', + 'printer-uri': + 'http://localhost:%s/printers/fake' % PORT} + ], + 'ResponseAttributes': + [{'attributes-charset': 'utf-8', + 'attributes-natural-language': 'en-us'} + ], + 'StatusCode': 'successful-ok', + 'Successful': True}], + 'Transfer': 'auto', + 'ipptoolVersion': 'CUPS v1.7.0'}, response + + @tornado.testing.gen_test + def test_async_timeout_call(self): + from pyipptool import wrapper + from pyipptool.core import TimeoutError + from pyipptool.forms import get_subscriptions_form + + class Handler(BaseHTTPServer.BaseHTTPRequestHandler): + """ + HTTP Handler that will make ipptool waiting + """ + protocol_version = 'HTTP/1.1' + + def do_POST(self): + time.sleep(0.2) + assassin = threading.Thread(target=self.server.shutdown) + assassin.daemon = True + assassin.start() + + PORT = 6789 + while True: + try: + httpd = SocketServer.TCPServer(("", PORT), Handler) + except socket.error as exe: + if exe.errno in (48, 98): + PORT += 1 + else: + raise + else: + break + httpd.allow_reuse_address = True + + thread = threading.Thread(target=httpd.serve_forever) + thread.daemon = True + thread.start() + + request = get_subscriptions_form.render( + {'header': + {'operation_attributes': + {'printer_uri': 'http://localhost:%s/printers/fake' % PORT}}} + ) + + try: + old_timeout = wrapper.config['timeout'] + wrapper.config['timeout'] = .1 + wrapper.io_loop = self.io_loop + with pytest.raises(TimeoutError): + yield wrapper._call_ipptool('http://localhost:%s/' % PORT, + request) + finally: + wrapper.io_loop = None + wrapper.config['timeout'] = old_timeout diff --git a/tests/test_form.py b/tests/test_form.py index 9b80d6b..f3c4863 100644 --- a/tests/test_form.py +++ b/tests/test_form.py @@ -1,6 +1,11 @@ +import os + import pytest +NO_TORNADO = os.getenv('NO_TORNADO', '').lower() in ('1', 'yes', 'true', 't') + +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cancel_job_form(): from pyipptool.forms import cancel_job_form request = cancel_job_form.render( @@ -17,6 +22,7 @@ def test_cancel_job_form(): assert 'ATTR boolean purge-job 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_release_job_form_with_job_id(): from pyipptool.forms import release_job_form request = release_job_form.render( @@ -29,6 +35,7 @@ def test_release_job_form_with_job_id(): assert 'ATTR integer job-id 7' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_release_job_form_with_job_uri(): from pyipptool.forms import release_job_form request = release_job_form.render( @@ -39,6 +46,7 @@ def test_release_job_form_with_job_uri(): assert 'ATTR uri job-uri https://localhost:631/jobs/7' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_create_printer_subscription_form(): from pyipptool.forms import create_printer_subscription_form request = create_printer_subscription_form.render( @@ -67,6 +75,7 @@ def test_create_printer_subscription_form(): assert 'ATTR integer notify-time-interval 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_create_job_subscription_form_for_pull_delivery_method(): from pyipptool.forms import create_job_subscription_form request = create_job_subscription_form.render( @@ -89,12 +98,14 @@ def test_create_job_subscription_form_for_pull_delivery_method(): assert 'ATTR uri printer-uri https://localhost:631/printer/p' in request assert 'ATTR integer notify-job-id 12' in request assert 'ATTR uri notify-recipient-uri rss://' in request - assert 'ATTR keyword notify-events job-completed,job-created,job-progress' in request + assert ('ATTR keyword notify-events job-completed,job-created,job-progress' + in request), request assert 'ATTR charset notify-charset utf-8' in request assert 'ATTR language notify-natural-language de' in request assert 'ATTR integer notify-time-interval 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_add_modify_class_form(): from pyipptool.forms import cups_add_modify_class_form m_uri_0 = 'ipp://localhost:631/printers/p0' @@ -125,6 +136,7 @@ def test_cups_add_modify_class_form(): assert 'ATTR name requesting-user-name-allowed me' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_add_modify_printer_form(): from pyipptool.forms import cups_add_modify_printer_form request = cups_add_modify_printer_form.render( @@ -162,6 +174,7 @@ def test_cups_add_modify_printer_form(): assert 'ATTR name requesting-user-name-allowed me' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_add_modify_printer_form_with_None(): from pyipptool.forms import cups_add_modify_printer_form with pytest.raises(ValueError) as exec_info: @@ -174,6 +187,7 @@ def test_cups_add_modify_printer_form_with_None(): " 'printer_state_message'") +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_delete_printer_form(): from pyipptool.forms import cups_delete_printer_form request = cups_delete_printer_form.render( @@ -185,6 +199,7 @@ def test_cups_delete_printer_form(): assert 'ATTR uri printer-uri https://localhost:631/printers/p0' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_delete_class_form(): from pyipptool.forms import cups_delete_class_form request = cups_delete_class_form.render( @@ -196,6 +211,7 @@ def test_cups_delete_class_form(): assert 'ATTR uri printer-uri https://localhost:631/classes/p0' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_get_classes_form(): from pyipptool.forms import cups_get_classes_form request = cups_get_classes_form.render( @@ -220,6 +236,7 @@ def test_cups_get_classes_form(): assert 'ATTR name requested-user-name john' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_get_devices_form(): from pyipptool.forms import cups_get_devices_form request = cups_get_devices_form.render( @@ -240,6 +257,7 @@ def test_cups_get_devices_form(): assert 'ATTR integer timeout 12' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_get_ppds_form(): from pyipptool.forms import cups_get_ppds_form request = cups_get_ppds_form.render( @@ -270,6 +288,7 @@ def test_cups_get_ppds_form(): assert 'ATTR integer limit 3' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_get_printers_form(): from pyipptool.forms import cups_get_printers_form request = cups_get_printers_form.render( @@ -294,6 +313,7 @@ def test_cups_get_printers_form(): assert 'ATTR name requested-user-name john' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cups_reject_jobs_form(): from pyipptool.forms import cups_reject_jobs_form request = cups_reject_jobs_form.render( @@ -307,6 +327,7 @@ def test_cups_reject_jobs_form(): assert 'ATTR text printer-state-message "You shall not pass"' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_get_job_attributes_form(): from pyipptool.forms import get_job_attributes_form request = get_job_attributes_form.render( @@ -326,6 +347,7 @@ def test_get_job_attributes_form(): assert 'ATTR keyword requested-attributes job-uri' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_get_jobs_form(): from pyipptool.forms import get_jobs_form request = get_jobs_form.render( @@ -347,6 +369,7 @@ def test_get_jobs_form(): assert 'ATTR keyword requested-attributes job-uri' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_get_printer_attributes_form(): from pyipptool.forms import get_printer_attributes_form request = get_printer_attributes_form.render( @@ -364,6 +387,7 @@ def test_get_printer_attributes_form(): ' printer-name,operations-supported' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_get_subscriptions_form(): from pyipptool.forms import get_subscriptions_form request = get_subscriptions_form.render( @@ -385,6 +409,7 @@ def test_get_subscriptions_form(): assert 'ATTR boolean my-subscriptions 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_get_notifications_form_for_one_notification(): from pyipptool.forms import get_notifications_form request = get_notifications_form.render( @@ -404,6 +429,7 @@ def test_get_notifications_form_for_one_notification(): assert 'ATTR boolean notify-wait 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_get_notifications_form_for_multiple_notifications(): from pyipptool.forms import get_notifications_form request = get_notifications_form.render( @@ -423,6 +449,7 @@ def test_get_notifications_form_for_multiple_notifications(): assert 'ATTR boolean notify-wait 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_pause_printer_form(): from pyipptool.forms import pause_printer_form request = pause_printer_form.render( @@ -436,6 +463,7 @@ def test_pause_printer_form(): assert 'ATTR name requesting-user-name yoda' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_resume_printer_form(): from pyipptool.forms import resume_printer_form request = resume_printer_form.render( @@ -449,6 +477,7 @@ def test_resume_printer_form(): assert 'ATTR name requesting-user-name yoda' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_hold_new_jobs_form(): from pyipptool.forms import hold_new_jobs_form request = hold_new_jobs_form.render( @@ -464,6 +493,7 @@ def test_hold_new_jobs_form(): assert 'ATTR text printer-message-from-operator "freeze jobs"' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_release_held_new_jobs_form(): from pyipptool.forms import release_held_new_jobs_form request = release_held_new_jobs_form.render( @@ -479,6 +509,7 @@ def test_release_held_new_jobs_form(): assert 'ATTR text printer-message-from-operator "melt jobs"' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_cancel_subscription_form(): from pyipptool.forms import cancel_subscription_form request = cancel_subscription_form.render( diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index 3741c87..442edae 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -1,4 +1,5 @@ import BaseHTTPServer +import os import SocketServer import socket import threading @@ -10,6 +11,10 @@ import pyipptool +NO_TORNADO = os.getenv('NO_TORNADO', '').lower() in ('1', 'yes', 'true', 't') + + +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_ipptool_create_job_subscription_pull_delivery_method(_call_ipptool): from pyipptool import create_job_subscription @@ -37,6 +42,7 @@ def test_ipptool_create_job_subscription_pull_delivery_method(_call_ipptool): assert 'notify-time-interval 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_ipptool_create_printer_subscription(_call_ipptool): from pyipptool import create_printer_subscription @@ -64,6 +70,7 @@ def test_ipptool_create_printer_subscription(_call_ipptool): assert 'notify-time-interval 1' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_add_modify_printer(_call_ipptool): from pyipptool import cups_add_modify_printer @@ -78,6 +85,7 @@ def test_cups_add_modify_printer(_call_ipptool): assert 'device-uri cups-pdf:/' in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_job_attributes_with_job_id(_call_ipptool): from pyipptool import get_job_attributes @@ -92,6 +100,7 @@ def test_get_job_attributes_with_job_id(_call_ipptool): assert 'job-uri' not in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_job_attributes_with_job_uri(_call_ipptool): from pyipptool import get_job_attributes @@ -104,6 +113,7 @@ def test_get_job_attributes_with_job_uri(_call_ipptool): assert 'printer-uri' not in request +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_timeout(): from pyipptool import wrapper from pyipptool.core import TimeoutError @@ -152,6 +162,7 @@ def do_POST(self): wrapper.config['timeout'] = old_timeout +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_authentication(): from pyipptool import IPPToolWrapper wrapper = IPPToolWrapper({'login': 'ezeep', @@ -162,6 +173,7 @@ def test_authentication(): 'http://ezeep:secret@localhost:631/printers/?arg=value') +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_release_job(_call_ipptool): from pyipptool import release_job @@ -170,6 +182,7 @@ def test_release_job(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cancel_job(_call_ipptool): from pyipptool import cancel_job @@ -178,6 +191,7 @@ def test_cancel_job(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_add_modify_class(_call_ipptool): from pyipptool import cups_add_modify_class @@ -187,6 +201,7 @@ def test_cups_add_modify_class(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_delete_printer(_call_ipptool): from pyipptool import cups_delete_printer @@ -195,6 +210,7 @@ def test_cups_delete_printer(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_delete_class(_call_ipptool): from pyipptool import cups_delete_class @@ -203,6 +219,7 @@ def test_cups_delete_class(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_classes(_call_ipptool): from pyipptool import cups_get_classes @@ -211,6 +228,7 @@ def test_cups_get_classes(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_printers(_call_ipptool): from pyipptool import cups_get_printers @@ -219,6 +237,7 @@ def test_cups_get_printers(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_devices(_call_ipptool): from pyipptool import cups_get_devices @@ -227,6 +246,7 @@ def test_cups_get_devices(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_ppds(_call_ipptool): from pyipptool import cups_get_ppds @@ -235,6 +255,7 @@ def test_cups_get_ppds(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_move_job(_call_ipptool): from pyipptool import cups_move_job @@ -243,6 +264,7 @@ def test_cups_move_job(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_reject_jobs(_call_ipptool): from pyipptool import cups_reject_jobs @@ -251,6 +273,7 @@ def test_cups_reject_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_jobs(_call_ipptool): from pyipptool import get_jobs @@ -259,6 +282,7 @@ def test_get_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_printer_attributes(_call_ipptool): from pyipptool import get_printer_attributes @@ -267,6 +291,7 @@ def test_get_printer_attributes(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_subscriptions(_call_ipptool): from pyipptool import get_subscriptions @@ -275,6 +300,7 @@ def test_get_subscriptions(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_notifications(_call_ipptool): from pyipptool import get_notifications @@ -285,6 +311,7 @@ def test_get_notifications(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_pause_printer(_call_ipptool): from pyipptool import pause_printer @@ -293,6 +320,7 @@ def test_pause_printer(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_hold_new_jobs(_call_ipptool): from pyipptool import hold_new_jobs @@ -301,6 +329,7 @@ def test_hold_new_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_release_held_new_jobs(_call_ipptool): from pyipptool import release_held_new_jobs @@ -309,6 +338,7 @@ def test_release_held_new_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_resume_printer(_call_ipptool): from pyipptool import resume_printer @@ -317,6 +347,7 @@ def test_resume_printer(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cancel_subscription(_call_ipptool): from pyipptool import cancel_subscription diff --git a/tests/test_widget.py b/tests/test_widget.py index 0e2ad97..d611e69 100644 --- a/tests/test_widget.py +++ b/tests/test_widget.py @@ -1,8 +1,13 @@ +import os + import colander from deform.tests.test_widget import DummyField, DummyRenderer import pytest +NO_TORNADO = os.getenv('NO_TORNADO', '').lower() in ('1', 'yes', 'true', 't') + + class DummyType(object): pass @@ -12,6 +17,7 @@ def __init__(self, **kw): self.__dict__.update(kw) +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_ipp_attribute_widget_null(): from pyipptool.widgets import IPPAttributeWidget widget = IPPAttributeWidget() @@ -21,6 +27,7 @@ def test_ipp_attribute_widget_null(): assert response == '' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_ipp_attribute_widget_with_value(): from pyipptool.widgets import IPPAttributeWidget widget = IPPAttributeWidget() @@ -32,6 +39,7 @@ def test_ipp_attribute_widget_with_value(): assert response == 'ATTR dummyType name hello' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_ipp_attribute_widget_with_None(): from pyipptool.widgets import IPPAttributeWidget widget = IPPAttributeWidget() @@ -44,6 +52,7 @@ def test_ipp_attribute_widget_with_None(): assert exc_info.value.message == "None value provided for 'name'" +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_ipp_display_widget(): from pyipptool.widgets import IPPDisplayWidget widget = IPPDisplayWidget() @@ -53,6 +62,7 @@ def test_ipp_display_widget(): assert response == 'DISPLAY name' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_ipp_name_widget(): from pyipptool.widgets import IPPNameWidget widget = IPPNameWidget() @@ -68,6 +78,7 @@ def test_ipp_name_widget(): assert response == 'NAME "FOO"' +@pytest.mark.skipif(not NO_TORNADO, reason='w/o tornado') def test_ipp_group_widget(): from pyipptool.widgets import IPPGroupWidget widget = IPPGroupWidget() diff --git a/tests/test_with_cups.py b/tests/test_with_cups.py index 028a208..02c8359 100644 --- a/tests/test_with_cups.py +++ b/tests/test_with_cups.py @@ -1,14 +1,16 @@ import os -import pyipptool import pytest TRAVIS = os.getenv('TRAVIS') TRAVIS_USER = os.getenv('USER') TRAVIS_BUILD_DIR = os.getenv('TRAVIS_BUILD_DIR') +NO_TORNADO = os.getenv('NO_TORNADO', '').lower() in ('1', 'yes', 'true', 't') -@pytest.mark.skipif(TRAVIS != 'true', reason='requires travis') + +@pytest.mark.skipif(TRAVIS != 'true' or not(NO_TORNADO), + reason='requires travis') class TestWithCups(object): ipptool_path = '%s/ipptool-20130731/ipptool' % TRAVIS_BUILD_DIR @@ -19,6 +21,7 @@ class TestWithCups(object): 'timeout': 5} def test_cups_get_printers(self): + import pyipptool ipptool = pyipptool.core.IPPToolWrapper(self.config) response = ipptool.cups_get_printers('http://localhost:631/') assert response['Name'] == 'CUPS Get Printers' @@ -32,6 +35,7 @@ def test_cups_get_printers(self): assert response['Successful'] def test_cups_get_ppds(self): + import pyipptool ipptool = pyipptool.core.IPPToolWrapper(self.config) response = ipptool.cups_get_ppds('http://localhost:631/', ppd_make_and_model='generic pdf')