Move asserts to StepVerifier (#90)

src/test/java/com/google/cloud/spanner/r2dbc/SpannerConnectionFactoryTest.java

@@ -26,6 +26,7 @@
 import org.junit.Test;
 import org.mockito.Mockito;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * Test for {@link SpannerConnectionFactory}.
@@ -63,9 +64,10 @@ public void createReturnsNewSpannerConnection() {
         .thenReturn(Mono.just(session));
 
     SpannerConnectionFactory factory = new SpannerConnectionFactory(mockClient, this.config);
-    SpannerConnection connection = Mono.from(factory.create()).block();
-
-    assertThat(connection.getSession().getName()).isEqualTo("jam session");
+    Mono<SpannerConnection> connection = Mono.from(factory.create());
 
+    StepVerifier.create(connection.map(con -> con.getSession().getName()))
+            .expectNext("jam session")
+            .verifyComplete();
   }
 }

src/test/java/com/google/cloud/spanner/r2dbc/SpannerConnectionTest.java

@@ -112,8 +112,9 @@ public void beginAndCommitTransactions() {
     when(this.mockClient.commitTransaction(TEST_SESSION,  Transaction.getDefaultInstance()))
         .thenReturn(commitTransactionProbe.mono());
 
-    Mono.from(connection.beginTransaction()).block();
-    Mono.from(connection.commitTransaction()).block();
+    Mono.from(connection.beginTransaction())
+            .then(Mono.from(connection.commitTransaction()))
+            .subscribe();
     verify(this.mockClient, times(1))
         .beginTransaction(TEST_SESSION);
     verify(this.mockClient, times(1))

src/test/java/com/google/cloud/spanner/r2dbc/SpannerResultTest.java

@@ -16,7 +16,6 @@
 
 package com.google.cloud.spanner.r2dbc;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.google.protobuf.Value;
@@ -26,11 +25,11 @@
 import com.google.spanner.v1.Type;
 import com.google.spanner.v1.TypeCode;
 import java.util.Collections;
-import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * Test for {@link SpannerResult}.
@@ -68,9 +67,10 @@ public void setup() {
 
   @Test
   public void getRowsUpdatedTest() {
-    assertThat(
-        ((Mono) new SpannerResult(this.resultSet,Mono.just(2)).getRowsUpdated()).block())
-        .isEqualTo(2);
+    StepVerifier.create(
+        ((Mono) new SpannerResult(this.resultSet,Mono.just(2)).getRowsUpdated()))
+        .expectNext(2)
+        .verifyComplete();
   }
 
   @Test
@@ -93,17 +93,15 @@ public void mapTest() {
         .getFields(0)
         .getName();
 
-    List<String> result =
+    Flux<String> result =
         new SpannerResult(this.resultSet, Mono.just(0))
             .map((row, metadata) ->
                 row.get(0, String.class)
                     + "-"
-                    + metadata.getColumnMetadata(0).getName())
-            .collectList()
-            .block();
-
-    assertThat(result)
-        .containsExactly("key1-" + columnName, "key2-" + columnName);
+                    + metadata.getColumnMetadata(0).getName());
 
+    StepVerifier.create(result)
+        .expectNext("key1-" + columnName, "key2-" + columnName)
+        .verifyComplete();
   }
 }

src/test/java/com/google/cloud/spanner/r2dbc/SpannerStatementTest.java

@@ -34,10 +34,11 @@
 import com.google.spanner.v1.Type;
 import com.google.spanner.v1.TypeCode;
 import io.r2dbc.spi.Result;
-import java.util.List;
+import java.util.Collections;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * Test for {@link SpannerStatement}.
@@ -77,13 +78,16 @@ public void executeDummyImplementation() {
         .thenReturn(Flux.just(partialResultSet));
 
     SpannerStatement statement
-        = new SpannerStatement(mockClient, TEST_SESSION, null,sql);
+        = new SpannerStatement(mockClient, TEST_SESSION, null, sql);
 
-    Mono<SpannerResult> result = (Mono<SpannerResult>)statement.execute();
+    Mono<SpannerResult> result = (Mono<SpannerResult>) statement.execute();
 
     assertThat(result).isNotNull();
 
-    result.block().map((r, m) -> (String)r.get(0)).blockFirst().equals("Odyssey");
+    StepVerifier.create(result.flatMapMany(
+        spannerResult -> spannerResult.map((row, rowMetadata) -> (String) row.get(0))))
+            .expectNext("Odyssey")
+            .verifyComplete();
 
     verify(mockClient).executeStreamingSql(TEST_SESSION, null, sql);
   }
@@ -103,10 +107,13 @@ public void readOneResultSetQueryTest() {
     Mono<Result> resultMono = Mono
         .from(new SpannerStatement(this.mockClient, null, null, null).execute());
 
-    assertThat(resultMono.flatMap(r -> Mono.from(r.getRowsUpdated())).block()).isZero();
-    assertThat(resultMono.flatMapMany(r -> r
-        .map((row, meta) -> row.get(0, Boolean.class).toString() + "-" + row.get(1, String.class)))
-        .collectList().block()).containsExactly("false-abc");
+    StepVerifier.create(resultMono.flatMap(r -> Mono.from(r.getRowsUpdated())))
+        .expectNext(0)
+        .verifyComplete();
+    StepVerifier.create(resultMono.flatMapMany(r -> r
+        .map((row, meta) -> row.get(0, Boolean.class).toString() + "-" + row.get(1, String.class))))
+        .expectNext("false-abc")
+        .verifyComplete();
   }
 
   @Test
@@ -123,8 +130,10 @@ public void readMultiResultSetQueryTest() {
 
     when(this.mockClient.executeStreamingSql(any(), any(), any())).thenReturn(inputs);
 
-    assertThat(Mono.from(new SpannerStatement(this.mockClient, null, null, null).execute())
-        .flatMap(r -> Mono.from(r.getRowsUpdated())).block()).isZero();
+    StepVerifier.create(Mono.from(new SpannerStatement(this.mockClient, null, null, null).execute())
+        .flatMap(r -> Mono.from(r.getRowsUpdated())))
+        .expectNext(0)
+        .verifyComplete();
   }
 
   @Test
@@ -137,8 +146,10 @@ public void readDmlQueryTest() {
 
     when(this.mockClient.executeStreamingSql(any(), any(), any())).thenReturn(inputs);
 
-    assertThat(Mono.from(new SpannerStatement(this.mockClient, null, null, null).execute())
-        .flatMap(r -> Mono.from(r.getRowsUpdated())).block()).isEqualTo(555);
+    StepVerifier.create(Mono.from(new SpannerStatement(this.mockClient, null, null, null).execute())
+        .flatMap(r -> Mono.from(r.getRowsUpdated())))
+        .expectNext(555)
+        .verifyComplete();
   }
 
   @Test
@@ -153,15 +164,19 @@ public void noopMapOnUpdateQueriesWhenNoRowsAffected() {
         .thenReturn(Flux.just(partialResultSet));
 
     SpannerStatement statement
-        = new SpannerStatement(mockClient, TEST_SESSION, null,sql);
+        = new SpannerStatement(mockClient, TEST_SESSION, null, sql);
 
-    SpannerResult result = ((Mono<SpannerResult>) statement.execute()).block();
+    Mono<SpannerResult> result = (Mono<SpannerResult>) statement.execute();
 
-    List<String> rowStrings = result.map((r, m) -> (String)r.get(0)).collectList().block();
-    assertThat(rowStrings).isEmpty();
+    StepVerifier.create(result.flatMap(
+        spannerResult -> spannerResult.map((row, rowMetadata) -> (String) row.get(0))
+            .collectList()))
+        .expectNext(Collections.emptyList())
+        .verifyComplete();
 
-    int rowsUpdated = Mono.from(result.getRowsUpdated()).block();
-    assertThat(rowsUpdated).isEqualTo(0);
+    StepVerifier.create(result.flatMap(results -> Mono.from(results.getRowsUpdated())))
+        .expectNext(0)
+        .verifyComplete();
 
     verify(mockClient, times(1)).executeStreamingSql(TEST_SESSION, null, sql);
   }

src/test/java/com/google/cloud/spanner/r2dbc/result/PartialResultRowExtractorTest.java

@@ -16,8 +16,6 @@
 
 package com.google.cloud.spanner.r2dbc.result;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
 import com.google.cloud.spanner.r2dbc.SpannerColumnMetadata;
 import com.google.cloud.spanner.r2dbc.SpannerRow;
 import com.google.protobuf.ListValue;
@@ -31,9 +29,9 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 /**
  * Tests the partial result flux converter.
@@ -190,34 +188,36 @@ public void handleEmptyPartialResultSet() {
 
     Flux<PartialResultSet> inputs = Flux.just(emptyResultSet);
 
-    List<SpannerRow> results =
-        inputs.flatMapIterable(new PartialResultRowExtractor())
-            .collectList()
-            .block();
-    assertThat(results).isEmpty();
+    Flux<SpannerRow> results =
+        inputs.flatMapIterable(new PartialResultRowExtractor());
+
+    StepVerifier.create(results)
+        .verifyComplete();
   }
 
   private void verifyRows(Flux<PartialResultSet> inputs) {
-    List<SpannerRow> results =
-        inputs.flatMapIterable(new PartialResultRowExtractor())
-            .collectList()
-            .block();
+    Flux<SpannerRow> results =
+        inputs.flatMapIterable(new PartialResultRowExtractor());
 
-    List<ColumnMetadata> columnMetadata = this.resultSetMetadata.getRowType().getFieldsList()
+    List<String> expectedColNames = this.resultSetMetadata.getRowType().getFieldsList()
         .stream()
         .map(SpannerColumnMetadata::new)
+        .map(ColumnMetadata::getName)
         .collect(Collectors.toList());
 
-    List<String> expectedColNames = columnMetadata.stream().map(ColumnMetadata::getName)
-        .collect(Collectors.toList());
+    String[] expectedColumnNames = new String[expectedColNames.size()];
+    expectedColNames.toArray(expectedColumnNames);
 
-    results.forEach(row -> assertThat(
-        StreamSupport.stream(row.getRowMetadata().getColumnMetadatas().spliterator(), false)
-            .map(
-                ColumnMetadata::getName).collect(Collectors.toList()))
-        .isEqualTo(expectedColNames));
+    StepVerifier.create(results
+        .flatMap(row -> Flux.fromIterable(row.getRowMetadata().getColumnMetadatas())
+            .map(ColumnMetadata::getName)))
+        .expectNext(expectedColumnNames)
+        .expectNext(expectedColumnNames)
+        .verifyComplete();
 
-    assertThat(results.get(0).getValues()).containsExactly(this.a1, this.a2, this.a3);
-    assertThat(results.get(1).getValues()).containsExactly(this.b1, this.b2, this.b3);
+    StepVerifier.create(results.map(SpannerRow::getValues))
+        .expectNext(Arrays.asList(this.a1, this.a2, this.a3))
+        .expectNext(Arrays.asList(this.b1, this.b2, this.b3))
+        .verifyComplete();
   }
 }

src/test/java/com/google/cloud/spanner/r2dbc/util/ObservableReactiveUtilTest.java

@@ -17,15 +17,16 @@
 package com.google.cloud.spanner.r2dbc.util;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.r2dbc.spi.R2dbcNonTransientException;
+import io.r2dbc.spi.R2dbcNonTransientResourceException;
 import io.r2dbc.spi.R2dbcTransientResourceException;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * Test for {@link ObservableReactiveUtil}.
@@ -38,26 +39,32 @@ public void unaryCallReturnsSingleValue() {
       observer.onNext(42);
       observer.onCompleted();
     });
-    assertThat(mono.block()).isEqualTo(42);
+    StepVerifier.create(mono)
+        .expectNext(42)
+        .verifyComplete();
   }
 
   @Test
   public void unaryCallForwardsError() {
     Mono<Integer> mono = ObservableReactiveUtil.unaryCall(observer -> {
       observer.onError(new IllegalArgumentException("oh no"));
     });
-    assertThatThrownBy(() -> mono.block())
-        .hasCauseInstanceOf(IllegalArgumentException.class)
-        .isInstanceOf(R2dbcNonTransientException.class)
-        .hasMessage("oh no");
+
+    StepVerifier.create(mono)
+        .expectErrorMatches(throwable -> throwable instanceof R2dbcNonTransientResourceException
+            && throwable.getMessage().equals("oh no"))
+        .verify();
   }
 
   @Test
   public void unaryCallThrowsExceptionIfCompletedWithNoValue() {
     Mono<Integer> mono = ObservableReactiveUtil.unaryCall(observer -> observer.onCompleted());
-    assertThatThrownBy(() -> mono.block())
-        .isInstanceOf(RuntimeException.class)
-        .hasMessage("Unary gRPC call completed without yielding a value or an error");
+
+    StepVerifier.create(mono)
+        .expectErrorMatches(throwable -> throwable instanceof RuntimeException
+            && throwable.getMessage().equals(
+                "Unary gRPC call completed without yielding a value or an error"))
+        .verify();
   }
 
   @Test
@@ -69,9 +76,11 @@ public void propagateTransientErrorUnaryCall() {
     Mono<Void> result =
         ObservableReactiveUtil.unaryCall(observer -> observer.onError(retryableException));
 
-    assertThatThrownBy(() -> result.block())
-        .hasCauseInstanceOf(StatusRuntimeException.class)
-        .isInstanceOf(R2dbcTransientResourceException.class);
+    StepVerifier.create(result)
+            .expectErrorSatisfies(throwable ->
+                    assertThat(throwable).hasCauseInstanceOf(StatusRuntimeException.class)
+                            .isInstanceOf(R2dbcTransientResourceException.class))
+            .verify();
   }
 
   @Test
@@ -80,9 +89,11 @@ public void propagateNonRetryableError() {
         ObservableReactiveUtil.unaryCall(
             observer -> observer.onError(new IllegalArgumentException()));
 
-    assertThatThrownBy(() -> result.block())
-        .hasCauseInstanceOf(IllegalArgumentException.class)
-        .isInstanceOf(R2dbcNonTransientException.class);
+    StepVerifier.create(result)
+            .expectErrorSatisfies(throwable ->
+                    assertThat(throwable).hasCauseInstanceOf(IllegalArgumentException.class)
+                            .isInstanceOf(R2dbcNonTransientException.class))
+            .verify();
   }
 
   @Test
@@ -94,8 +105,11 @@ public void propagateTransientErrorStreamingCall() {
     Flux<Void> result =
         ObservableReactiveUtil.streamingCall(observer -> observer.onError(retryableException));
 
-    assertThatThrownBy(() -> result.blockFirst())
-        .hasCauseInstanceOf(StatusRuntimeException.class)
-        .isInstanceOf(R2dbcTransientResourceException.class);
+    StepVerifier.create(result)
+            .expectErrorSatisfies(throwable ->
+                    assertThat(throwable)
+                        .hasCauseInstanceOf(StatusRuntimeException.class)
+                        .isInstanceOf(R2dbcTransientResourceException.class))
+            .verify();
   }
 }