@Test
@Ignore("To be investigated - the switch to the strict serializer broken the cancellation check")
public void testWhatTheWhenStreamFailsTheUpstreamIsCancelled() {
    AtomicBoolean subscribed = new AtomicBoolean();
    AtomicBoolean cancelled = new AtomicBoolean();
    Multi<Integer> multi = failingAfter1
            .on().subscribed(sub -> subscribed.set(true))
            .on().cancellation(() -> cancelled.set(true));

    AtomicInteger count = new AtomicInteger();
    Multi<Integer> retry = multi
            .onFailure().retry()
            .when(other -> other.flatMap(l -> count.getAndIncrement() == 0 ? Multi.createFrom().item(l)
                    : Multi.createFrom().failure(new IllegalStateException("boom"))));

    MultiAssertSubscriber<Integer> subscriber = retry.subscribe()
            .withSubscriber(MultiAssertSubscriber.create(10));
    subscriber
            .assertSubscribed()
            .assertReceived(1, 1)
            .assertHasFailedWith(IllegalStateException.class, "boom");

    assertThat(subscribed.get()).isTrue();
    assertThat(cancelled.get()).isTrue();
}
 

@Test
@Ignore("this test is failing on CI - must be investigated")
public void testConcurrentEmissionWithMerge() {
    ExecutorService service = Executors.newFixedThreadPool(10);
    Multi<Integer> m1 = Multi.createFrom().range(1, 100).emitOn(service);
    Multi<Integer> m2 = Multi.createFrom().range(100, 150).emitOn(service).emitOn(service).emitOn(service);
    Multi<Integer> m3 = Multi.createFrom().range(150, 200).emitOn(service).emitOn(service);

    Multi<Integer> merged = m1.transform().byMergingWith(m2, m3);
    MultiAssertSubscriber<Integer> subscriber = merged.subscribe()
            .withSubscriber(MultiAssertSubscriber.create(1000));

    subscriber.await();
    List<Integer> items = subscriber.items();
    assertThat(Collections.singleton(items)).noneSatisfy(list -> assertThat(list).isSorted());
}
 

@Test
@Ignore
public void testConcurrentTransactionPropagation() {
    // create an executor that propagates the transaction context
    ManagedExecutor executor = createExecutor("testConcurrentTransactionPropagation");
    UserTransaction ut = getUserTransaction("testConcurrentTransactionPropagation");

    if (executor == null || ut == null) {
        return; // the implementation does not support transaction propagation
    }

    try {
        int result = transactionalService.testConcurrentTransactionPropagation(executor);
        if (result != UNSUPPORTED) {
            Assert.assertEquals(2, result, "testTransactionPropagation failed%n");
        }
        verifyNoTransaction();
    }
    finally {
        executor.shutdownNow();
    }
}
 

@Ignore("This test case should be re-enabled in case of InternalSDXDistroXTest has been removed")
@Test(dataProvider = TEST_CONTEXT)
@UseSpotInstances
@Description(
        given = "there is a running Cloudbreak",
        when = "a basic DistroX create request is sent",
        then = "DistroX should be available and deletable"
)
public void testDistroXWithPrewarmedImageCanBeCreatedSuccessfully(TestContext testContext) {
    String distrox = resourcePropertyProvider().getName();

    testContext
            .given(distrox, DistroXTestDto.class)
            .when(distroXTestClient.create(), key(distrox))
            .await(STACK_AVAILABLE)
            .then((context, dto, client) -> {
                dto.getResponse();
                return dto;
            })
            .when(distroXTestClient.get())
            .validate();
}
 

@Test
@Ignore("The current JDK parses a number of '12-54.234' to 12 without error!")
public void testParse_mismatchingPattern_throws_exception() {
    AmountNumberToken token = new AmountNumberToken(contextForLocale(US, null), PATTERN);
    ParseContext context = new ParseContext("12-54.234");
    try {
        token.parse(context);
        fail();
    } catch (MonetaryParseException e) {
        assertEquals(e.getInput(), "12354:234");
        assertEquals(e.getErrorIndex(), 0);
        assertEquals(e.getMessage(), "Unparseable amount: \"12354ghd.234\"");
    }
    assertEquals(context.getIndex(), 0);
    assertFalse(context.isComplete());
    assertTrue(context.hasError());
    assertEquals(context.getErrorMessage(), "Unparseable amount: \"12354ghd.234\"");
    assertNull(context.getParsedNumber());
}
 

@Test
@Ignore
public void testDownloadOfRemoteManifest() throws Exception {
    S3BucketService s3BucketService = new S3BucketService(getTransferManagerFactory(), getBackupOperationRequest());

    try {
        s3BucketService.create(BUCKET_NAME);

        AmazonS3 amazonS3Client = getTransferManagerFactory().build(getBackupOperationRequest()).getAmazonS3Client();

        amazonS3Client.putObject(BUCKET_NAME, "cluster/dc/node/manifests/snapshot-name-" + BUCKET_NAME, "hello");

        Thread.sleep(5000);

        ObjectListing objectListing = amazonS3Client.listObjects(BUCKET_NAME);

        objectListing.getObjectSummaries().forEach(summary -> System.out.println(summary.getKey()));

        RestoreOperationRequest restoreOperationRequest = new RestoreOperationRequest();
        restoreOperationRequest.storageLocation = new StorageLocation("s3://" + BUCKET_NAME + "/cluster/dc/node");

        S3Restorer s3Restorer = new S3Restorer(getTransferManagerFactory(), new FixedTasksExecutorSupplier(), restoreOperationRequest);

        final Path downloadedFile = s3Restorer.downloadFileToDir(Paths.get("/tmp"), Paths.get("manifests"), new Predicate<String>() {
            @Override
            public boolean test(final String s) {
                return s.contains("manifests/snapshot-name");
            }
        });

        assertTrue(Files.exists(downloadedFile));
    } finally {
        s3BucketService.delete(BUCKET_NAME);
        deleteDirectory(Paths.get(target("commitlog_download_dir")));
    }
}
 

@Ignore (value = "NonValueAggregate query is not supported by the SDK as of now")
@Test(groups = { "simple" }, timeOut = 2 * TIMEOUT, dataProvider = "queryMetricsArgProvider")
public void queryDocumentsWithAggregates(boolean qmEnabled) throws Exception {

    FeedOptions options = new FeedOptions();
    options.setEnableCrossPartitionQuery(true);
    options.setPopulateQueryMetrics(qmEnabled);
    options.setMaxDegreeOfParallelism(2);
    
    for (QueryConfig queryConfig : queryConfigs) {
        // Cross partition Non value aggregates are not supported
        if(queryConfig.query.contains("VALUE")){
            options.setEnableCrossPartitionQuery(true);
        }else{
            options.setEnableCrossPartitionQuery(false);
        }

        Observable<FeedResponse<Document>> queryObservable = client
            .queryDocuments(createdCollection.getSelfLink(), queryConfig.query, options);

        FeedResponseListValidator<Document> validator = new FeedResponseListValidator.Builder<Document>()
            .withAggregateValue(queryConfig.expected)
            .numberOfPages(1)
            .hasValidQueryMetrics(qmEnabled)
            .build();

        validateQuerySuccess(queryObservable, validator);
    }
}
 

/**
 * Test for #119.
 */
@Ignore
@Test
void gitlabMappingAsUser() throws Exception {
  final GitLabToken userToken = createToken(user, userPassword, false);

  try (SvnTestServer server = createServer(userToken, dir -> new GitLabMappingConfig(dir, GitCreateMode.EMPTY))) {
    openSvnRepository(server, gitlabProject, root, rootPassword).getLatestRevision();
  }
}
 

@Test
@Ignore // FIXME: see https://github.com/martinpaljak/GlobalPlatformPro/issues/116
public void testCPLCDateParse() throws Exception {
    byte[] b = HexUtils.hex2bin("1210");
    Assert.assertEquals("2011-07-29", GPData.CPLC.toDate(b));
    b = HexUtils.hex2bin("0000");
    Assert.assertEquals("2010-01-01", GPData.CPLC.toDate(b));
    System.out.println("Today is " + HexUtils.bin2hex(GPData.CPLC.today()));
}
 

@Test
@Ignore
public void testHardlinkingBackupAndRestore() throws Exception {
    liveCassandraTest(hardlinkingArguments());
}
 

@Test
@Ignore
public void testHardlinkingBackupAndRestore() throws Exception {
    liveCassandraTest(hardlinkingArguments());
}
 

@Test
@Ignore
public void testHardlinkingBackupAndRestore() throws Exception {
    liveCassandraTest(hardlinkingArguments());
}
 

@Ignore
@Test(dataProvider = TEST_CONTEXT_WITH_MOCK)
@Description(
        given = "there is a running DistroX cluster",
        when = "a scale, start stop called many times",
        then = "the cluster should be available")
public void testCreateNewRegularDistroXClusterScaleStartStop(MockedTestContext testContext, ITestContext testNgContext) {
    DistroXStartStopTestParameters params = new DistroXStartStopTestParameters(testNgContext.getCurrentXmlTest().getAllParameters());
    String stack = resourcePropertyProvider().getName();
    int step = params.getStep();
    int current = step;
    DistroXTestDto currentContext = testContext
            .given(DIX_NET_KEY, DistroXNetworkTestDto.class)
            .given(DIX_IMG_KEY, DistroXImageTestDto.class)
            .withImageCatalog(getImageCatalogName(testContext))
            .withImageId(IMAGE_CATALOG_ID)
            .given(CM_FOR_DISTRO_X, DistroXClouderaManagerTestDto.class)
            .given(CLUSTER_KEY, DistroXClusterTestDto.class)
            .withValidateBlueprint(false)
            .withClouderaManager(CM_FOR_DISTRO_X)
            .given(stack, DistroXTestDto.class)
            .withGatewayPort(testContext.getSparkServer().getPort())
            .withCluster(CLUSTER_KEY)
            .withName(stack)
            .withImageSettings(DIX_IMG_KEY)
            .withNetwork(DIX_NET_KEY)
            .when(distroXClient.create(), key(stack))
            .awaitForFlow(key(stack))
            .await(STACK_AVAILABLE, key(stack));
    for (int i = 0; i < params.getTimes(); i++, current += step) {
        currentContext = currentContext
                .when(distroXClient.stop(), key(stack))
                .awaitForFlow(key(stack))
                .await(STACK_STOPPED, key(stack))
                .then(auditGrpcServiceAssertion::distroxStop)
                .when(distroXClient.start(), key(stack))
                .awaitForFlow(key(stack))
                .await(STACK_AVAILABLE, key(stack))
                .then(auditGrpcServiceAssertion::distroxStart)
                .when(distroXClient.scale(params.getHostgroup(), current))
                .awaitForFlow(key(stack))
                .await(DistroXTestDto.class, STACK_AVAILABLE, key(stack), POLLING_INTERVAL);
    }

    currentContext
            .validate();
}