### Question:
Database { public <T> ReactiveExecutionBuilder<T> execute(Function<EntityManager, T> function) { return new ReactiveExecutionBuilder<>(function, this); } Database(Map<String, String> settings, List<String> resourcePackages);  Database(Map<String, String> settings, List<String> resourcePackages,
      PoolConfiguration configuration);  Database(SessionFactory sessionFactory, int maxPoolSize); ReactiveExecutionBuilder<T> execute(Function<EntityManager, T> function); StreamerBuilder<T> stream(String query, Class<T> type);  }### Answer:
@Test public void testMonoExecution() throws SQLException { TransactionDefinition definition = TransactionDefinition.builder() .timeout(10) .isolation(IsolationLevel.SERIALIZABLE) .isReadonly(true) .build(); setupTransactionMocks(definition); Database database = new Database(sessionFactory, 1); StepVerifier.create(database.execute(entityManager -> 1).mono()) .expectNext(1) .verifyComplete(); StepVerifier.create(database.execute(entityManager -> 1) .transaction(definition) .mono()) .expectNext(1) .verifyComplete(); }
@SuppressWarnings("unchecked") @Test public void testFluxExecution() throws SQLException { TransactionDefinition definition = TransactionDefinition.builder() .timeout(10) .isolation(IsolationLevel.SERIALIZABLE) .isReadonly(true) .build(); setupTransactionMocks(definition); List<String> list = new ArrayList<>(3); list.add("a"); list.add("b"); list.add("c"); Database database = new Database(sessionFactory, 1); StepVerifier.create(database.execute(entityManager -> list).flux()) .expectNextSequence(list) .verifyComplete(); StepVerifier.create(database.execute(entityManager -> 1) .transaction(definition) .flux()) .expectNext(1) .verifyComplete(); }
@Test public void testExceptionallyExecution() throws SQLException { Database database = new Database(sessionFactory, 1); StepVerifier.create(database.execute(entityManager -> { throw new RuntimeException(); }).mono()) .verifyError(); }