Add Async (Tornado) support

.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

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'],

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},

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