Merge remote-tracking branch 'origin/dmw'

* origin/dmw:
  issue #615: fix up FileService tests for new logic
  issue #615: another Py3x fix.
  issue #615: Py3x fix.
  issue #615: update Changelog.
  issue #615: use FileService for target->controll file transfers

Author: dw

ansible_mitogen/connection.py

@@ -950,11 +950,12 @@ def fetch_file(self, in_path, out_path):
         :param str out_path:
             Local filesystem path to write.
         """
-        output = self.get_chain().call(
-            ansible_mitogen.target.read_path,
-            mitogen.utils.cast(in_path),
+        self._connect()
+        ansible_mitogen.target.transfer_file(
+            context=self.context,
+            in_path=in_path,
+            out_path=out_path
         )
-        ansible_mitogen.target.write_path(out_path, output)
 
     def put_data(self, out_path, data, mode=None, utimes=None):
         """

docs/changelog.rst

@@ -116,6 +116,12 @@ Mitogen for Ansible
   module, and for any other action plug-ins that establish new connections of
   their own.
 
+* `#615 <https://github.com/dw/mitogen/issues/615>`_: streaming file transfer
+  is implemented for the ``fetch`` and any other action that transfers files
+  from the target to the controller. Previously the file would be sent as a
+  single message, requiring the file to fit in RAM and be smaller than internal
+  limits on the size of a single message.
+
 * `7ae926b3 <https://github.com/dw/mitogen/commit/7ae926b3>`_: the
   ``lineinfile`` module began leaking writable temporary file descriptors since
   Ansible 2.7.0. When ``lineinfile`` was used to create or modify a script, and
@@ -219,6 +225,7 @@ bug reports, testing, features and fixes in this release contributed by
 `Szabó Dániel Ernő <https://github.com/r3ap3rpy>`_,
 `Ulrich Schreiner <https://github.com/ulrichSchreiner>`_,
 `Yuki Nishida <https://github.com/yuki-nishida-exa>`_,
+`@alexhexabeam <https://github.com/alexhexabeam>`_,
 `@DavidVentura <https://github.com/DavidVentura>`_,
 `@ghp-rr <https://github.com/ghp-rr>`_,
 `@rizzly <https://github.com/rizzly>`_, and

mitogen/core.py

@@ -395,6 +395,12 @@ def bytes_partition(s, sep):
         return _partition(s, sep, s.find) or (s, '', '')
 
 
+def _has_parent_authority(context_id):
+    return (
+        (context_id == mitogen.context_id) or
+        (context_id in mitogen.parent_ids)
+    )
+
 def has_parent_authority(msg, _stream=None):
     """
     Policy function for use with :class:`Receiver` and
@@ -403,8 +409,7 @@ def has_parent_authority(msg, _stream=None):
     <Stream.auth_id>` has been set to that of a parent context or the current
     context.
     """
-    return (msg.auth_id == mitogen.context_id or
-            msg.auth_id in mitogen.parent_ids)
+    return _has_parent_authority(msg.auth_id)
 
 
 def _signals(obj, signal):

mitogen/service.py

@@ -1023,7 +1023,11 @@ def fetch(self, path, sender, msg):
         :raises Error:
             Unregistered path, or Sender did not match requestee context.
         """
-        if path not in self._paths and not self._prefix_is_authorized(path):
+        if (
+            (path not in self._paths) and
+            (not self._prefix_is_authorized(path)) and
+            (not mitogen.core._has_parent_authority(msg.auth_id))
+        ):
             msg.reply(mitogen.core.CallError(
                 Error(self.unregistered_msg % (path,))
             ))

tests/ansible/tests/connection_test.py

@@ -99,6 +99,21 @@ def test_is_junk(self):
         self.assertEquals(None, self.func({1:2}))
 
 
+class FetchFileTest(ConnectionMixin, testlib.TestCase):
+    def test_success(self):
+        with tempfile.NamedTemporaryFile(prefix='mitotest') as ifp:
+            with tempfile.NamedTemporaryFile(prefix='mitotest') as ofp:
+                ifp.write(b'x' * (1048576 * 4))
+                ifp.flush()
+                ifp.seek(0)
+
+                self.conn.fetch_file(ifp.name, ofp.name)
+                # transfer_file() uses os.rename rather than direct data
+                # overwrite, so we must reopen.
+                with open(ofp.name, 'rb') as fp:
+                    self.assertEquals(ifp.read(), fp.read())
+
+
 class PutDataTest(ConnectionMixin, testlib.TestCase):
     def test_out_path(self):
         path = tempfile.mktemp(prefix='mitotest')

tests/file_service_test.py

@@ -22,15 +22,26 @@ def replyable_msg(self, **kwargs):
         return recv, msg
 
     def test_unauthorized(self):
+        l1 = self.router.local()
+
         service = self.klass(self.router)
-        recv, msg = self.replyable_msg()
-        service.fetch(
-            path='/etc/shadow',
-            sender=None,
-            msg=msg,
+        pool = mitogen.service.Pool(
+            router=self.router,
+            services=[service],
+            size=1,
         )
-        e = self.assertRaises(mitogen.core.CallError,
-            lambda: recv.get().unpickle())
+        try:
+            e = self.assertRaises(mitogen.core.CallError,
+                lambda: l1.call(
+                    mitogen.service.FileService.get,
+                    context=self.router.myself(),
+                    path='/etc/shadow',
+                    out_fp=None,
+                )
+            )
+        finally:
+            pool.stop()
+
         expect = service.unregistered_msg % ('/etc/shadow',)
         self.assertTrue(expect in e.args[0])
 
@@ -85,30 +96,57 @@ def test_prefix_authorized(self):
         self._validate_response(recv.get().unpickle())
 
     def test_prefix_authorized_abspath_bad(self):
-        recv = mitogen.core.Receiver(self.router)
+        l1 = self.router.local()
+
         service = self.klass(self.router)
         service.register_prefix('/etc')
-        recv, msg = self.replyable_msg()
-        service.fetch(
-            path='/etc/foo/bar/../../../passwd',
-            sender=recv.to_sender(),
-            msg=msg,
+
+        pool = mitogen.service.Pool(
+            router=self.router,
+            services=[service],
+            size=1,
         )
-        self.assertEquals(None, recv.get().unpickle())
+        path = '/etc/foo/bar/../../../passwd'
+        try:
+            e = self.assertRaises(mitogen.core.CallError,
+                lambda: l1.call(
+                    mitogen.service.FileService.get,
+                    context=self.router.myself(),
+                    path=path,
+                    out_fp=None,
+                )
+            )
+        finally:
+            pool.stop()
+
+        expect = service.unregistered_msg % (path,)
+        self.assertTrue(expect in e.args[0])
+
+    def test_prefix_authorized_abspath_good(self):
+        l1 = self.router.local()
 
-    def test_prefix_authorized_abspath_bad(self):
-        recv = mitogen.core.Receiver(self.router)
         service = self.klass(self.router)
         service.register_prefix('/etc')
-        recv, msg = self.replyable_msg()
-        service.fetch(
-            path='/etc/../shadow',
-            sender=recv.to_sender(),
-            msg=msg,
+        path = '/etc/../shadow'
+
+        pool = mitogen.service.Pool(
+            router=self.router,
+            services=[service],
+            size=1,
         )
-        e = self.assertRaises(mitogen.core.CallError,
-            lambda: recv.get().unpickle())
-        expect = service.unregistered_msg % ('/etc/../shadow',)
+        try:
+            e = self.assertRaises(mitogen.core.CallError,
+                lambda: l1.call(
+                    mitogen.service.FileService.get,
+                    context=self.router.myself(),
+                    path=path,
+                    out_fp=None
+                )
+            )
+        finally:
+            pool.stop()
+
+        expect = service.unregistered_msg % (path,)
         self.assertTrue(expect in e.args[0])