From 3579e95398625ab1b3fbfee8cb8eef52c5e462f2 Mon Sep 17 00:00:00 2001
From: Kirill Rakhman <kirill.rakhman@gmail.com>
Date: Sun, 17 Jul 2016 13:49:28 +0200
Subject: [PATCH] modify rx tests to work with real asynchronous observables

---
 .../src/test/kotlin/AsyncRxTest.kt            | 32 ++++++++++---------
 1 file changed, 17 insertions(+), 15 deletions(-)

diff --git a/kotlinx-coroutines-rx/src/test/kotlin/AsyncRxTest.kt b/kotlinx-coroutines-rx/src/test/kotlin/AsyncRxTest.kt
index 6b4fc5296b..09ffa8d564 100644
--- a/kotlinx-coroutines-rx/src/test/kotlin/AsyncRxTest.kt
+++ b/kotlinx-coroutines-rx/src/test/kotlin/AsyncRxTest.kt
@@ -2,6 +2,7 @@ package kotlinx.coroutines
 
 import org.junit.Test
 import rx.Observable
+import java.util.concurrent.TimeUnit
 import kotlin.test.assertEquals
 import kotlin.test.fail
 
@@ -17,6 +18,18 @@ class AsyncRxTest {
         }
     }
 
+    @Test
+    fun testSingleWithDelay() {
+        val observable = asyncRx<String> {
+            Observable.timer(50, TimeUnit.MILLISECONDS).map { "O" }.awaitSingle() + "K"
+        }
+
+        checkObservableWithSingleValue(observable) {
+            assertEquals("OK", it)
+        }
+    }
+
+
     @Test
     fun testSingleException() {
         val observable = asyncRx<String> {
@@ -117,26 +130,15 @@ class AsyncRxTest {
             observable: Observable<*>,
             checker: (Throwable) -> Unit
     ) {
-        var onErrorCalled = false
-        observable.subscribe({ fail("Next item on erroneous observable") }) {
-            checker(it)
-            onErrorCalled = true
-        }
-
-        assert(onErrorCalled)
+        val singleNotification = observable.materialize().toBlocking().single()
+        checker(singleNotification.throwable)
     }
 
     private fun <T> checkObservableWithSingleValue(
             observable: Observable<T>,
             checker: (T) -> Unit
     ) {
-        var subscribeCalled = false
-
-        observable.single().subscribe {
-            checker(it)
-            subscribeCalled = true
-        }
-
-        assert(subscribeCalled)
+        val singleValue = observable.toBlocking().single()
+        checker(singleValue)
     }
 }