SurveyDataRepository implements SurveyRepository { @VisibleForTesting Observable<UploadResult> syncTransmission(final Transmission transmission, final String deviceId) { final DatabaseDataSource dataBaseDataSource = dataSourceFactory.getDataBaseDataSource(); final long transmissionId = transmission.getId(); final long surveyInstanceId = transmission.getRespondentId(); final String formId = transmission.getFormId(); return s3RestApi.uploadFile(transmission) .concatMap(new Function<Response<ResponseBody>, Observable<?>>() { @Override public Observable<?> apply(Response ignored) { S3File s3File = transmission.getS3File(); return restApi.notifyFileAvailable(s3File.getAction(), transmission.getFormId(), s3File.getFile().getName(), deviceId); } }) .doOnNext(new Consumer<Object>() { @Override public void accept(Object o) { dataBaseDataSource.setFileTransmissionSucceeded(transmissionId); } }) .map(new Function<Object, UploadResult>() { @Override public UploadResult apply(Object ignored) { return new UploadSuccess(surveyInstanceId); } }) .onErrorReturn(new Function<Throwable, UploadResult>() { @Override public UploadResult apply(Throwable throwable) { boolean formNotFound = throwable instanceof HttpException && ( ((HttpException) throwable).code() == 404); if (formNotFound) { dataBaseDataSource.setFileTransmissionFormDeleted(transmissionId); return new UploadFormDeletedError(surveyInstanceId, formId); } else { Timber.e(throwable); dataBaseDataSource.setFileTransmissionFailed(transmissionId); return new UploadError(surveyInstanceId); } } }); } @Inject  SurveyDataRepository(DataSourceFactory dataSourceFactory,
                                DataPointMapper dataPointMapper, RestApi restApi, SurveyMapper surveyMapper,
                                UserMapper userMapper, TransmissionFilenameMapper transmissionFilenameMapper,
                                TransmissionMapper transmissionMapper, FormInstanceMapper formInstanceMapper,
                                FormIdMapper formIdMapper, FormInstanceMetadataMapper formInstanceMetadataMapper,
                                S3RestApi s3RestApi); @Override Observable<List<Survey>> getSurveys(); @Override Observable<List<DataPoint>> getDataPoints(Long surveyGroupId,
            Double latitude, Double longitude, Integer orderBy); @Override Single<DataPoint> getDataPoint(final String datapointId); @Override Observable<Boolean> deleteSurvey(long surveyToDeleteId); @Override Observable<List<User>> getUsers(); @Override Observable<Boolean> editUser(User user); @Override Observable<Boolean> deleteUser(User user); @Override Observable<Long> createUser(String userName); @Override Observable<User> getUser(Long userId); @Override Observable<Boolean> clearResponses(); @Override Observable<Boolean> clearAllData(); @Override Observable<Boolean> unSyncedTransmissionsExist(); @Override Observable<List<String>> getAllTransmissionFileNames(); @Override Observable<List<String>> getFormIds(String surveyId); @Override Observable<List<String>> getFormIds(); @Override Observable<Set<String>> processTransmissions(final String deviceId,
            @NonNull String surveyId); @Override Observable<Set<String>> processTransmissions(final String deviceId); @Nullable @Override Completable setSurveyViewed(long surveyId); @Override Completable cleanDataPoints(Long surveyGroupId); @Override Single<List<InstanceIdUuid>> getSubmittedInstances(); @Override Completable setInstanceStatusToRequested(long id); @Override Single<List<Long>> getPendingSurveyInstances(); @Override Single<FormInstanceMetadata> getFormInstanceData(final Long instanceId,
            final String deviceId); @Override Completable createTransmissions(final Long instanceId, final String formId,
            Set<String> fileNames);  }
@Test public void shouldReturnFormDeletedErrorWhenFormDeleted() { TestObserver observer = new TestObserver<Set<String>>(); MockResponse s3Response = new MockResponse().setResponseCode(200); s3Response.addHeader("ETag", "\"123\""); s3Response.setBody("{}"); MockResponse gaeResponse = new MockResponse().setResponseCode(404); mockWebServer.enqueue(s3Response); mockWebServer.enqueue(gaeResponse); Transmission transmission = new Transmission(1L, 2L, "123", mockS3File); surveyDataRepository.syncTransmission(transmission, "123").subscribe(observer); observer.awaitTerminalEvent(2, TimeUnit.SECONDS); observer.assertNoErrors(); observer.assertValueCount(1); assertTrue(observer.values().get(0) instanceof UploadFormDeletedError); verify(mockDataBaseDataSource, times(0)).setFileTransmissionSucceeded(anyLong()); verify(mockDataBaseDataSource, times(1)).setFileTransmissionFormDeleted(anyLong()); verify(mockDataBaseDataSource, times(0)).setFileTransmissionFailed(anyLong()); }
@Test public void shouldReturnFileTransmissionFailedIfError() { TestObserver observer = new TestObserver<Set<String>>(); MockResponse s3Response = new MockResponse().setResponseCode(200); s3Response.addHeader("ETag", "\"123\""); s3Response.setBody("{}"); MockResponse gaeResponse = new MockResponse().setResponseCode(500); mockWebServer.enqueue(s3Response); mockWebServer.enqueue(gaeResponse); Transmission transmission = new Transmission(1L, 2L, "123", mockS3File); surveyDataRepository.syncTransmission(transmission, "123").subscribe(observer); observer.awaitTerminalEvent(2, TimeUnit.SECONDS); observer.assertNoErrors(); observer.assertValueCount(1); assertTrue(observer.values().get(0) instanceof UploadError); verify(mockDataBaseDataSource, times(0)).setFileTransmissionSucceeded(anyLong()); verify(mockDataBaseDataSource, times(0)).setFileTransmissionFormDeleted(anyLong()); verify(mockDataBaseDataSource, times(1)).setFileTransmissionFailed(anyLong()); }
@Test public void shouldReturnFileTransmissionFailedIfWrongEtag() { TestObserver observer = new TestObserver<Set<String>>(); MockResponse s3Response = new MockResponse().setResponseCode(200); s3Response.addHeader("ETag", "\"1234\""); s3Response.setBody("{}"); MockResponse gaeResponse = new MockResponse().setResponseCode(200); mockWebServer.enqueue(s3Response); mockWebServer.enqueue(gaeResponse); Transmission transmission = new Transmission(1L, 2L, "123", mockS3File); surveyDataRepository.syncTransmission(transmission, "123").subscribe(observer); observer.awaitTerminalEvent(2, TimeUnit.SECONDS); observer.assertNoErrors(); observer.assertValueCount(1); assertTrue(observer.values().get(0) instanceof UploadError); verify(mockDataBaseDataSource, times(0)).setFileTransmissionSucceeded(anyLong()); verify(mockDataBaseDataSource, times(0)).setFileTransmissionFormDeleted(anyLong()); verify(mockDataBaseDataSource, times(1)).setFileTransmissionFailed(anyLong()); }
@Test public void shouldReturnSuccessIfAllCallsSuccessful() { TestObserver observer = new TestObserver<Set<String>>(); MockResponse s3Response = new MockResponse().setResponseCode(200); s3Response.addHeader("ETag", "\"123\""); s3Response.setBody("{}"); MockResponse gaeResponse = new MockResponse().setResponseCode(200); mockWebServer.enqueue(s3Response); mockWebServer.enqueue(gaeResponse); Transmission transmission = new Transmission(1L, 2L, "123", mockS3File); surveyDataRepository.syncTransmission(transmission, "123").subscribe(observer); observer.awaitTerminalEvent(2, TimeUnit.SECONDS); observer.assertNoErrors(); observer.assertValueCount(1); assertTrue(observer.values().get(0) instanceof UploadSuccess); verify(mockDataBaseDataSource, times(1)).setFileTransmissionSucceeded(anyLong()); verify(mockDataBaseDataSource, times(0)).setFileTransmissionFormDeleted(anyLong()); verify(mockDataBaseDataSource, times(0)).setFileTransmissionFailed(anyLong()); }