diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/IntegrationTestBase.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/IntegrationTestBase.java index 5fe821608315..1d4b3edd0bbb 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/IntegrationTestBase.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/IntegrationTestBase.java @@ -6,7 +6,6 @@ import com.azure.core.amqp.AmqpTransportType; import com.azure.core.amqp.ProxyAuthenticationType; import com.azure.core.amqp.ProxyOptions; -import com.azure.core.amqp.implementation.ConnectionStringProperties; import com.azure.core.amqp.models.AmqpMessageBody; import com.azure.core.credential.TokenCredential; import com.azure.core.experimental.util.tracing.LoggingTracerProvider; @@ -43,6 +42,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -72,6 +72,7 @@ public abstract class IntegrationTestBase extends TestBase { private List toClose = new ArrayList<>(); private String testName; private final Scheduler scheduler = Schedulers.parallel(); + private final AtomicReference credentialCached = new AtomicReference<>(); protected static final byte[] CONTENTS_BYTES = "Some-contents".getBytes(StandardCharsets.UTF_8); protected String sessionId; @@ -90,7 +91,7 @@ public void setupTest(TestContextManager testContextManager) { logger.info("========= SET-UP [{}] =========", testName); - assumeTrue(getTestMode() == TestMode.RECORD); + assertRunnable(); toClose = new ArrayList<>(); optionsWithTracing = new ClientOptions().setTracingOptions(new LoggingTracerProvider.LoggingTracingOptions()); @@ -108,39 +109,6 @@ public void teardownTest() { dispose(); } - /** - * Gets the test mode for this API test. If AZURE_TEST_MODE equals {@link TestMode#RECORD} and Event Hubs connection - * string is set, then we return {@link TestMode#RECORD}. Otherwise, {@link TestMode#PLAYBACK} is returned. - */ - @Override - public TestMode getTestMode() { - if (super.getTestMode() == TestMode.PLAYBACK) { - return TestMode.PLAYBACK; - } - - return CoreUtils.isNullOrEmpty(getConnectionString()) ? TestMode.PLAYBACK : TestMode.RECORD; - } - - public static String getConnectionString() { - return TestUtils.getConnectionString(false); - } - - public static String getConnectionString(boolean withSas) { - return TestUtils.getConnectionString(withSas); - } - - protected static ConnectionStringProperties getConnectionStringProperties() { - return new ConnectionStringProperties(getConnectionString(false)); - } - - protected static ConnectionStringProperties getConnectionStringProperties(boolean withSas) { - return new ConnectionStringProperties(getConnectionString(withSas)); - } - - public String getFullyQualifiedDomainName() { - return TestUtils.getFullyQualifiedDomainName(); - } - /** * Gets the name of the queue. * @@ -204,46 +172,67 @@ public ProxyOptions getProxyConfiguration() { } /** - * Creates a new instance of {@link ServiceBusClientBuilder} with the default integration test settings and uses a - * connection string to authenticate. + * Creates a new instance of {@link ServiceBusClientBuilder} with authentication set up in {@link TestMode#LIVE} and + * {@link TestMode#RECORD} modes. + * + * @return the builder with authentication set up. + * @throws org.opentest4j.TestAbortedException if the test mode is {@link TestMode#PLAYBACK}. */ - protected ServiceBusClientBuilder getBuilder() { - return getBuilder(false); + protected ServiceBusClientBuilder getAuthenticatedBuilder() { + final TestMode mode = super.getTestMode(); + final ServiceBusClientBuilder builder = new ServiceBusClientBuilder(); + if (mode == TestMode.LIVE) { + final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(false); + assumeTrue(!CoreUtils.isNullOrEmpty(fullyQualifiedDomainName), "FullyQualifiedDomainName is not set."); + final TokenCredential credential = TestUtils.getPipelineCredential(credentialCached); + return builder.credential(fullyQualifiedDomainName, credential); + } else if (mode == TestMode.RECORD) { + final String connectionString = TestUtils.getConnectionString(false); + if (CoreUtils.isNullOrEmpty(connectionString)) { + final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(false); + assumeTrue(!CoreUtils.isNullOrEmpty(fullyQualifiedDomainName), "FullyQualifiedDomainName is not set."); + final TokenCredential credential = new DefaultAzureCredentialBuilder().build(); + return builder.credential(fullyQualifiedDomainName, credential); + } else { + return builder.connectionString(connectionString); + } + } else { + // Throws org.opentest4j.TestAbortedException exception. + assumeTrue(false, "Integration tests are not enabled in playback mode."); + return null; + } } /** - * Creates a new instance of {@link ServiceBusClientBuilder} with the default integration test settings and uses a - * connection string to authenticate if {@code useCredentials} is false. Otherwise, uses a service principal through - * {@link com.azure.identity.ClientSecretCredential}. + * Creates a new instance of {@link ServiceBusClientBuilder} with the default integration test settings. */ - protected ServiceBusClientBuilder getBuilder(boolean useCredentials) { - final ServiceBusClientBuilder builder = new ServiceBusClientBuilder() + protected ServiceBusClientBuilder getBuilder() { + return getAuthenticatedBuilder() .proxyOptions(ProxyOptions.SYSTEM_DEFAULTS) .retryOptions(RETRY_OPTIONS) .clientOptions(optionsWithTracing) .transportType(AmqpTransportType.AMQP) .scheduler(scheduler) .configuration(v1OrV2(true)); + } - logger.info("Getting Builder using credentials : [{}] ", useCredentials); - if (useCredentials) { - final String fullyQualifiedDomainName = getFullyQualifiedDomainName(); - - assumeTrue(fullyQualifiedDomainName != null && !fullyQualifiedDomainName.isEmpty(), - "AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME variable needs to be set when using credentials."); - - final TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build(); - - return builder.credential(fullyQualifiedDomainName, tokenCredential); + protected ServiceBusClientBuilder getBuilder(boolean sharedConnection) { + ServiceBusClientBuilder builder; + if (sharedConnection && sharedBuilder == null) { + sharedBuilder = getBuilder(); + builder = sharedBuilder; + } else if (sharedConnection) { + builder = sharedBuilder; } else { - return builder.connectionString(getConnectionString()); + builder = getBuilder(); } + return builder; } - protected ServiceBusSenderClientBuilder getSenderBuilder(boolean useCredentials, MessagingEntityType entityType, + protected ServiceBusSenderClientBuilder getSenderBuilder(MessagingEntityType entityType, int entityIndex, boolean isSessionAware, boolean sharedConnection) { - ServiceBusClientBuilder builder = getBuilder(useCredentials, sharedConnection); + ServiceBusClientBuilder builder = getBuilder(sharedConnection); switch (entityType) { case QUEUE: final String queueName = isSessionAware ? getSessionQueueName(entityIndex) : getQueueName(entityIndex); @@ -260,10 +249,10 @@ protected ServiceBusSenderClientBuilder getSenderBuilder(boolean useCredentials, } } - protected ServiceBusReceiverClientBuilder getReceiverBuilder(boolean useCredentials, MessagingEntityType entityType, + protected ServiceBusReceiverClientBuilder getReceiverBuilder(MessagingEntityType entityType, int entityIndex, boolean sharedConnection) { - ServiceBusClientBuilder builder = getBuilder(useCredentials, sharedConnection); + ServiceBusClientBuilder builder = getBuilder(sharedConnection); switch (entityType) { case QUEUE: final String queueName = getQueueName(entityIndex); @@ -283,10 +272,10 @@ protected ServiceBusReceiverClientBuilder getReceiverBuilder(boolean useCredenti } } - protected ServiceBusSessionReceiverClientBuilder getSessionReceiverBuilder(boolean useCredentials, - MessagingEntityType entityType, int entityIndex, boolean sharedConnection, AmqpRetryOptions retryOptions) { + protected ServiceBusSessionReceiverClientBuilder getSessionReceiverBuilder(MessagingEntityType entityType, + int entityIndex, boolean sharedConnection, AmqpRetryOptions retryOptions) { - ServiceBusClientBuilder builder = getBuilder(useCredentials, sharedConnection); + ServiceBusClientBuilder builder = getBuilder(sharedConnection); switch (entityType) { case QUEUE: @@ -453,19 +442,6 @@ protected void assertMessageEquals(ServiceBusReceivedMessage message, String mes } } - protected ServiceBusClientBuilder getBuilder(boolean useCredentials, boolean sharedConnection) { - ServiceBusClientBuilder builder; - if (sharedConnection && sharedBuilder == null) { - sharedBuilder = getBuilder(useCredentials); - builder = sharedBuilder; - } else if (sharedConnection) { - builder = sharedBuilder; - } else { - builder = getBuilder(useCredentials); - } - return builder; - } - protected final Configuration v1OrV2(boolean isV2) { final TestConfigurationSource configSource = new TestConfigurationSource(); if (isV2) { @@ -486,4 +462,38 @@ protected final Configuration v1OrV2(boolean isV2) { return new ConfigurationBuilder(configSource) .build(); } + + /** + * Asserts that if the integration tests can be run. This method is expected to be called at the beginning of each + * test run. + * + * @throws org.opentest4j.TestAbortedException if the integration tests cannot be run. + */ + protected void assertRunnable() { + final TestMode mode = super.getTestMode(); + if (mode == TestMode.PLAYBACK) { + // AMQP traffic never gets recorded so there is no PLAYBACK supported. + assumeTrue(false, "Skipping integration tests in playback mode."); + return; + } + + if (mode == TestMode.RECORD) { + // RECORD mode used in SDK-dev setup not on CI pipeline. + if (!CoreUtils.isNullOrEmpty(TestUtils.getConnectionString(false))) { + // integration tests are runnable using the connection string. + return; + } + if (!CoreUtils.isNullOrEmpty(TestUtils.getFullyQualifiedDomainName(false))) { + // best effort check: + // integration tests are potentially runnable using DefaultAzureCredential. Here we're assuming that + // in RECORD mode with FullyQualifiedDomainName set, the dev environment is also set up for one of the + // token credential type in DefaultAzureCredential (all token credentials requires FullyQualifiedDomainName). + return; + } + assumeTrue(false, "Not running integration in record mode (missing authentication set up)."); + return; + } + // The CI pipeline is expected to have federated identity configured, so tests are runnable. + assert mode == TestMode.LIVE; + } } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxyReceiveTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxyReceiveTest.java index f35d3e10ded5..9909ca951f25 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxyReceiveTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxyReceiveTest.java @@ -79,21 +79,18 @@ public void receiveMessage() { final String messageTracking = UUID.randomUUID().toString(); final List messages = TestUtils.getServiceBusMessages(NUMBER_OF_EVENTS, messageTracking); - final ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder() + final ServiceBusSenderAsyncClient sender = getAuthenticatedBuilder() .transportType(AmqpTransportType.AMQP_WEB_SOCKETS) .verifyMode(SslDomain.VerifyMode.ANONYMOUS_PEER) - .connectionString(getConnectionString()) - .sender() .queueName(queueName) .buildAsyncClient(); toClose(sender); - final ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder() + final ServiceBusReceiverAsyncClient receiver = getAuthenticatedBuilder() .transportType(AmqpTransportType.AMQP_WEB_SOCKETS) .verifyMode(SslDomain.VerifyMode.ANONYMOUS_PEER) - .connectionString(getConnectionString()) .receiver() .receiveMode(ServiceBusReceiveMode.RECEIVE_AND_DELETE) .queueName(queueName) diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySelectorTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySelectorTest.java index 1dff7bec6184..64b67c3ba64b 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySelectorTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySelectorTest.java @@ -68,8 +68,7 @@ public void connectFailed(URI uri, SocketAddress sa, IOException ioe) { }); final ServiceBusMessage message = new ServiceBusMessage(BinaryData.fromString("Hello")); - final ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusSenderAsyncClient sender = getAuthenticatedBuilder() .transportType(AmqpTransportType.AMQP_WEB_SOCKETS) .retryOptions(new AmqpRetryOptions().setTryTimeout(Duration.ofSeconds(10))) .sender() diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySendTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySendTest.java index 78833565375c..51b7583ca4c5 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySendTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ProxySendTest.java @@ -77,8 +77,7 @@ public void sendEvents() { final String messageId = UUID.randomUUID().toString(); final List messages = TestUtils.getServiceBusMessages(NUMBER_OF_EVENTS, messageId); - final ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusSenderAsyncClient sender = getAuthenticatedBuilder() .transportType(AmqpTransportType.AMQP_WEB_SOCKETS) .verifyMode(SslDomain.VerifyMode.ANONYMOUS_PEER) .retryOptions(new AmqpRetryOptions().setTryTimeout(Duration.ofSeconds(10))) diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusClientBuilderTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusClientBuilderTest.java index 57e91cd71f69..f470a7155f84 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusClientBuilderTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusClientBuilderTest.java @@ -6,7 +6,6 @@ import com.azure.core.amqp.AmqpTransportType; import com.azure.core.amqp.ProxyAuthenticationType; import com.azure.core.amqp.ProxyOptions; -import com.azure.core.amqp.implementation.ConnectionStringProperties; import com.azure.core.credential.AzureNamedKeyCredential; import com.azure.core.credential.AzureSasCredential; import com.azure.core.util.Configuration; @@ -21,7 +20,6 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; -import reactor.test.StepVerifier; import java.net.InetSocketAddress; import java.net.Proxy; @@ -34,8 +32,6 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -import static java.nio.charset.StandardCharsets.UTF_8; - class ServiceBusClientBuilderTest extends IntegrationTestBase { private static final String NAMESPACE_NAME = "dummyNamespaceName"; private static final String DEFAULT_DOMAIN_NAME = TestUtils.getEndpoint().substring(1) + "/"; @@ -266,55 +262,6 @@ public void testConnectionStringWithSas() { .connectionString("Endpoint=sb://sb-name" + TestUtils.getEndpoint() + "/;EntityPath=sb-name")); } - @Test - public void testBatchSendEventByAzureNameKeyCredential() { - ConnectionStringProperties properties = getConnectionStringProperties(); - String fullyQualifiedNamespace = getFullyQualifiedDomainName(); - String sharedAccessKeyName = properties.getSharedAccessKeyName(); - String sharedAccessKey = properties.getSharedAccessKey(); - String queueName = getQueueName(TestUtils.USE_CASE_DEFAULT); - - final ServiceBusMessage testData = new ServiceBusMessage(TEST_MESSAGE.getBytes(UTF_8)); - - ServiceBusSenderAsyncClient senderAsyncClient = toClose(new ServiceBusClientBuilder() - .credential(fullyQualifiedNamespace, new AzureNamedKeyCredential(sharedAccessKeyName, sharedAccessKey)) - .sender() - .queueName(queueName) - .buildAsyncClient()); - StepVerifier.create( - senderAsyncClient.createMessageBatch().flatMap(batch -> { - assertTrue(batch.tryAddMessage(testData)); - return senderAsyncClient.sendMessages(batch); - })) - .expectComplete() - .verify(TIMEOUT); - } - - @Test - public void testBatchSendEventByAzureSasCredential() { - ConnectionStringProperties properties = getConnectionStringProperties(true); - String fullyQualifiedNamespace = getFullyQualifiedDomainName(); - String sharedAccessSignature = properties.getSharedAccessSignature(); - String queueName = getQueueName(TestUtils.USE_CASE_DEFAULT); - - final ServiceBusMessage testData = new ServiceBusMessage(TEST_MESSAGE.getBytes(UTF_8)); - - ServiceBusSenderAsyncClient senderAsyncClient = toClose(new ServiceBusClientBuilder() - .credential(fullyQualifiedNamespace, - new AzureSasCredential(sharedAccessSignature)) - .sender() - .queueName(queueName) - .buildAsyncClient()); - - StepVerifier.create( - senderAsyncClient.createMessageBatch().flatMap(batch -> { - assertTrue(batch.tryAddMessage(testData)); - return senderAsyncClient.sendMessages(batch); - })) - .expectComplete() - .verify(TIMEOUT); - } - @Test public void testConnectionWithAzureNameKeyCredential() { String fullyQualifiedNamespace = "sb-name" + TestUtils.getEndpoint(); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusMixClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusMixClientIntegrationTest.java index 9709bc8f7726..316731660eda 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusMixClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusMixClientIntegrationTest.java @@ -84,7 +84,6 @@ protected void afterTest() { void crossEntityQueueTransaction(boolean isSessionEnabled) throws InterruptedException { // Arrange - final boolean useCredentials = false; final MessagingEntityType entityType = MessagingEntityType.QUEUE; final int receiveQueueAIndex = TestUtils.USE_CASE_TXN_1; final int sendQueueBIndex = TestUtils.USE_CASE_TXN_2; @@ -99,7 +98,7 @@ void crossEntityQueueTransaction(boolean isSessionEnabled) throws InterruptedExc message.setSessionId(sessionId); final List messages = Arrays.asList(message); - ServiceBusClientBuilder builder = getBuilder(useCredentials).enableCrossEntityTransactions(); + ServiceBusClientBuilder builder = getBuilder().enableCrossEntityTransactions(); // Initialize sender final ServiceBusSenderAsyncClient senderAsyncA = toClose(builder.sender().queueName(queueA).buildAsyncClient()); @@ -171,7 +170,7 @@ void crossEntityQueueTransaction(boolean isSessionEnabled) throws InterruptedExc // Verify that message is received by queue B if (!isSessionEnabled) { - setSenderAndReceiver(entityType, sendQueueBIndex, false); + setSenderAndReceiver(entityType, sendQueueBIndex); StepVerifier.create(receiver.receiveMessages().take(1)) .assertNext(receivedMessage -> { assertMessageEquals(receivedMessage, messageId, isSessionEnabled); @@ -193,7 +192,6 @@ void crossEntityQueueTransaction(boolean isSessionEnabled) throws InterruptedExc void crossEntitySubscriptionTransaction(boolean isSessionEnabled) throws InterruptedException { // Arrange - final boolean useCredentials = false; final MessagingEntityType entityType = MessagingEntityType.SUBSCRIPTION; final int receiveQueueAIndex = TestUtils.USE_CASE_TXN_1; final int sendQueueBIndex = TestUtils.USE_CASE_TXN_2; @@ -209,7 +207,7 @@ void crossEntitySubscriptionTransaction(boolean isSessionEnabled) throws Interru message.setSessionId(sessionId); final List messages = Arrays.asList(message); - ServiceBusClientBuilder builder = getBuilder(useCredentials).enableCrossEntityTransactions(); + ServiceBusClientBuilder builder = getBuilder().enableCrossEntityTransactions(); // Initialize sender final ServiceBusSenderAsyncClient senderAsyncA = toClose(builder.sender().topicName(topicA).buildAsyncClient()); @@ -289,7 +287,7 @@ void crossEntitySubscriptionTransaction(boolean isSessionEnabled) throws Interru // Verify that message is received by queue B if (!isSessionEnabled) { - setSenderAndReceiver(entityType, sendQueueBIndex, false); + setSenderAndReceiver(entityType, sendQueueBIndex); StepVerifier.create(receiver.receiveMessages().take(1)) .assertNext(receivedMessage -> { assertMessageEquals(receivedMessage, messageId, isSessionEnabled); @@ -311,7 +309,6 @@ void crossEntitySubscriptionTransaction(boolean isSessionEnabled) throws Interru void crossEntityQueueTransactionWithReceiverSenderTest(boolean isSessionEnabled) throws InterruptedException { // Arrange - final boolean useCredentials = false; final MessagingEntityType entityType = MessagingEntityType.QUEUE; final int receiveQueueAIndex = TestUtils.USE_CASE_TXN_1; final int sendQueueBIndex = TestUtils.USE_CASE_TXN_2; @@ -326,7 +323,7 @@ void crossEntityQueueTransactionWithReceiverSenderTest(boolean isSessionEnabled) message.setSessionId(sessionId); final List messages = Arrays.asList(message); - ServiceBusClientBuilder builder = getBuilder(useCredentials).enableCrossEntityTransactions(); + ServiceBusClientBuilder builder = getBuilder().enableCrossEntityTransactions(); // Initialize sender final ServiceBusSenderAsyncClient senderAsyncA = toClose(builder.sender().queueName(queueA).buildAsyncClient()); @@ -376,7 +373,7 @@ void crossEntityQueueTransactionWithReceiverSenderTest(boolean isSessionEnabled) // Assert // Verify that message is received by entity B if (!isSessionEnabled) { - setSenderAndReceiver(entityType, sendQueueBIndex, false); + setSenderAndReceiver(entityType, sendQueueBIndex); StepVerifier.create(receiver.receiveMessages() .filter(receiveMessage -> messageId.equals(receiveMessage.getMessageId())).take(1)) .assertNext(receivedMessage -> { @@ -399,7 +396,6 @@ void crossEntityQueueTransactionWithReceiverSenderTest(boolean isSessionEnabled) void crossEntitySubscriptionTransactionWithReceiverSenderTest(boolean isSessionEnabled) throws InterruptedException { // Arrange - final boolean useCredentials = false; final MessagingEntityType entityType = MessagingEntityType.SUBSCRIPTION; final int receiveQueueAIndex = TestUtils.USE_CASE_TXN_1; final int sendQueueBIndex = TestUtils.USE_CASE_TXN_2; @@ -414,7 +410,7 @@ void crossEntitySubscriptionTransactionWithReceiverSenderTest(boolean isSessionE message.setSessionId(sessionId); final List messages = Arrays.asList(message); - ServiceBusClientBuilder builder = getBuilder(useCredentials).enableCrossEntityTransactions(); + ServiceBusClientBuilder builder = getBuilder().enableCrossEntityTransactions(); // Initialize sender final ServiceBusSenderAsyncClient senderAsyncA = toClose(builder.sender().topicName(topicA).buildAsyncClient()); @@ -462,7 +458,7 @@ void crossEntitySubscriptionTransactionWithReceiverSenderTest(boolean isSessionE // Assert // Verify that message is received by entity B if (!isSessionEnabled) { - setSenderAndReceiver(entityType, sendQueueBIndex, false); + setSenderAndReceiver(entityType, sendQueueBIndex); StepVerifier.create(receiver.receiveMessages().take(1)) .assertNext(receivedMessage -> { assertMessageEquals(receivedMessage, messageId, isSessionEnabled); @@ -476,13 +472,13 @@ void crossEntitySubscriptionTransactionWithReceiverSenderTest(boolean isSessionE /** * Sets the sender and receiver. If session is enabled, then a single-named session receiver is created. */ - private void setSenderAndReceiver(MessagingEntityType entityType, int entityIndex, boolean useCredentials) { + private void setSenderAndReceiver(MessagingEntityType entityType, int entityIndex) { final boolean isSessionAware = false; final boolean sharedConnection = true; - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionAware, sharedConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionAware, sharedConnection) .buildAsyncClient()); - this.receiver = toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, sharedConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, sharedConnection) .receiveMode(ServiceBusReceiveMode.RECEIVE_AND_DELETE) .disableAutoComplete() .buildAsyncClient()); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusNonFederatedIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusNonFederatedIntegrationTest.java new file mode 100644 index 000000000000..37c89b67b80d --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusNonFederatedIntegrationTest.java @@ -0,0 +1,134 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.messaging.servicebus; + +import com.azure.core.amqp.implementation.ConnectionStringProperties; +import com.azure.core.credential.AzureNamedKeyCredential; +import com.azure.core.credential.AzureSasCredential; +import com.azure.core.test.TestMode; +import com.azure.core.util.CoreUtils; +import com.azure.core.util.logging.ClientLogger; +import com.azure.messaging.servicebus.administration.ServiceBusAdministrationAsyncClient; +import com.azure.messaging.servicebus.administration.ServiceBusAdministrationClientBuilder; +import com.azure.messaging.servicebus.administration.models.NamespaceProperties; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import reactor.test.StepVerifier; + +import java.time.Duration; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static com.azure.messaging.servicebus.TestUtils.getConnectionString; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Live tests with non-federated authentication. + */ +@Tag("integration") +@Execution(ExecutionMode.SAME_THREAD) +public class ServiceBusNonFederatedIntegrationTest extends IntegrationTestBase { + private static final Duration TIMEOUT = Duration.ofSeconds(20); + private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(10); + private static final String TEST_MESSAGE = "SSLorem ipsum dolor sit amet, consectetur adipiscing elit. Donec vehicula posuere lobortis. Aliquam finibus volutpat dolor, faucibus pellentesque ipsum bibendum vitae. Class aptent taciti sociosqu ad litora torquent per conubia nostra, per inceptos himenaeos. Ut sit amet urna hendrerit, dapibus justo a, sodales justo. Mauris finibus augue id pulvinar congue. Nam maximus luctus ipsum, at commodo ligula euismod ac. Phasellus vitae lacus sit amet diam porta placerat. \nUt sodales efficitur sapien ut posuere. Morbi sed tellus est. Proin eu erat purus. Proin massa nunc, condimentum id iaculis dignissim, consectetur et odio. Cras suscipit sem eu libero aliquam tincidunt. Nullam ut arcu suscipit, eleifend velit in, cursus libero. Ut eleifend facilisis odio sit amet feugiat. Phasellus at nunc sit amet elit sagittis commodo ac in nisi. Fusce vitae aliquam quam. Integer vel nibh euismod, tempus elit vitae, pharetra est. Duis vulputate enim a elementum dignissim. Morbi dictum enim id elit scelerisque, in elementum nulla pharetra. \nAenean aliquet aliquet condimentum. Proin dapibus dui id libero tempus feugiat. Sed commodo ligula a lectus mattis, vitae tincidunt velit auctor. Fusce quis semper dui. Phasellus eu efficitur sem. Ut non sem sit amet enim condimentum venenatis id dictum massa. Nullam sagittis lacus a neque sodales, et ultrices arcu mattis. Aliquam erat volutpat. \nAenean fringilla quam elit, id mattis purus vestibulum nec. Praesent porta eros in dapibus molestie. Vestibulum orci libero, tincidunt et turpis eget, condimentum lobortis enim. Fusce suscipit ante et mauris consequat cursus nec laoreet lorem. Maecenas in sollicitudin diam, non tincidunt purus. Nunc mauris purus, laoreet eget interdum vitae, placerat a sapien. In mi risus, blandit eu facilisis nec, molestie suscipit leo. Pellentesque molestie urna vitae dui faucibus bibendum. \nDonec quis ipsum ultricies, imperdiet ex vel, scelerisque eros. Ut at urna arcu. Vestibulum rutrum odio dolor, vitae cursus nunc pulvinar vel. Donec accumsan sapien in malesuada tempor. Maecenas in condimentum eros. Sed vestibulum facilisis massa a iaculis. Etiam et nibh felis. Donec maximus, sem quis vestibulum gravida, turpis risus congue dolor, pharetra tincidunt lectus nisi at velit."; + + ServiceBusNonFederatedIntegrationTest() { + super(new ClientLogger(ServiceBusNonFederatedIntegrationTest.class)); + } + + @Test + public void testBatchSendEventByAzureNameKeyCredential() { + ConnectionStringProperties properties = new ConnectionStringProperties(TestUtils.getConnectionString(false)); + String fullyQualifiedNamespace = TestUtils.getFullyQualifiedDomainName(true); + String sharedAccessKeyName = properties.getSharedAccessKeyName(); + String sharedAccessKey = properties.getSharedAccessKey(); + String queueName = getQueueName(TestUtils.USE_CASE_DEFAULT); + + final ServiceBusMessage testData = new ServiceBusMessage(TEST_MESSAGE.getBytes(UTF_8)); + + ServiceBusSenderAsyncClient senderAsyncClient = toClose(new ServiceBusClientBuilder() + .credential(fullyQualifiedNamespace, new AzureNamedKeyCredential(sharedAccessKeyName, sharedAccessKey)) + .sender() + .queueName(queueName) + .buildAsyncClient()); + StepVerifier.create( + senderAsyncClient.createMessageBatch().flatMap(batch -> { + assertTrue(batch.tryAddMessage(testData)); + return senderAsyncClient.sendMessages(batch); + })) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + public void testBatchSendEventByAzureSasCredential() { + ConnectionStringProperties properties = new ConnectionStringProperties(TestUtils.getConnectionString(true)); + String fullyQualifiedNamespace = TestUtils.getFullyQualifiedDomainName(true); + String sharedAccessSignature = properties.getSharedAccessSignature(); + String queueName = getQueueName(TestUtils.USE_CASE_DEFAULT); + + final ServiceBusMessage testData = new ServiceBusMessage(TEST_MESSAGE.getBytes(UTF_8)); + + ServiceBusSenderAsyncClient senderAsyncClient = toClose(new ServiceBusClientBuilder() + .credential(fullyQualifiedNamespace, + new AzureSasCredential(sharedAccessSignature)) + .sender() + .queueName(queueName) + .buildAsyncClient()); + + StepVerifier.create( + senderAsyncClient.createMessageBatch().flatMap(batch -> { + assertTrue(batch.tryAddMessage(testData)); + return senderAsyncClient.sendMessages(batch); + })) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void azureSasCredentialsTest() { + final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(false); + assumeTrue(!CoreUtils.isNullOrEmpty(fullyQualifiedDomainName), "FullyQualifiedDomainName is not set."); + String connectionString = getConnectionString(true); + Pattern sasPattern = Pattern.compile("SharedAccessSignature=(.*);?", Pattern.CASE_INSENSITIVE); + Matcher matcher = sasPattern.matcher(connectionString); + assertTrue(matcher.find(), "Couldn't find SAS from connection string"); + + ServiceBusAdministrationAsyncClient client = new ServiceBusAdministrationClientBuilder() + .endpoint("https://" + fullyQualifiedDomainName) + .credential(new AzureSasCredential(matcher.group(1))) + .buildAsyncClient(); + + StepVerifier.create(client.getNamespacePropertiesWithResponse()) + .assertNext(response -> { + final NamespaceProperties np = response.getValue(); + assertNotNull(np.getName()); + }) + .expectComplete() + .verify(DEFAULT_TIMEOUT); + } + + @Override + protected void assertRunnable() { + final TestMode mode = super.getTestMode(); + if (mode == TestMode.PLAYBACK) { + // AMQP traffic never gets recorded so there is no PLAYBACK supported. + assumeTrue(false, "Skipping integration tests in playback mode."); + return; + } + + if (mode == TestMode.RECORD || mode == TestMode.LIVE) { + // RECORD mode used in SDK-dev setup and LIVE mode on CI pipeline. + if (!CoreUtils.isNullOrEmpty(TestUtils.getConnectionString(false))) { + // non-federated integration tests are runnable using the connection string. + return; + } + } + assumeTrue(false, "Not running integration in record mode (missing authentication set up)."); + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusProcessorClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusProcessorClientIntegrationTest.java index 8c0440f72887..bcaad916bdc1 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusProcessorClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusProcessorClientIntegrationTest.java @@ -4,8 +4,6 @@ package com.azure.messaging.servicebus; import com.azure.core.amqp.AmqpRetryOptions; -import com.azure.core.amqp.AmqpTransportType; -import com.azure.core.amqp.ProxyOptions; import com.azure.core.util.CoreUtils; import com.azure.core.util.logging.ClientLogger; import com.azure.messaging.servicebus.implementation.MessagingEntityType; @@ -84,7 +82,7 @@ void receiveMessage(MessagingEntityType entityType, boolean isSessionEnabled) th assertNotNull(sessionId, "'sessionId' should have been set."); AmqpRetryOptions amqpRetryOptions = new AmqpRetryOptions() .setTryTimeout(Duration.ofSeconds(2 * lockTimeoutDurationSeconds)); - processor = toClose(getSessionProcessorBuilder(false, entityType, entityIndex, false, amqpRetryOptions) + processor = toClose(getSessionProcessorBuilder(entityType, entityIndex, false, amqpRetryOptions) .maxAutoLockRenewDuration(expectedMaxAutoLockRenew) .disableAutoComplete() .processMessage(context -> processMessage(context, countDownLatch, messageId, lastMessageReceivedTime, lockTimeoutDurationSeconds)) @@ -92,7 +90,7 @@ void receiveMessage(MessagingEntityType entityType, boolean isSessionEnabled) th .buildProcessorClient()); } else { - processor = toClose(getProcessorBuilder(false, entityType, entityIndex, false) + processor = toClose(getProcessorBuilder(entityType, entityIndex, false) .maxAutoLockRenewDuration(expectedMaxAutoLockRenew) .disableAutoComplete() .processMessage(context -> processMessage(context, countDownLatch, messageId, lastMessageReceivedTime, lockTimeoutDurationSeconds)) @@ -117,7 +115,7 @@ void rollingSessionOnIdleTimeout(MessagingEntityType entityType) throws Interrup ServiceBusSenderAsyncClient sender = createSender(entityType, entityIndex, true); ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processorBuilder = - getSessionProcessorBuilder(false, entityType, entityIndex, false, RETRY_OPTIONS) + getSessionProcessorBuilder(entityType, entityIndex, false, RETRY_OPTIONS) .sessionIdleTimeout(sessionIdleTimeout) .disableAutoComplete(); @@ -132,7 +130,7 @@ void rollingSessionOnTryTimeout(MessagingEntityType entityType) throws Interrupt ServiceBusSenderAsyncClient sender = createSender(entityType, entityIndex, true); ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processorBuilder = - getSessionProcessorBuilder(false, entityType, entityIndex, false, + getSessionProcessorBuilder(entityType, entityIndex, false, new AmqpRetryOptions().setTryTimeout(tryTimeout)) .disableAutoComplete(); @@ -198,10 +196,10 @@ private void processError(ServiceBusErrorContext context, CountDownLatch countdo context.getFullyQualifiedNamespace(), context.getEntityPath()); } - private ServiceBusClientBuilder.ServiceBusProcessorClientBuilder getProcessorBuilder(boolean useCredentials, - MessagingEntityType entityType, int entityIndex, boolean sharedConnection) { + private ServiceBusClientBuilder.ServiceBusProcessorClientBuilder getProcessorBuilder(MessagingEntityType entityType, + int entityIndex, boolean sharedConnection) { - ServiceBusClientBuilder builder = getBuilder(useCredentials, sharedConnection); + ServiceBusClientBuilder builder = getBuilder(sharedConnection); switch (entityType) { case QUEUE: final String queueName = getQueueName(entityIndex); @@ -216,14 +214,14 @@ private ServiceBusClientBuilder.ServiceBusProcessorClientBuilder getProcessorBui return builder.processor().topicName(topicName).subscriptionName(subscriptionName); default: - throw LOGGER.logExceptionAsError(new IllegalArgumentException("Unknown entity type: " + entityType)); + throw logger.logExceptionAsError(new IllegalArgumentException("Unknown entity type: " + entityType)); } } - protected ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder getSessionProcessorBuilder(boolean useCredentials, - MessagingEntityType entityType, int entityIndex, boolean sharedConnection, AmqpRetryOptions amqpRetryOptions) { + private ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder getSessionProcessorBuilder(MessagingEntityType entityType, + int entityIndex, boolean sharedConnection, AmqpRetryOptions amqpRetryOptions) { - ServiceBusClientBuilder builder = getBuilder(useCredentials, sharedConnection); + ServiceBusClientBuilder builder = getBuilder(sharedConnection); builder.retryOptions(amqpRetryOptions); switch (entityType) { @@ -242,24 +240,13 @@ protected ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder getSes return builder.sessionProcessor() .topicName(topicName).subscriptionName(subscriptionName); default: - throw LOGGER.logExceptionAsError(new IllegalArgumentException("Unknown entity type: " + entityType)); + throw logger.logExceptionAsError(new IllegalArgumentException("Unknown entity type: " + entityType)); } } - protected ServiceBusClientBuilder getBuilder(boolean useCredentials, boolean sharedConnection) { - return new ServiceBusClientBuilder() - .connectionString(getConnectionString()) - .proxyOptions(ProxyOptions.SYSTEM_DEFAULTS) - .retryOptions(RETRY_OPTIONS) - .transportType(AmqpTransportType.AMQP) - .scheduler(scheduler) - .configuration(v1OrV2(true)); - } - private ServiceBusSenderAsyncClient createSender(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled) { final boolean shareConnection = false; - final boolean useCredentials = false; - return toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection) + return toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, shareConnection) .buildAsyncClient()); } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java index d06803c10c61..4f91b2cf234f 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java @@ -242,11 +242,10 @@ void transactionSendReceiveAndCommit(DispositionStatus dispositionStatus) { void transactionReceiveCompleteCommitMixClient(MessagingEntityType entityType) { // Arrange final boolean shareConnection = true; - final boolean useCredentials = false; final int entityIndex = 0; - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, shareConnection) .buildAsyncClient()); - this.receiver = toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, shareConnection) .buildAsyncClient()); final String messageId = UUID.randomUUID().toString(); @@ -287,10 +286,9 @@ void receiveTwoMessagesAutoComplete(MessagingEntityType entityType, boolean isSe // Arrange final int entityIndex = 0; final boolean shareConnection = false; - final boolean useCredentials = false; final Duration shortWait = Duration.ofSeconds(3); - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, shareConnection) .buildAsyncClient()); final String messageId = UUID.randomUUID().toString(); @@ -300,11 +298,11 @@ void receiveTwoMessagesAutoComplete(MessagingEntityType entityType, boolean isSe // Now create receiver if (isSessionEnabled) { assertNotNull(sessionId, "'sessionId' should have been set."); - this.sessionReceiver = toClose(getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) + this.sessionReceiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) .buildAsyncClient()); this.receiver = toClose(sessionReceiver.acceptSession(sessionId).block()); } else { - this.receiver = toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, shareConnection) .buildAsyncClient()); } @@ -328,9 +326,8 @@ void receiveMessageAutoComplete(MessagingEntityType entityType, boolean isSessio // Arrange final int entityIndex = USE_CASE_AUTO_COMPLETE; final boolean shareConnection = false; - final boolean useCredentials = false; - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, shareConnection) .buildAsyncClient()); final String messageId = CoreUtils.randomUuid().toString(); final ServiceBusMessage message = getMessage(messageId, isSessionEnabled); @@ -342,11 +339,11 @@ void receiveMessageAutoComplete(MessagingEntityType entityType, boolean isSessio // Now create receiver if (isSessionEnabled) { assertNotNull(sessionId, "'sessionId' should have been set."); - this.sessionReceiver = toClose(getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) + this.sessionReceiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) .buildAsyncClient()); this.receiver = toClose(this.sessionReceiver.acceptSession(sessionId).block()); } else { - this.receiver = toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, shareConnection) .buildAsyncClient()); } @@ -702,11 +699,10 @@ void receiveMessageAmqpTypes(MessagingEntityType entityType, boolean isSessionEn // Arrange final int entityIndex = TestUtils.USE_CASE_AMQP_TYPES; final boolean shareConnection = false; - final boolean useCredentials = false; final Duration shortWait = Duration.ofSeconds(3); final Long expectedLongValue = Long.parseLong("6"); - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, shareConnection) .buildAsyncClient()); // Send value Object @@ -728,11 +724,11 @@ void receiveMessageAmqpTypes(MessagingEntityType entityType, boolean isSessionEn // Now create receiver if (isSessionEnabled) { assertNotNull(sessionId, "'sessionId' should have been set."); - this.sessionReceiver = toClose(getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) + this.sessionReceiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) .buildAsyncClient()); this.receiver = toClose(this.sessionReceiver.acceptSession(sessionId).block()); } else { - this.receiver = toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, shareConnection) .buildAsyncClient()); } @@ -1157,7 +1153,7 @@ void receiveFromDeadLetter(MessagingEntityType entityType, boolean isSessionEnab final String queueName = isSessionEnabled ? getSessionQueueName(entityIndex) : getQueueName(entityIndex); assertNotNull(queueName, "'queueName' cannot be null."); - deadLetterReceiver = toClose(getBuilder(false).receiver() + deadLetterReceiver = toClose(getBuilder().receiver() .queueName(queueName) .subQueue(SubQueue.DEAD_LETTER_QUEUE) .buildAsyncClient()); @@ -1168,7 +1164,7 @@ void receiveFromDeadLetter(MessagingEntityType entityType, boolean isSessionEnab assertNotNull(topicName, "'topicName' cannot be null."); assertNotNull(subscriptionName, "'subscriptionName' cannot be null."); - deadLetterReceiver = toClose(getBuilder(false).receiver() + deadLetterReceiver = toClose(getBuilder().receiver() .topicName(topicName) .subscriptionName(subscriptionName) .subQueue(SubQueue.DEAD_LETTER_QUEUE) @@ -1443,7 +1439,7 @@ void autoComplete(MessagingEntityType entityType) { .verifyComplete(); final ServiceBusReceiverAsyncClient autoCompleteReceiver = - toClose(getReceiverBuilder(false, entityType, index, false) + toClose(getReceiverBuilder(entityType, index, false) .buildAsyncClient()); Set sequenceNumbers = new HashSet<>(); @@ -1494,10 +1490,9 @@ private void setReceiver(MessagingEntityType entityType, int entityIndex, boolea private ServiceBusReceiverAsyncClient createReceiver(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled, ClientCreationOptions options) { final boolean shareConnection = false; - final boolean useCredentials = false; if (isSessionEnabled) { assertNotNull(sessionId, "'sessionId' should have been set."); - sessionReceiver = toClose(getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) + sessionReceiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, shareConnection, DEFAULT_RETRY_OPTIONS) .maxAutoLockRenewDuration(options.getMaxAutoLockRenewDuration()) .sessionIdleTimeout(options.getSessionIdleTimeout()) .disableAutoComplete() @@ -1505,7 +1500,7 @@ private ServiceBusReceiverAsyncClient createReceiver(MessagingEntityType entityT return toClose(sessionReceiver.acceptSession(sessionId).block()); } - return toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection) + return toClose(getReceiverBuilder(entityType, entityIndex, shareConnection) .maxAutoLockRenewDuration(options.getMaxAutoLockRenewDuration()) .disableAutoComplete() .buildAsyncClient()); @@ -1513,8 +1508,7 @@ private ServiceBusReceiverAsyncClient createReceiver(MessagingEntityType entityT private void setSender(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled) { final boolean shareConnection = false; - final boolean useCredentials = false; - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, shareConnection) .buildAsyncClient()); } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverClientIntegrationTest.java index 75286f87f34a..3cee975edcae 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverClientIntegrationTest.java @@ -855,18 +855,18 @@ private void setSender(MessagingEntityType entityType, int entityIndex, boolean } private void setSender(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled, boolean sharedConnection) { - this.sender = toClose(getSenderBuilder(false, entityType, entityIndex, isSessionEnabled, sharedConnection).buildClient()); + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionEnabled, sharedConnection).buildClient()); } private void setReceiver(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled, boolean sharedConnection) { if (isSessionEnabled) { assertNotNull(sessionId, "'sessionId' should have been set."); - this.sessionReceiver = toClose(getSessionReceiverBuilder(false, entityType, entityIndex, sharedConnection, null) + this.sessionReceiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, sharedConnection, null) .buildClient()); this.receiver = toClose(this.sessionReceiver.acceptSession(sessionId)); } else { - this.receiver = toClose(getReceiverBuilder(false, entityType, entityIndex, sharedConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, sharedConnection) .buildClient()); } } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientIntegrationTest.java index d62e825e3a4d..1170b459b85d 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientIntegrationTest.java @@ -37,7 +37,7 @@ @Tag("integration") class ServiceBusSenderAsyncClientIntegrationTest extends IntegrationTestBase { private static final ClientLogger LOGGER = new ClientLogger(ServiceBusSenderAsyncClientIntegrationTest.class); - + private ServiceBusSenderAsyncClient sender; private ServiceBusReceiverAsyncClient receiver; private final AtomicInteger messagesPending = new AtomicInteger(); @@ -93,7 +93,7 @@ static Stream transactionMessageSendAndCompleteTransaction() { @ParameterizedTest void nonSessionQueueSendMessage(MessagingEntityType entityType) { // Arrange - setSenderAndReceiver(entityType, 0, false); + setSenderAndReceiver(entityType, 0); final String messageId = UUID.randomUUID().toString(); final ServiceBusMessage message = TestUtils.getServiceBusMessage(CONTENTS_BYTES, messageId); @@ -111,7 +111,7 @@ void nonSessionQueueSendMessage(MessagingEntityType entityType) { @ParameterizedTest void nonSessionEntitySendMessageList(MessagingEntityType entityType) { // Arrange - setSenderAndReceiver(entityType, TestUtils.USE_CASE_DEFAULT, false); + setSenderAndReceiver(entityType, TestUtils.USE_CASE_DEFAULT); int count = 4; final List messages = TestUtils.getServiceBusMessages(count, UUID.randomUUID().toString(), CONTENTS_BYTES); @@ -130,7 +130,7 @@ void nonSessionEntitySendMessageList(MessagingEntityType entityType) { @ParameterizedTest void nonSessionMessageBatch(MessagingEntityType entityType) { // Arrange - setSenderAndReceiver(entityType, TestUtils.USE_CASE_DEFAULT, false); + setSenderAndReceiver(entityType, TestUtils.USE_CASE_DEFAULT); final String messageId = UUID.randomUUID().toString(); final CreateMessageBatchOptions options = new CreateMessageBatchOptions().setMaximumSizeInBytes(1024); @@ -156,7 +156,6 @@ void nonSessionMessageBatch(MessagingEntityType entityType) { @Test void viaQueueMessageSendTest() { // Arrange - final boolean useCredentials = false; final Duration shortTimeout = Duration.ofSeconds(15); final int viaIntermediateEntity = TestUtils.USE_CASE_TXN_1; final int destinationEntity = TestUtils.USE_CASE_TXN_2; @@ -169,13 +168,13 @@ void viaQueueMessageSendTest() { final List messages = TestUtils.getServiceBusMessages(total, messageId, CONTENTS_BYTES); final String viaQueueName = getQueueName(viaIntermediateEntity); - setSenderAndReceiver(entityType, viaIntermediateEntity, useCredentials); + setSenderAndReceiver(entityType, viaIntermediateEntity); - final ServiceBusSenderAsyncClient destination1ViaSender = toClose(getSenderBuilder(useCredentials, entityType, + final ServiceBusSenderAsyncClient destination1ViaSender = toClose(getSenderBuilder(entityType, destinationEntity, false, shareConnection) //.viaQueueName(viaQueueName) .buildAsyncClient()); - final ServiceBusReceiverAsyncClient destination1Receiver = toClose(getReceiverBuilder(useCredentials, entityType, + final ServiceBusReceiverAsyncClient destination1Receiver = toClose(getReceiverBuilder(entityType, destinationEntity, shareConnection) .receiveMode(ServiceBusReceiveMode.RECEIVE_AND_DELETE) .disableAutoComplete() @@ -245,7 +244,6 @@ void viaQueueMessageSendTest() { @Test void viaTopicMessageSendTest() { // Arrange - final boolean useCredentials = false; final Duration shortTimeout = Duration.ofSeconds(15); final int viaIntermediateEntity = TestUtils.USE_CASE_SEND_VIA_TOPIC_1; final int destinationEntity = TestUtils.USE_CASE_SEND_VIA_TOPIC_2; @@ -258,16 +256,16 @@ void viaTopicMessageSendTest() { final List messages = TestUtils.getServiceBusMessages(total, messageId, CONTENTS_BYTES); final String viaTopicName = getTopicName(viaIntermediateEntity); - setSenderAndReceiver(entityType, viaIntermediateEntity, useCredentials); + setSenderAndReceiver(entityType, viaIntermediateEntity); final ServiceBusReceiverAsyncClient intermediateReceiver = receiver; final ServiceBusSenderAsyncClient intermediateSender = sender; - final ServiceBusSenderAsyncClient destination1ViaSender = toClose(getSenderBuilder(useCredentials, entityType, + final ServiceBusSenderAsyncClient destination1ViaSender = toClose(getSenderBuilder(entityType, destinationEntity, false, shareConnection) //.viaTopicName(viaTopicName) .buildAsyncClient()); - final ServiceBusReceiverAsyncClient destination1Receiver = toClose(getReceiverBuilder(useCredentials, entityType, + final ServiceBusReceiverAsyncClient destination1Receiver = toClose(getReceiverBuilder(entityType, destinationEntity, shareConnection) .receiveMode(ServiceBusReceiveMode.RECEIVE_AND_DELETE) .disableAutoComplete() @@ -334,7 +332,7 @@ void viaTopicMessageSendTest() { void transactionMessageSendAndCompleteTransaction(MessagingEntityType entityType, boolean isCommit) { // Arrange Duration shortTimeout = Duration.ofSeconds(15); - setSenderAndReceiver(entityType, TestUtils.USE_CASE_SEND_READ_BACK_MESSAGES, false); + setSenderAndReceiver(entityType, TestUtils.USE_CASE_SEND_READ_BACK_MESSAGES); final boolean isSessionEnabled = false; final String messageId = UUID.randomUUID().toString(); final int total = 3; @@ -393,7 +391,7 @@ void transactionMessageSendAndCompleteTransaction(MessagingEntityType entityType @ParameterizedTest void sendWithCredentials(MessagingEntityType entityType) { // Arrange - setSenderAndReceiver(entityType, 0, true); + setSenderAndReceiver(entityType, 0); final String messageId = UUID.randomUUID().toString(); final List messages = TestUtils.getServiceBusMessages(5, messageId, CONTENTS_BYTES); @@ -418,7 +416,7 @@ void transactionScheduleAndCommitTest(MessagingEntityType entityType) { // Arrange boolean isSessionEnabled = false; - setSenderAndReceiver(entityType, 0, isSessionEnabled); + setSenderAndReceiver(entityType, 0); final Duration scheduleDuration = Duration.ofSeconds(3); final String messageId = UUID.randomUUID().toString(); final ServiceBusMessage message = getMessage(messageId, isSessionEnabled); @@ -464,7 +462,7 @@ void transactionScheduleMessagesTest(MessagingEntityType entityType) { final int total = 2; final Duration shortWait = Duration.ofSeconds(3); - setSenderAndReceiver(entityType, TestUtils.USE_CASE_SCHEDULE_MESSAGES, isSessionEnabled); + setSenderAndReceiver(entityType, TestUtils.USE_CASE_SCHEDULE_MESSAGES); final Duration scheduleDuration = Duration.ofSeconds(5); final String messageId = UUID.randomUUID().toString(); @@ -519,7 +517,7 @@ void cancelScheduledMessagesTest(MessagingEntityType entityType) { final boolean isSessionEnabled = false; final Duration shortWaitTime = Duration.ofSeconds(5); final int total = 2; - setSenderAndReceiver(entityType, TestUtils.USE_CASE_CANCEL_MESSAGES, isSessionEnabled); + setSenderAndReceiver(entityType, TestUtils.USE_CASE_CANCEL_MESSAGES); final Duration scheduleDuration = Duration.ofSeconds(15); final String messageId = UUID.randomUUID().toString(); final List messages = new ArrayList<>(); @@ -544,13 +542,13 @@ void cancelScheduledMessagesTest(MessagingEntityType entityType) { /** * Sets the sender and receiver. If session is enabled, then a single-named session receiver is created. */ - private void setSenderAndReceiver(MessagingEntityType entityType, int entityIndex, boolean useCredentials) { + private void setSenderAndReceiver(MessagingEntityType entityType, int entityIndex) { final boolean isSessionAware = false; final boolean sharedConnection = true; - this.sender = toClose(getSenderBuilder(useCredentials, entityType, entityIndex, isSessionAware, sharedConnection) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, isSessionAware, sharedConnection) .buildAsyncClient()); - this.receiver = toClose(getReceiverBuilder(useCredentials, entityType, entityIndex, sharedConnection) + this.receiver = toClose(getReceiverBuilder(entityType, entityIndex, sharedConnection) .receiveMode(ServiceBusReceiveMode.RECEIVE_AND_DELETE) .disableAutoComplete() .buildAsyncClient()); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerIntegrationTest.java index c2edf791e44d..a62a2cfa41c8 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerIntegrationTest.java @@ -106,8 +106,7 @@ void rollingSessionOnIdleTimeout(MessagingEntityType entityType) throws Interrup final Duration sessionIdleTimeout = Duration.ofSeconds(3); setSender(entityType, entityIndex); - this.receiver = toClose(getSessionReceiverBuilder(false, - entityType, entityIndex, false, DEFAULT_RETRY_OPTIONS) + this.receiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, false, DEFAULT_RETRY_OPTIONS) .disableAutoComplete() .maxConcurrentSessions(1) .sessionIdleTimeout(sessionIdleTimeout) @@ -124,8 +123,7 @@ void rollingSessionOnTryTimeout(MessagingEntityType entityType) throws Interrupt final AmqpRetryOptions retryOptions = new AmqpRetryOptions().setTryTimeout(tryTimeout); setSender(entityType, entityIndex); - this.receiver = toClose(getSessionReceiverBuilder(false, - entityType, entityIndex, false, retryOptions) + this.receiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, false, retryOptions) .disableAutoComplete() .maxConcurrentSessions(1) .buildAsyncClientForProcessor()); @@ -165,8 +163,7 @@ void noRollingSessionWhenNoConcurrentSessions(MessagingEntityType entityType) { ServiceBusMessage message1 = getServiceBusMessage(contents, randomPrefix + "1").setSessionId(randomPrefix + "1"); AmqpRetryOptions retryOptions = new AmqpRetryOptions().setTryTimeout(tryTimeout); - this.receiver = toClose(getSessionReceiverBuilder(false, - entityType, entityIndex, false, retryOptions) + this.receiver = toClose(getSessionReceiverBuilder(entityType, entityIndex, false, retryOptions) .disableAutoComplete() .sessionIdleTimeout(sessionIdleTimeout) .buildAsyncClientForProcessor()); @@ -191,14 +188,14 @@ void noRollingSessionWhenNoConcurrentSessions(MessagingEntityType entityType) { * Sets the sender and receiver. If session is enabled, then a single-named session receiver is created. */ private void setSender(MessagingEntityType entityType, int entityIndex) { - this.sender = toClose(getSenderBuilder(false, entityType, entityIndex, true, false) + this.sender = toClose(getSenderBuilder(entityType, entityIndex, true, false) .buildAsyncClient()); } private void setReceiver(MessagingEntityType entityType, int entityIndex, - Function onBuild) { - ServiceBusSessionReceiverClientBuilder sessionBuilder = getSessionReceiverBuilder(false, - entityType, entityIndex, false, DEFAULT_RETRY_OPTIONS).disableAutoComplete(); + Function onBuild) { + ServiceBusSessionReceiverClientBuilder sessionBuilder = getSessionReceiverBuilder(entityType, entityIndex, + false, DEFAULT_RETRY_OPTIONS).disableAutoComplete(); this.sessionReceiver = toClose(onBuild.apply(sessionBuilder).buildAsyncClient()); this.receiver = toClose(this.sessionReceiver.acceptSession(sessionId).block()); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TestUtils.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TestUtils.java index 0bb44625ed38..85efebf168c5 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TestUtils.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TestUtils.java @@ -5,11 +5,14 @@ import com.azure.core.amqp.exception.AmqpResponseCode; import com.azure.core.amqp.implementation.ConnectionStringProperties; +import com.azure.core.credential.TokenCredential; import com.azure.core.util.BinaryData; import com.azure.core.util.Configuration; import com.azure.core.util.CoreUtils; import com.azure.core.util.logging.ClientLogger; import com.azure.core.util.logging.LogLevel; +import com.azure.identity.AzurePipelinesCredential; +import com.azure.identity.AzurePipelinesCredentialBuilder; import com.azure.messaging.servicebus.administration.models.AccessRights; import com.azure.messaging.servicebus.administration.models.AuthorizationRule; import org.apache.qpid.proton.Proton; @@ -19,6 +22,8 @@ import org.apache.qpid.proton.amqp.messaging.Data; import org.apache.qpid.proton.amqp.messaging.MessageAnnotations; import org.apache.qpid.proton.message.Message; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; import javax.crypto.Mac; import javax.crypto.spec.SecretKeySpec; @@ -36,6 +41,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -46,6 +52,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; public class TestUtils { private static final ClientLogger LOGGER = new ClientLogger(TestUtils.class); @@ -158,10 +165,16 @@ public static String getConnectionString(boolean withSas) { /** * Gets the fully qualified domain name for the service bus resource. * + * @param useFallbackValue If {@code true}, a redacted dummy name is returned if the fully qualified domain name environment variable is not set. + * * @return The fully qualified domain name for the service bus resource. */ - public static String getFullyQualifiedDomainName() { - return getPropertyValue("AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME", "REDACTED.servicebus.windows.net"); + public static String getFullyQualifiedDomainName(boolean useFallbackValue) { + if (useFallbackValue) { + return getPropertyValue("AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME", "REDACTED.servicebus.windows.net"); + } else { + return getPropertyValue("AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME"); + } } public static String getEndpoint() { @@ -222,6 +235,51 @@ public static String getSessionSubscriptionBaseName() { return getPropertyValue("AZURE_SERVICEBUS_SESSION_SUBSCRIPTION_NAME", "subscription-session"); } + /** + * Obtain a {@link com.azure.identity.AzurePipelinesCredentialBuilder} when running in Azure pipelines that is + * configured with service connections federated identity. + * + * @return A {@link com.azure.identity.AzurePipelinesCredentialBuilder} when running in Azure pipelines that is + * configured with service connections federated identity, {@code null} otherwise. + */ + public static AzurePipelinesCredentialBuilder getPipelineCredentialBuilder() { + final String serviceConnectionId = getPropertyValue("AZURESUBSCRIPTION_SERVICE_CONNECTION_ID"); + final String clientId = getPropertyValue("AZURESUBSCRIPTION_CLIENT_ID"); + final String tenantId = getPropertyValue("AZURESUBSCRIPTION_TENANT_ID"); + final String systemAccessToken = getPropertyValue("SYSTEM_ACCESSTOKEN"); + if (CoreUtils.isNullOrEmpty(serviceConnectionId) || CoreUtils.isNullOrEmpty(clientId) + || CoreUtils.isNullOrEmpty(tenantId) || CoreUtils.isNullOrEmpty(systemAccessToken)) { + return null; + } + return new AzurePipelinesCredentialBuilder().systemAccessToken(systemAccessToken) + .clientId(clientId) + .tenantId(tenantId) + .serviceConnectionId(serviceConnectionId); + } + + /** + * Obtain the Azure Pipelines credential if running in Azure Pipelines configured with service connections federated identity. + * + * @return the Azure Pipelines credential. + * @throws org.opentest4j.TestAbortedException if the test is not running in Azure Pipelines configured with service connections federated identity. + */ + public static TokenCredential getPipelineCredential(AtomicReference credentialCached) { + return credentialCached.updateAndGet(cached -> { + if (cached != null) { + return cached; + } + final AzurePipelinesCredentialBuilder builder = TestUtils.getPipelineCredentialBuilder(); + if (builder == null) { + // Throws org.opentest4j.TestAbortedException exception. + assumeTrue(false, "Test required to run on Azure Pipelines that is configured with service connections federated identity."); + return null; + } + final AzurePipelinesCredential pipelinesCredential = builder.build(); + return request -> Mono.defer(() -> pipelinesCredential.getToken(request)) + .subscribeOn(Schedulers.boundedElastic()); + }); + } + /** * Gets the name of an entity based on its base name. * diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java index 6bc5730921d2..254d15ff68db 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java @@ -68,7 +68,7 @@ public TracingIntegrationTests() { @Override protected void beforeTest() { - spanProcessor = new TestSpanProcessor(getFullyQualifiedDomainName(), getQueueName(0)); + spanProcessor = new TestSpanProcessor(TestUtils.getFullyQualifiedDomainName(true), getQueueName(0)); OpenTelemetryTracingOptions tracingOptions = new OpenTelemetryTracingOptions() .setOpenTelemetry(OpenTelemetrySdk.builder() .setTracerProvider( @@ -78,15 +78,15 @@ protected void beforeTest() { .build()); clientOptions = new ClientOptions().setTracingOptions(tracingOptions); - sender = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder senderBuilder = getAuthenticatedBuilder(); + sender = toClose(senderBuilder .clientOptions(clientOptions) .sender() .queueName(getQueueName(0)) .buildAsyncClient()); - receiver = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder receiverBuilder = getAuthenticatedBuilder(); + receiver = toClose(receiverBuilder .clientOptions(clientOptions) .receiver() .maxAutoLockRenewDuration(Duration.ZERO) @@ -94,8 +94,8 @@ protected void beforeTest() { .disableAutoComplete() .buildAsyncClient()); - receiverSync = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder receiverSyncBuilder = getAuthenticatedBuilder(); + receiverSync = toClose(receiverSyncBuilder .clientOptions(clientOptions) .receiver() .queueName(getQueueName(0)) @@ -189,16 +189,16 @@ public void receiveAndRenewSessionLockWithDuration() throws InterruptedException ServiceBusMessage message = new ServiceBusMessage(CONTENTS_BYTES) .setSessionId(sessionId); - OpenTelemetry otel = configureOTel(getFullyQualifiedDomainName(), getSessionQueueName(0)); - sender = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + OpenTelemetry otel = configureOTel(TestUtils.getFullyQualifiedDomainName(true), getSessionQueueName(0)); + final ServiceBusClientBuilder senderBuilder = getAuthenticatedBuilder(); + sender = toClose(senderBuilder .clientOptions(new ClientOptions().setTracingOptions(new OpenTelemetryTracingOptions().setOpenTelemetry(otel))) .sender() .queueName(getSessionQueueName(0)) .buildAsyncClient()); - ServiceBusSessionReceiverAsyncClient sessionReceiver = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder receiverBuilder = getAuthenticatedBuilder(); + ServiceBusSessionReceiverAsyncClient sessionReceiver = toClose(receiverBuilder .clientOptions(new ClientOptions().setTracingOptions(new OpenTelemetryTracingOptions().setOpenTelemetry(otel))) .sessionReceiver() .disableAutoComplete() @@ -287,8 +287,8 @@ public void sendAndReceiveParallelNoAutoCompleteAndLockRenewal() throws Interrup CountDownLatch processedFound = new CountDownLatch(messageCount); spanProcessor.notifyIfCondition(processedFound, span -> span.getName().equals("ServiceBus.process")); - ServiceBusReceiverAsyncClient receiverWithLockRenewal = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder receiverBuilder = getAuthenticatedBuilder(); + ServiceBusReceiverAsyncClient receiverWithLockRenewal = toClose(receiverBuilder .clientOptions(clientOptions) .receiver() .queueName(getQueueName(0)) @@ -348,8 +348,8 @@ public void sendAndReceiveParallelAutoComplete() throws InterruptedException { CountDownLatch processedFound = new CountDownLatch(messageCount); spanProcessor.notifyIfCondition(processedFound, span -> span.getName().equals("ServiceBus.process")); - ServiceBusReceiverAsyncClient receiverAutoComplete = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder receiverBuilder = getAuthenticatedBuilder(); + ServiceBusReceiverAsyncClient receiverAutoComplete = toClose(receiverBuilder .clientOptions(clientOptions) .receiver() .queueName(getQueueName(0)) @@ -548,8 +548,8 @@ public void sendAndProcessNoAutoComplete() throws InterruptedException { AtomicReference currentInProcess = new AtomicReference<>(); AtomicReference receivedMessage = new AtomicReference<>(); - processor = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder processorBuilder = getAuthenticatedBuilder(); + processor = toClose(processorBuilder .clientOptions(clientOptions) .processor() .queueName(getQueueName(0)) @@ -598,10 +598,10 @@ public void sendAndProcessSessionNoAutoComplete() throws InterruptedException { String sessionId = UUID.randomUUID().toString(); ServiceBusMessage message = new ServiceBusMessage(CONTENTS_BYTES).setSessionId(sessionId); - OpenTelemetry otel = configureOTel(getFullyQualifiedDomainName(), sessionQueueName); + OpenTelemetry otel = configureOTel(TestUtils.getFullyQualifiedDomainName(true), sessionQueueName); clientOptions = new ClientOptions().setTracingOptions(new OpenTelemetryTracingOptions().setOpenTelemetry(otel)); - sender = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder senderBuilder = getAuthenticatedBuilder(); + sender = toClose(senderBuilder .clientOptions(clientOptions) .sender() .queueName(sessionQueueName) @@ -614,8 +614,8 @@ public void sendAndProcessSessionNoAutoComplete() throws InterruptedException { AtomicReference currentInProcess = new AtomicReference<>(); AtomicReference receivedMessage = new AtomicReference<>(); - processor = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder processorBuilder = getAuthenticatedBuilder(); + processor = toClose(processorBuilder .clientOptions(clientOptions) .sessionProcessor() .maxConcurrentSessions(1) @@ -668,8 +668,8 @@ public void sendAndProcessParallel() throws InterruptedException { CountDownLatch processedFound = new CountDownLatch(messageCount * 2); spanProcessor.notifyIfCondition(processedFound, span -> span.getName().equals("ServiceBus.process") || span.getName().equals("ServiceBus.complete")); - processor = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder processorBuilder = getAuthenticatedBuilder(); + processor = toClose(processorBuilder .clientOptions(clientOptions) .processor() .queueName(getQueueName(0)) @@ -710,8 +710,8 @@ public void sendAndProcessParallelNoAutoComplete() throws InterruptedException { CountDownLatch processedFound = new CountDownLatch(messageCount); spanProcessor.notifyIfCondition(processedFound, span -> span.getName().equals("ServiceBus.process")); - processor = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder processorBuilder = getAuthenticatedBuilder(); + processor = toClose(processorBuilder .clientOptions(clientOptions) .processor() .queueName(getQueueName(0)) @@ -753,8 +753,8 @@ public void sendProcessAndFail() throws InterruptedException { "ServiceBus.process".equals(span.getName()) && span.getParentSpanContext().getSpanId().equals(message1SpanId)); AtomicReference receivedMessage = new AtomicReference<>(); - processor = toClose(new ServiceBusClientBuilder() - .connectionString(getConnectionString()) + final ServiceBusClientBuilder processorBuilder = getAuthenticatedBuilder(); + processor = toClose(processorBuilder .clientOptions(clientOptions) .processor() .queueName(getQueueName(0)) diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationAsyncClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationAsyncClientIntegrationTest.java index a64d6ac584d1..9a5b6be7a886 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationAsyncClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationAsyncClientIntegrationTest.java @@ -4,7 +4,6 @@ package com.azure.messaging.servicebus.administration; import com.azure.core.credential.AccessToken; -import com.azure.core.credential.AzureSasCredential; import com.azure.core.credential.TokenCredential; import com.azure.core.exception.ClientAuthenticationException; import com.azure.core.exception.ResourceExistsException; @@ -14,11 +13,14 @@ import com.azure.core.http.netty.NettyAsyncHttpClientBuilder; import com.azure.core.http.policy.HttpLogDetailLevel; import com.azure.core.http.policy.HttpLogOptions; +import com.azure.core.test.InterceptorManager; import com.azure.core.test.TestProxyTestBase; import com.azure.core.test.models.CustomMatcher; import com.azure.core.test.models.TestProxyRequestMatcher; import com.azure.core.test.models.TestProxySanitizer; import com.azure.core.test.models.TestProxySanitizerType; +import com.azure.core.test.utils.MockTokenCredential; +import com.azure.core.util.CoreUtils; import com.azure.identity.DefaultAzureCredentialBuilder; import com.azure.messaging.servicebus.TestUtils; import com.azure.messaging.servicebus.administration.models.AccessRights; @@ -28,7 +30,6 @@ import com.azure.messaging.servicebus.administration.models.CreateTopicOptions; import com.azure.messaging.servicebus.administration.models.EmptyRuleAction; import com.azure.messaging.servicebus.administration.models.FalseRuleFilter; -import com.azure.messaging.servicebus.administration.models.NamespaceProperties; import com.azure.messaging.servicebus.administration.models.NamespaceType; import com.azure.messaging.servicebus.administration.models.QueueRuntimeProperties; import com.azure.messaging.servicebus.administration.models.RuleProperties; @@ -39,8 +40,8 @@ import com.azure.messaging.servicebus.administration.models.TopicProperties; import com.azure.messaging.servicebus.administration.models.TopicRuntimeProperties; import com.azure.messaging.servicebus.administration.models.TrueRuleFilter; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Tag; -import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -53,12 +54,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.regex.Matcher; -import java.util.regex.Pattern; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Stream; import static com.azure.messaging.servicebus.TestUtils.assertAuthorizationRules; -import static com.azure.messaging.servicebus.TestUtils.getConnectionString; import static com.azure.messaging.servicebus.TestUtils.getEntityName; import static com.azure.messaging.servicebus.TestUtils.getSessionSubscriptionBaseName; import static com.azure.messaging.servicebus.TestUtils.getSubscriptionBaseName; @@ -100,21 +99,27 @@ class ServiceBusAdministrationAsyncClientIntegrationTest extends TestProxyTestBa TEST_PROXY_REQUEST_MATCHERS = Collections.singletonList(customMatcher); } + private final AtomicReference credentialCached = new AtomicReference<>(); + public static Stream createHttpClients() { return Stream.of(Arguments.of(new NettyAsyncHttpClientBuilder().build())); } /** - * Test to connect to the service bus with an azure identity TokenCredential. - * com.azure.identity.ClientSecretCredential is used in this test. - * ServiceBusSharedKeyCredential doesn't need a specific test method because other tests below - * use connection string, which is converted to a ServiceBusSharedKeyCredential internally. + * Test to connect to the service bus using com.azure.identity.ClientSecretCredential. + *

+ * This is a potential test eligible to run in auxiliary tenant with secret auth. The CI Pipeline cannot be enabled + * for both Federated Managed Identity auth and Secret auth, the ARM deployment will fail if an attempt is made to + * enable two auth types - Exception calling "Invoke" with "0" argument(s): "Cannot process command because of one or + * more missing mandatory parameters: testApplicationSecret.". Hence, this test is disabled until auxiliary tenant is + * available. + *

*/ @ParameterizedTest @MethodSource("createHttpClients") - void azureIdentityCredentials(HttpClient httpClient) { - - final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(); + @Disabled("The CI Pipeline cannot be enabled for both Federated Managed Identity auth and Secret auth") + void azureClientSecretCredential(HttpClient httpClient) { + final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(true); final TokenCredential tokenCredential; if (interceptorManager.isPlaybackMode()) { tokenCredential = request -> Mono.fromCallable(() -> @@ -130,8 +135,7 @@ void azureIdentityCredentials(HttpClient httpClient) { } else if (interceptorManager.isLiveMode()) { builder.httpClient(httpClient); } else { - builder.httpClient(httpClient) - .addPolicy(interceptorManager.getRecordPolicy()); + builder.httpClient(httpClient).addPolicy(interceptorManager.getRecordPolicy()); } final ServiceBusAdministrationAsyncClient client = builder @@ -141,9 +145,8 @@ void azureIdentityCredentials(HttpClient httpClient) { StepVerifier.create(client.getNamespaceProperties()) .assertNext(properties -> { assertNotNull(properties); - if (!interceptorManager.isPlaybackMode()) { - final String[] split = TestUtils.getFullyQualifiedDomainName().split("\\.", 2); + final String[] split = TestUtils.getFullyQualifiedDomainName(true).split("\\.", 2); assertEquals(split[0], properties.getName()); } }) @@ -151,40 +154,6 @@ void azureIdentityCredentials(HttpClient httpClient) { .verify(DEFAULT_TIMEOUT); } - /** - * Test to connect to the service bus with an azure sas credential. - * ServiceBusSharedKeyCredential doesn't need a specific test method because other tests below - * use connection string, which is converted to a ServiceBusSharedKeyCredential internally. - */ - @Test - void azureSasCredentialsTest() { - // Arrange - assumeTrue(interceptorManager.isLiveMode(), "Azure Identity test is for live test only"); - final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(); - - assumeTrue(fullyQualifiedDomainName != null && !fullyQualifiedDomainName.isEmpty(), - "AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME variable needs to be set when using credentials."); - - String connectionString = getConnectionString(true); - Pattern sasPattern = Pattern.compile("SharedAccessSignature=(.*);?", Pattern.CASE_INSENSITIVE); - Matcher matcher = sasPattern.matcher(connectionString); - assertTrue(matcher.find(), "Couldn't find SAS from connection string"); - - ServiceBusAdministrationAsyncClient client = new ServiceBusAdministrationClientBuilder() - .endpoint("https://" + fullyQualifiedDomainName) - .credential(new AzureSasCredential(matcher.group(1))) - .buildAsyncClient(); - - // Act & Assert - StepVerifier.create(client.getNamespacePropertiesWithResponse()) - .assertNext(response -> { - final NamespaceProperties np = response.getValue(); - assertNotNull(np.getName()); - }) - .expectComplete() - .verify(DEFAULT_TIMEOUT); - } - //region Create tests @ParameterizedTest @@ -734,7 +703,7 @@ void getNamespace(HttpClient httpClient) { .assertNext(properties -> { assertEquals(NamespaceType.MESSAGING, properties.getNamespaceType()); if (!interceptorManager.isPlaybackMode()) { - final String[] split = TestUtils.getFullyQualifiedDomainName().split("\\.", 2); + final String[] split = TestUtils.getFullyQualifiedDomainName(true).split("\\.", 2); assertEquals(split[0], properties.getName()); } }) @@ -1212,19 +1181,36 @@ void updateRuleResponse(HttpClient httpClient) { } private ServiceBusAdministrationAsyncClient createClient(HttpClient httpClient) { - final String connectionString = interceptorManager.isPlaybackMode() - ? "Endpoint=sb://foo.servicebus.windows.net;SharedAccessKeyName=dummyKey;SharedAccessKey=dummyAccessKey" - : TestUtils.getConnectionString(false); - final ServiceBusAdministrationClientBuilder builder = new ServiceBusAdministrationClientBuilder() - .httpLogOptions(new HttpLogOptions().setLogLevel(HttpLogDetailLevel.BODY_AND_HEADERS)) - .connectionString(connectionString); + .httpLogOptions(new HttpLogOptions().setLogLevel(HttpLogDetailLevel.BODY_AND_HEADERS)); + configure(builder, httpClient, interceptorManager, credentialCached); + return builder.buildAsyncClient(); + } + static void configure(ServiceBusAdministrationClientBuilder builder, + HttpClient httpClient, InterceptorManager interceptorManager, AtomicReference credentialCached) { if (interceptorManager.isPlaybackMode()) { + builder.credential(TestUtils.getFullyQualifiedDomainName(true), new MockTokenCredential()); builder.httpClient(interceptorManager.getPlaybackClient()); } else if (interceptorManager.isLiveMode()) { - builder.httpClient(httpClient); + final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(false); + assumeTrue(!CoreUtils.isNullOrEmpty(fullyQualifiedDomainName), "FullyQualifiedDomainName is not set."); + final TokenCredential credential = TestUtils.getPipelineCredential(credentialCached); + builder.credential(fullyQualifiedDomainName, credential); + if (httpClient != null) { + builder.httpClient(httpClient); + } } else { + // Record Mode. + final String connectionString = TestUtils.getConnectionString(false); + if (CoreUtils.isNullOrEmpty(connectionString)) { + final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(false); + assumeTrue(!CoreUtils.isNullOrEmpty(fullyQualifiedDomainName), "FullyQualifiedDomainName is not set."); + final TokenCredential credential = new DefaultAzureCredentialBuilder().build(); + builder.credential(fullyQualifiedDomainName, credential); + } else { + builder.connectionString(connectionString); + } builder.httpClient(httpClient).addPolicy(interceptorManager.getRecordPolicy()); } @@ -1232,7 +1218,5 @@ private ServiceBusAdministrationAsyncClient createClient(HttpClient httpClient) interceptorManager.addSanitizers(TEST_PROXY_SANITIZERS); interceptorManager.addMatchers(TEST_PROXY_REQUEST_MATCHERS); } - - return builder.buildAsyncClient(); } } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationClientIntegrationTest.java index 021dea3c418a..6355814842a8 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/administration/ServiceBusAdministrationClientIntegrationTest.java @@ -3,21 +3,16 @@ package com.azure.messaging.servicebus.administration; -import com.azure.core.credential.AccessToken; -import com.azure.core.credential.AzureSasCredential; import com.azure.core.credential.TokenCredential; import com.azure.core.exception.ClientAuthenticationException; import com.azure.core.exception.ResourceExistsException; import com.azure.core.exception.ResourceNotFoundException; -import com.azure.core.http.policy.FixedDelayOptions; import com.azure.core.http.policy.HttpLogDetailLevel; import com.azure.core.http.policy.HttpLogOptions; -import com.azure.core.http.policy.RetryOptions; import com.azure.core.http.rest.PagedIterable; import com.azure.core.http.rest.Response; import com.azure.core.test.TestProxyTestBase; import com.azure.core.util.Context; -import com.azure.identity.DefaultAzureCredentialBuilder; import com.azure.messaging.servicebus.TestUtils; import com.azure.messaging.servicebus.administration.models.AccessRights; import com.azure.messaging.servicebus.administration.models.CreateQueueOptions; @@ -44,7 +39,6 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.parallel.Execution; import org.junit.jupiter.api.parallel.ExecutionMode; -import reactor.core.publisher.Mono; import java.time.Clock; import java.time.Duration; @@ -52,24 +46,21 @@ import java.util.Collections; import java.util.List; import java.util.Optional; -import java.util.regex.Matcher; -import java.util.regex.Pattern; +import java.util.concurrent.atomic.AtomicReference; import static com.azure.messaging.servicebus.TestUtils.assertAuthorizationRules; -import static com.azure.messaging.servicebus.TestUtils.getConnectionString; import static com.azure.messaging.servicebus.TestUtils.getEntityName; -import static com.azure.messaging.servicebus.TestUtils.getFullyQualifiedDomainName; import static com.azure.messaging.servicebus.TestUtils.getQueueBaseName; import static com.azure.messaging.servicebus.TestUtils.getRuleBaseName; import static com.azure.messaging.servicebus.TestUtils.getSubscriptionBaseName; import static com.azure.messaging.servicebus.TestUtils.getTopicBaseName; +import static com.azure.messaging.servicebus.administration.ServiceBusAdministrationAsyncClientIntegrationTest.configure; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assumptions.assumeTrue; /** * Tests {@link ServiceBusAdministrationClient}. @@ -78,73 +69,7 @@ @Execution(ExecutionMode.SAME_THREAD) public class ServiceBusAdministrationClientIntegrationTest extends TestProxyTestBase { private static final Duration TIMEOUT = Duration.ofSeconds(20); - - /** - * Test to connect to the service bus with an azure identity TokenCredential. - * com.azure.identity.ClientSecretCredential is used in this test. - * ServiceBusSharedKeyCredential doesn't need a specific test method because other tests below - * use connection string, which is converted to a ServiceBusSharedKeyCredential internally. - */ - @Test - void azureIdentityCredentials() { - // Arrange - final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(); - final TokenCredential tokenCredential; - if (interceptorManager.isPlaybackMode()) { - tokenCredential = request -> Mono.fromCallable(() -> - new AccessToken("foo-bar", OffsetDateTime.now().plus(Duration.ofMinutes(5)))); - } else { - tokenCredential = new DefaultAzureCredentialBuilder().build(); - } - - final ServiceBusAdministrationClientBuilder builder = new ServiceBusAdministrationClientBuilder(); - - if (interceptorManager.isPlaybackMode()) { - builder.httpClient(interceptorManager.getPlaybackClient()); - } else if (interceptorManager.isRecordMode()) { - builder.addPolicy(interceptorManager.getRecordPolicy()); - } - - final ServiceBusAdministrationClient client = builder - .credential(fullyQualifiedDomainName, tokenCredential) - .buildClient(); - - // Act - NamespaceProperties properties = client.getNamespaceProperties(); - - // Assert - assertNotNull(properties); - - if (!interceptorManager.isPlaybackMode()) { - final String[] split = TestUtils.getFullyQualifiedDomainName().split("\\.", 2); - assertEquals(split[0], properties.getName()); - } - } - - /** - * Test to connect to the service bus with an azure sas credential. - * ServiceBusSharedKeyCredential doesn't need a specific test method because other tests below - * use connection string, which is converted to a ServiceBusSharedKeyCredential internally. - */ - @Test - void azureSasCredentialsTest() { - assumeTrue(interceptorManager.isLiveMode(), "Azure Identity test is for live test only"); - final String fullyQualifiedDomainName = TestUtils.getFullyQualifiedDomainName(); - - assumeTrue(fullyQualifiedDomainName != null && !fullyQualifiedDomainName.isEmpty(), - "AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME variable needs to be set when using credentials."); - - String connectionString = getConnectionString(true); - Pattern sasPattern = Pattern.compile("SharedAccessSignature=(.*);?", Pattern.CASE_INSENSITIVE); - Matcher matcher = sasPattern.matcher(connectionString); - assertTrue(matcher.find(), "Couldn't find SAS from connection string"); - ServiceBusAdministrationClient client = new ServiceBusAdministrationClientBuilder() - .endpoint("https://" + fullyQualifiedDomainName) - .credential(new AzureSasCredential(matcher.group(1))) - .buildClient(); - NamespaceProperties np = client.getNamespaceProperties(); - assertNotNull(np.getName()); - } + private final AtomicReference credentialCached = new AtomicReference<>(); //region Create tests @@ -505,7 +430,7 @@ void getNamespace() { final NamespaceProperties namespaceProperties = client.getNamespaceProperties(); assertEquals(NamespaceType.MESSAGING, namespaceProperties.getNamespaceType()); if (!interceptorManager.isPlaybackMode()) { - final String[] split = TestUtils.getFullyQualifiedDomainName().split("\\.", 2); + final String[] split = TestUtils.getFullyQualifiedDomainName(true).split("\\.", 2); assertEquals(split[0], namespaceProperties.getName()); } } @@ -932,26 +857,9 @@ void listRules() { //endregion private ServiceBusAdministrationClient getClient() { - final String connectionString = interceptorManager.isPlaybackMode() - ? "Endpoint=sb://" + getFullyQualifiedDomainName() + ";SharedAccessKeyName=dummyKey;SharedAccessKey=dummyAccessKey" - : TestUtils.getConnectionString(false); - final ServiceBusAdministrationClientBuilder builder = new ServiceBusAdministrationClientBuilder() - .httpLogOptions(new HttpLogOptions().setLogLevel(HttpLogDetailLevel.BODY_AND_HEADERS)) - .connectionString(connectionString) - .retryOptions(new RetryOptions(new FixedDelayOptions(1, TIMEOUT))); - - if (interceptorManager.isPlaybackMode()) { - builder.httpClient(interceptorManager.getPlaybackClient()); - } else if (interceptorManager.isRecordMode()) { - builder.addPolicy(interceptorManager.getRecordPolicy()); - } - - if (!interceptorManager.isLiveMode()) { - interceptorManager.addSanitizers(ServiceBusAdministrationAsyncClientIntegrationTest.TEST_PROXY_SANITIZERS); - interceptorManager.addMatchers(ServiceBusAdministrationAsyncClientIntegrationTest.TEST_PROXY_REQUEST_MATCHERS); - } - + .httpLogOptions(new HttpLogOptions().setLogLevel(HttpLogDetailLevel.BODY_AND_HEADERS)); + configure(builder, null, interceptorManager, credentialCached); return builder.buildClient(); } } diff --git a/sdk/servicebus/test-resources.json b/sdk/servicebus/test-resources.json index c30d4ef9dfee..717792dfa590 100644 --- a/sdk/servicebus/test-resources.json +++ b/sdk/servicebus/test-resources.json @@ -35,12 +35,6 @@ "description": "The application client id used to run tests." } }, - "testApplicationSecret": { - "type": "string", - "metadata": { - "description": "The application client secret used to run tests." - } - }, "location": { "type": "string", "defaultValue": "[resourceGroup().location]", @@ -75,7 +69,7 @@ "name": "[variables('namespaceName')]", "location": "[parameters('location')]", "sku": { - "name": "Standard" + "name": "Premium" }, "properties": {}, "resources": [ @@ -108,30 +102,6 @@ "principalId": "[parameters('testApplicationOid')]" } }, - { - "type": "Microsoft.Authorization/roleAssignments", - "apiVersion": "2018-01-01-preview", - "name": "[guid(concat('dataReceiverRoleId', parameters('baseName')))]", - "dependsOn": [ - "[concat('Microsoft.ServiceBus/namespaces/', variables('namespaceName'))]" - ], - "properties": { - "roleDefinitionId": "[variables('serviceBusDataReceiverRoleId')]", - "principalId": "[parameters('testApplicationOid')]" - } - }, - { - "type": "Microsoft.Authorization/roleAssignments", - "apiVersion": "2018-01-01-preview", - "name": "[guid(concat('dataSenderRoleId', parameters('baseName')))]", - "dependsOn": [ - "[concat('Microsoft.ServiceBus/namespaces/', variables('namespaceName'))]" - ], - "properties": { - "roleDefinitionId": "[variables('serviceBusDataSenderRoleId')]", - "principalId": "[parameters('testApplicationOid')]" - } - }, { "type": "Microsoft.ServiceBus/namespaces/topics", "apiVersion": "2021-01-01-preview", @@ -227,10 +197,6 @@ "type": "string", "value": "[parameters('testApplicationId')]" }, - "AZURE_CLIENT_SECRET": { - "type": "string", - "value": "[parameters('testApplicationSecret')]" - }, "AZURE_SERVICEBUS_FULLY_QUALIFIED_DOMAIN_NAME": { "type": "string", "value": "[concat(variables('namespaceName'), parameters('serviceBusEndpointSuffix'))]" diff --git a/sdk/servicebus/tests.yml b/sdk/servicebus/tests.yml index 255795fc2aae..9be1da120763 100644 --- a/sdk/servicebus/tests.yml +++ b/sdk/servicebus/tests.yml @@ -3,20 +3,25 @@ trigger: none extends: template: /eng/pipelines/templates/stages/archetype-sdk-tests.yml parameters: + UseFederatedAuth: true ServiceDirectory: servicebus - Artifacts: - - name: azure-messaging-servicebus - groupId: com.azure - safeName: azuremessagingservicebus - SupportedClouds: 'Public,UsGov,China' + TestResourceDirectories: + - servicebus + SupportedClouds: 'Public' + CloudConfig: + Public: + ServiceConnection: azure-sdk-tests + SubscriptionConfigurationFilePaths: + - eng/common/TestResources/sub-config/AzurePublicMsft.json + SubscriptionConfiguration: $(sub-config-azure-cloud-test-resources) EnvVars: AZURE_LOG_LEVEL: 1 - AZURE_CLIENT_ID: $(SERVICEBUS_CLIENT_ID) - AZURE_TENANT_ID: $(SERVICEBUS_TENANT_ID) - AZURE_CLIENT_SECRET: $(SERVICEBUS_CLIENT_SECRET) - AZURE_SUBSCRIPTION_ID: $(SERVICEBUS_SUBSCRIPTION_ID) AdditionalMatrixConfigs: - Name: version_overrides_tests Path: sdk/servicebus/version-overrides-matrix.json Selection: all GenerateVMJobs: true + Artifacts: + - name: azure-messaging-servicebus + groupId: com.azure + safeName: azuremessagingservicebus