update how single-results are determined in AsyncResults

single msg_id (not a list) implies single result

fixes issue where `targets='all'` could imply single-result when only one engine.

IPython/parallel/client/asyncresult.py

@@ -73,6 +73,9 b' class AsyncResult(object):'
         if isinstance(msg_ids, basestring):
             # always a list
             msg_ids = [msg_ids]
+            self._single_result = True
+        else:
+            self._single_result = False
         if tracker is None:
             # default to always done
             tracker = finished_tracker
@@ -81,14 +84,11 b' class AsyncResult(object):'
         self._fname=fname
         self._targets = targets
         self._tracker = tracker
+        
         self._ready = False
         self._outputs_ready = False
         self._success = None
         self._metadata = [ self._client.metadata.get(id) for id in self.msg_ids ]
-        if len(msg_ids) == 1:
-            self._single_result = not isinstance(targets, (list, tuple))
-        else:
-            self._single_result = False
 
     def __repr__(self):
         if self._ready:

IPython/parallel/client/view.py

@@ -549,8 +549,8 b' class DirectView(View):'
         block = self.block if block is None else block
         track = self.track if track is None else track
         targets = self.targets if targets is None else targets
-
-        _idents = self.client._build_targets(targets)[0]
+        
+        _idents, _targets = self.client._build_targets(targets)
         msg_ids = []
         trackers = []
         for ident in _idents:
@@ -559,8 +559,10 b' class DirectView(View):'
             if track:
                 trackers.append(msg['tracker'])
             msg_ids.append(msg['header']['msg_id'])
+        if isinstance(targets, int):
+            msg_ids = msg_ids[0]
         tracker = None if track is False else zmq.MessageTracker(*trackers)
-        ar = AsyncResult(self.client, msg_ids, fname=getname(f), targets=targets, tracker=tracker)
+        ar = AsyncResult(self.client, msg_ids, fname=getname(f), targets=_targets, tracker=tracker)
         if block:
             try:
                 return ar.get()
@@ -631,13 +633,15 b' class DirectView(View):'
         block = self.block if block is None else block
         targets = self.targets if targets is None else targets
 
-        _idents = self.client._build_targets(targets)[0]
+        _idents, _targets = self.client._build_targets(targets)
         msg_ids = []
         trackers = []
         for ident in _idents:
             msg = self.client.send_execute_request(self._socket, code, silent=silent, ident=ident)
             msg_ids.append(msg['header']['msg_id'])
-        ar = AsyncResult(self.client, msg_ids, fname='execute', targets=targets)
+        if isinstance(targets, int):
+            msg_ids = msg_ids[0]
+        ar = AsyncResult(self.client, msg_ids, fname='execute', targets=_targets)
         if block:
             try:
                 ar.get()

IPython/parallel/tests/test_asyncresult.py

@@ -308,4 +308,18 b' class AsyncResultTest(ClusterTestCase):'
         ar.get(5)
         nt.assert_in(4, found)
         self.assertTrue(len(found) > 1, "should have seen data multiple times, but got: %s" % found)
+    
+    def test_not_single_result(self):
+        save_build = self.client._build_targets
+        def single_engine(*a, **kw):
+            idents, targets = save_build(*a, **kw)
+            return idents[:1], targets[:1]
+        ids = single_engine('all')[1]
+        self.client._build_targets = single_engine
+        for targets in ('all', None, ids):
+            dv = self.client.direct_view(targets=targets)
+            ar = dv.apply_async(lambda : 5)
+            self.assertEqual(ar.get(10), [5])
+        self.client._build_targets = save_build
+