S3WaitOperatorFactory implements OperatorFactory { @Override public Operator newOperator(OperatorContext context) { return new S3WaitOperator(context); } @Inject  S3WaitOperatorFactory(@Environment Map<String, String> environment); @VisibleForTesting  S3WaitOperatorFactory(
            AmazonS3ClientFactory s3ClientFactory,
            Map<String, String> environment); String getType(); @Override Operator newOperator(OperatorContext context);  }
@Test public void testDefaults() throws Exception { Config config = newConfig(); config.set("_command", BUCKET + "/" + KEY); when(taskRequest.getConfig()).thenReturn(config); ObjectMetadata objectMetadata = new ObjectMetadata(); objectMetadata.setContentType(CONTENT_TYPE); objectMetadata.setContentLength(CONTENT_LENGTH); objectMetadata.setUserMetadata(USER_METADATA); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenReturn(objectMetadata); TaskResult taskResult = operator.run(); verify(s3ClientFactory).create(credentialsCaptor.capture(), clientConfigurationCaptor.capture()); ClientConfiguration clientConfiguration = clientConfigurationCaptor.getValue(); assertThat(clientConfiguration.getProxyHost(), is(nullValue())); assertThat(clientConfiguration.getProxyPort(), is(-1)); assertThat(clientConfiguration.getProxyUsername(), is(nullValue())); assertThat(clientConfiguration.getProxyPassword(), is(nullValue())); verify(s3Client).setS3ClientOptions(s3ClientOptionsCaptor.capture()); S3ClientOptions s3ClientOptions = s3ClientOptionsCaptor.getValue(); assertThat(s3ClientOptions.isPathStyleAccess(), is(false)); AWSCredentials credentials = credentialsCaptor.getValue(); assertThat(credentials.getAWSAccessKeyId(), is(ACCESS_KEY_ID)); assertThat(credentials.getAWSSecretKey(), is(SECRET_ACCESS_KEY)); GetObjectMetadataRequest objectMetadataRequest = objectMetadataRequestCaptor.getValue(); assertThat(objectMetadataRequest.getKey(), is(KEY)); assertThat(objectMetadataRequest.getBucketName(), is(BUCKET)); assertThat(objectMetadataRequest.getSSECustomerKey(), is(nullValue())); Config expectedStoreParams = newConfig(); expectedStoreParams .getNestedOrSetEmpty("s3") .getNestedOrSetEmpty("last_object") .set("metadata", objectMetadata.getRawMetadata()) .set("user_metadata", objectMetadata.getUserMetadata()); assertThat(taskResult.getStoreParams(), is(expectedStoreParams)); }
@Test public void testExponentialBackoff() throws Exception { Config config = newConfig(); config.set("_command", BUCKET + "/" + KEY); when(taskRequest.getConfig()).thenReturn(config); ObjectMetadata objectMetadata = new ObjectMetadata(); objectMetadata.setContentType(CONTENT_TYPE); objectMetadata.setContentLength(CONTENT_LENGTH); objectMetadata.setUserMetadata(USER_METADATA); when(s3Client.getObjectMetadata(any(GetObjectMetadataRequest.class))).thenThrow(NOT_FOUND_EXCEPTION); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); List<Integer> retryIntervals = new ArrayList<>(); for (int i = 0; i < 10; i++) { try { operator.run(); fail(); } catch (TaskExecutionException e) { assertThat(e.isError(), is(false)); assertThat(e.getRetryInterval().isPresent(), is(true)); retryIntervals.add(e.getRetryInterval().get()); Config lastStateParams = e.getStateParams(configFactory).get(); when(taskRequest.getLastStateParams()).thenReturn(lastStateParams); } } for (int i = 1; i < retryIntervals.size(); i++) { int prevInterval = retryIntervals.get(i - 1); int interval = retryIntervals.get(i); assertThat(interval, is(Math.min(MAX_POLL_INTERVAL, prevInterval * 2))); } assertThat(retryIntervals.get(retryIntervals.size() - 1), is(MAX_POLL_INTERVAL)); }
@Test public void testCustomRegionAndPathStyleAccess() throws Exception { Config config = newConfig(); config.set("path_style_access", true); config.set("_command", BUCKET + "/" + KEY); when(taskRequest.getConfig()).thenReturn(config); when(s3Secrets.getSecretOptional("region")).thenReturn(Optional.of(REGION)); when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenThrow(NOT_FOUND_EXCEPTION); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); try { operator.run(); fail(); } catch (TaskExecutionException ignore) { } verify(s3Client).setS3ClientOptions(s3ClientOptionsCaptor.capture()); S3ClientOptions s3ClientOptions = s3ClientOptionsCaptor.getValue(); assertThat(s3ClientOptions.isPathStyleAccess(), is(true)); verify(s3Client).setRegion(RegionUtils.getRegion(REGION)); }
@Test public void testVersionId() throws Exception { Config config = newConfig(); config.set("_command", BUCKET + "/" + KEY); config.set("version_id", VERSION_ID); when(taskRequest.getConfig()).thenReturn(config); when(s3Secrets.getSecretOptional("region")).thenReturn(Optional.of(REGION)); when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenThrow(NOT_FOUND_EXCEPTION); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); try { operator.run(); fail(); } catch (TaskExecutionException ignore) { } GetObjectMetadataRequest objectMetadataRequest = objectMetadataRequestCaptor.getValue(); assertThat(objectMetadataRequest.getVersionId(), is(VERSION_ID)); }
@Test public void testSSEC() throws Exception { Config config = newConfig(); config.set("_command", BUCKET + "/" + KEY); when(taskRequest.getConfig()).thenReturn(config); when(s3Secrets.getSecretOptional("sse_c_key")).thenReturn(Optional.of(SSE_C_KEY)); when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenThrow(NOT_FOUND_EXCEPTION); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); try { operator.run(); fail(); } catch (TaskExecutionException ignore) { } GetObjectMetadataRequest objectMetadataRequest = objectMetadataRequestCaptor.getValue(); assertThat(objectMetadataRequest.getSSECustomerKey().getKey(), is(SSE_C_KEY)); assertThat(objectMetadataRequest.getSSECustomerKey().getAlgorithm(), is("AES256")); assertThat(objectMetadataRequest.getSSECustomerKey().getMd5(), is(nullValue())); }
@Test public void testSSECWithAlgorithmAndMd5() throws Exception { Config config = newConfig(); config.set("_command", BUCKET + "/" + KEY); when(taskRequest.getConfig()).thenReturn(config); when(s3Secrets.getSecretOptional("sse_c_key")).thenReturn(Optional.of(SSE_C_KEY)); when(s3Secrets.getSecretOptional("sse_c_key_algorithm")).thenReturn(Optional.of(SSE_C_KEY_ALGORITHM)); when(s3Secrets.getSecretOptional("sse_c_key_md5")).thenReturn(Optional.of(SSE_C_KEY_MD5)); when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenThrow(NOT_FOUND_EXCEPTION); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); try { operator.run(); fail(); } catch (TaskExecutionException ignore) { } GetObjectMetadataRequest objectMetadataRequest = objectMetadataRequestCaptor.getValue(); assertThat(objectMetadataRequest.getSSECustomerKey().getKey(), is(SSE_C_KEY)); assertThat(objectMetadataRequest.getSSECustomerKey().getAlgorithm(), is(SSE_C_KEY_ALGORITHM)); assertThat(objectMetadataRequest.getSSECustomerKey().getMd5(), is(SSE_C_KEY_MD5)); }
@Test public void testProxy() throws Exception { Config config = newConfig(); config.set("_command", BUCKET + "/" + KEY); when(taskRequest.getConfig()).thenReturn(config); environment.put("http_proxy", "http: when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenThrow(NOT_FOUND_EXCEPTION); Operator operator = factory.newOperator(newContext(projectPath, taskRequest)); try { operator.run(); fail(); } catch (TaskExecutionException ignore) { } verify(s3ClientFactory).create(any(AWSCredentials.class), clientConfigurationCaptor.capture()); ClientConfiguration clientConfiguration = clientConfigurationCaptor.getValue(); assertThat(clientConfiguration.getProxyHost(), is("1.2.3.4")); assertThat(clientConfiguration.getProxyPort(), is(4711)); assertThat(clientConfiguration.getProxyUsername(), is("foo")); assertThat(clientConfiguration.getProxyPassword(), is("bar")); }