Test link refusal

.github/workflows/test-pr.yml

@@ -8,7 +8,6 @@ on:
 jobs:
   build:
     runs-on: ubuntu-24.04
-
     steps:
       - uses: actions/checkout@v5
       - name: Checkout tls-gen

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

@@ -19,13 +19,15 @@
 
 import static com.rabbitmq.client.amqp.Management.ExchangeType.FANOUT;
 import static com.rabbitmq.client.amqp.impl.TestConditions.BrokerVersion.RABBITMQ_4_1_0;
+import static com.rabbitmq.client.amqp.impl.TestConditions.BrokerVersion.RABBITMQ_4_2_0;
 import static com.rabbitmq.client.amqp.impl.TestUtils.*;
 import static java.nio.charset.StandardCharsets.*;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.qpid.protonj2.client.DeliveryMode.AT_LEAST_ONCE;
 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 +425,76 @@ void dynamicReceiver() throws Exception {
       assertThat(delivery.message().body()).isEqualTo(body);
     }
   }
+
+  @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_2_0)
+  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
+  @BrokerVersionAtLeast(RABBITMQ_4_2_0)
+  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
+  @BrokerVersionAtLeast(RABBITMQ_4_2_0)
+  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
+  @BrokerVersionAtLeast(RABBITMQ_4_2_0)
+  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);
+  }
 }