Test link refusal

References rabbitmq/rabbitmq-server#14389

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -9,7 +9,7 @@ jobs:
   build:
     runs-on: ubuntu-24.04
 
-    steps:
+steps:
       - uses: actions/checkout@v5
       - name: Checkout tls-gen
         uses: actions/checkout@v5
@@ -24,6 +24,8 @@ jobs:
           cache: 'maven'
       - name: Start broker
         run: ci/start-broker.sh
+        env:
+          RABBITMQ_IMAGE: pivotalrabbitmq/rabbitmq:pr-14389-otp28
       - name: Start toxiproxy
         run: ci/start-toxiproxy.sh
       - name: Display Java version

src/test/java/com/rabbitmq/client/amqp/impl/ClientTest.java

@@ -26,6 +26,7 @@
 import static org.apache.qpid.protonj2.client.DeliveryState.released;
 import static org.assertj.core.api.Assertions.*;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.InstanceOfAssertFactories.*;
 
 import com.rabbitmq.client.amqp.Environment;
 import com.rabbitmq.client.amqp.Management;
@@ -423,4 +424,72 @@ void dynamicReceiver() throws Exception {
       assertThat(delivery.message().body()).isEqualTo(body);
     }
   }
+
+  @Test
+  void refuseLinkSenderToMissingExchangeShouldReturnNotFound() throws Exception {
+    try (Client client = client()) {
+      org.apache.qpid.protonj2.client.Connection c = connection(client);
+      Session s = c.openSession();
+      assertThatThrownBy(
+              () -> ExceptionUtils.wrapGet(s.openSender("/exchanges/missing").openFuture()))
+          .isInstanceOf(ClientLinkRemotelyClosedException.class)
+          .asInstanceOf(throwable(ClientLinkRemotelyClosedException.class))
+          .matches(e -> "amqp:not-found".equals(e.getErrorCondition().condition()));
+      checkSession(s);
+    }
+  }
+
+  @Test
+  void refuseLinkSenderToInvalidAddressShouldReturnInvalidField() throws Exception {
+    try (Client client = client()) {
+      org.apache.qpid.protonj2.client.Connection c = connection(client);
+      Session s = c.openSession();
+      assertThatThrownBy(() -> ExceptionUtils.wrapGet(s.openSender("/fruit/orange").openFuture()))
+          .isInstanceOf(ClientLinkRemotelyClosedException.class)
+          .asInstanceOf(throwable(ClientLinkRemotelyClosedException.class))
+          .matches(e -> "amqp:invalid-field".equals(e.getErrorCondition().condition()));
+      checkSession(s);
+    }
+  }
+
+  @Test
+  void refuseLinkReceiverToMissingQueueShouldReturnNotFound() throws Exception {
+    try (Client client = client()) {
+      org.apache.qpid.protonj2.client.Connection c = connection(client);
+      Session s = c.openSession().openFuture().get(10, SECONDS);
+      assertThatThrownBy(
+              () ->
+                  ExceptionUtils.wrapGet(
+                      s.openReceiver("/queues/missing", new ReceiverOptions().creditWindow(0))
+                          .openFuture()))
+          .isInstanceOf(ClientLinkRemotelyClosedException.class)
+          .asInstanceOf(throwable(ClientLinkRemotelyClosedException.class))
+          .matches(e -> "amqp:not-found".equals(e.getErrorCondition().condition()));
+      checkSession(s);
+    }
+  }
+
+  @Test
+  void refuseLinkReceiverToInvalidAddressShouldReturnInvalidField() throws Exception {
+    try (Client client = client()) {
+      org.apache.qpid.protonj2.client.Connection c = connection(client);
+      Session s = c.openSession().openFuture().get(10, SECONDS);
+      assertThatThrownBy(
+              () ->
+                  ExceptionUtils.wrapGet(
+                      s.openReceiver("/fruit/orange", new ReceiverOptions().creditWindow(0))
+                          .openFuture()))
+          .isInstanceOf(ClientLinkRemotelyClosedException.class)
+          .asInstanceOf(throwable(ClientLinkRemotelyClosedException.class))
+          .matches(e -> "amqp:invalid-field".equals(e.getErrorCondition().condition()));
+      checkSession(s);
+    }
+  }
+
+  private static void checkSession(Session s) throws Exception {
+    ReceiverOptions receiverOptions = new ReceiverOptions();
+    receiverOptions.sourceOptions().capabilities("temporary-queue");
+    Receiver receiver = s.openDynamicReceiver(receiverOptions);
+    receiver.openFuture().get(10, SECONDS);
+  }
 }