More accurate documentation on single

reactivex/operators/__init__.py

@@ -2622,7 +2622,7 @@ def scan(
     Applies an accumulator function over an observable sequence and
     returns each intermediate result. The optional seed value is used
     as the initial accumulator value. For aggregation behavior with no
-    intermediate results, see `aggregate()` or `Observable()`.
+    intermediate results, see `reduce()` or `Observable()`.
 
     .. marble::
         :alt: scan
@@ -2716,12 +2716,29 @@ def single(
     the condition in the optional predicate, and reports an exception
     if there is not exactly one element in the observable sequence.
 
+    If the predicates does not match any item, the resulting sequence
+    errors once the source completes.
+
+    If the predicate matches more than one item, the resulting sequence
+    errors immediately, without waiting for the source to complete.
+
+    If the source never completes, the resulting sequence does not
+    emit anything, nor completes.
+
     .. marble::
         :alt: single
 
         ----1--2--3--4-----|
-        [     single(3)    ]
-        ----------3--------|
+        [   single(x==3)   ]
+        -----------------3-|
+
+        ----1--3--3--4-----|
+        [   single(x==3)   ]
+        ----------x
+
+        ----1--1--1--1-----|
+        [   single(x==3)   ]
+        -------------------x
 
     Example:
         >>> res = single()

tests/test_observable/test_single.py

@@ -22,6 +22,18 @@ def _raise(ex):
 
 
 class TestSingle(unittest.TestCase):
+    def test_single_source_never_completes(self):
+        scheduler = TestScheduler()
+        xs = scheduler.create_hot_observable(on_next(150, 1), on_next(250, 2))
+
+        def create():
+            return xs.pipe(ops.single())
+
+        res = scheduler.start(create=create)
+
+        assert [] == res.messages
+        assert xs.subscriptions == [subscribe(200, 1000)]
+
     def test_single_async_empty(self):
         scheduler = TestScheduler()
         xs = scheduler.create_hot_observable(on_next(150, 1), on_completed(250))
@@ -52,9 +64,14 @@ def create():
         assert xs.subscriptions == [subscribe(200, 250)]
 
     def test_single_async_many(self):
+        """Should error as soon as a second "valid" element is encountered."""
         scheduler = TestScheduler()
         xs = scheduler.create_hot_observable(
-            on_next(150, 1), on_next(210, 2), on_next(220, 3), on_completed(250)
+            on_next(150, 1),
+            on_next(210, 2),
+            on_next(220, 3),
+            on_next(230, 3),
+            on_completed(250),
         )
 
         def create():