test(sse): sync raw frame and consumer error contracts

Author: loong10k

src/test/java/io/github/easy4j/hermes/api/sse/SseLifecycleContractTest.java

@@ -21,6 +21,7 @@
 import okhttp3.extension.logging.HttpLogLevel;
 
 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.assertTrue;
 
@@ -29,7 +30,7 @@ class SseLifecycleContractTest {
     @Test
     void standardChatChunkDoesNotRequireNestedDataField() throws Exception {
         OkHttpClient client = new OkHttpClient.Builder().addInterceptor(chain -> {
-            String body = "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n"
+            String body = "data: {\"choices\":[{\"delta\":{\"content\":\"你好 \"}}]}\n\n"
                     + "data: [DONE]\n\n";
             return new Response.Builder()
                     .request(chain.request())
@@ -53,7 +54,9 @@ void standardChatChunkDoesNotRequireNestedDataField() throws Exception {
 
             assertTrue(complete.await(3, TimeUnit.SECONDS));
             assertNotNull(received.get());
-            assertEquals("hello", received.get().deltaText());
+            assertEquals("你好 ", received.get().deltaText());
+            assertEquals("{\"choices\":[{\"delta\":{\"content\":\"你好 \"}}]}",
+                    received.get().getData());
         }
     }
 
@@ -83,15 +86,97 @@ void malformedJsonAndConsumerFailureRemainIsolated() throws Exception {
         AtomicInteger consumerCalls = new AtomicInteger();
 
         try (HermesSseClient sse = new HermesSseClient(config, null, client)) {
-            CountDownLatch complete = new CountDownLatch(1);
-            sse.subscribeChat(request, ignored -> {
+            CountDownLatch failed = new CountDownLatch(1);
+            AtomicReference<Throwable> failure = new AtomicReference<>();
+            SseSubscription subscription = sse.subscribeChat(request, ignored -> {
                 consumerCalls.incrementAndGet();
                 throw new IllegalStateException("consumer failure");
-            }, complete::countDown, ignored -> { });
+            }, () -> { }, error -> {
+                failure.set(error);
+                failed.countDown();
+            });
 
-            assertTrue(complete.await(3, TimeUnit.SECONDS));
+            assertTrue(failed.await(3, TimeUnit.SECONDS));
             assertEquals(1, consumerCalls.get(),
                     "malformed JSON must not be delivered to the business consumer");
+            assertTrue(failure.get() instanceof SseConsumerException,
+                    "consumer callback failures must not be reported as JSON parsing failures");
+            assertFalse(subscription.isActive());
+            assertEquals(0, sse.activeSubscriptionCount());
+        }
+    }
+
+    @Test
+    void multiLineCrLfCommentAndFragmentedUtf8PreserveRawFrame() throws Exception {
+        try (MockWebServer server = new MockWebServer()) {
+            String body = ": keepalive\r\n"
+                    + "id: evt-1\r\n"
+                    + "event: assistant.delta\r\n"
+                    + "data: {\"choices\":[{\"delta\":\r\n"
+                    + "data: {\"content\":\"你 好 \"}}]}\r\n\r\n"
+                    + "data: [DONE]\r\n\r\n";
+            server.enqueue(new MockResponse()
+                    .setResponseCode(200)
+                    .setHeader("Content-Type", "text/event-stream")
+                    .setChunkedBody(body, 1));
+            server.start();
+
+            HermesHttpClientConfig config = new HermesHttpClientConfig()
+                    .setEndpointPolicy(io.github.easy4j.hermes.security.EndpointPolicy
+                            .trustedLocal("127.0.0.1", server.getPort()))
+                    .setBaseUrl("http://127.0.0.1:" + server.getPort());
+            ChatRequest request = new ChatRequest();
+            request.setMessages(Collections.singletonList(
+                    new ChatRequest.Message("user", "hello")));
+
+            try (HermesSseClient sse = new HermesSseClient(config, null, null)) {
+                CountDownLatch complete = new CountDownLatch(1);
+                AtomicReference<SseEvent> received = new AtomicReference<>();
+                sse.subscribeChat(request, received::set, complete::countDown, ignored -> { });
+
+                assertTrue(complete.await(3, TimeUnit.SECONDS));
+                SseEvent event = received.get();
+                assertNotNull(event);
+                assertEquals("evt-1", event.getId());
+                assertEquals("assistant.delta", event.getEvent());
+                assertEquals("你 好 ", event.deltaText());
+                assertTrue(event.getData().contains("\n"),
+                        "multiple data lines must remain visible in the raw SSE data");
+            }
+        }
+    }
+
+    @Test
+    void unknownEventPreservesServerIdentityAndRawData() throws Exception {
+        OkHttpClient client = new OkHttpClient.Builder().addInterceptor(chain -> {
+            String body = "id: mystery-7\n"
+                    + "event: mystery.data\n"
+                    + "data: {\"foo\":\"bar\"}\n\n"
+                    + "data: [DONE]\n\n";
+            return new Response.Builder()
+                    .request(chain.request())
+                    .protocol(Protocol.HTTP_1_1)
+                    .code(200)
+                    .message("OK")
+                    .body(ResponseBody.create(body, MediaType.get("text/event-stream")))
+                    .build();
+        }).build();
+
+        HermesHttpClientConfig config = new HermesHttpClientConfig();
+        config.markUnsafeBaseUrlOverriddenForTest(true);
+        ChatRequest request = new ChatRequest();
+        request.setMessages(Collections.singletonList(new ChatRequest.Message("user", "hello")));
+
+        try (HermesSseClient sse = new HermesSseClient(config, null, client)) {
+            CountDownLatch complete = new CountDownLatch(1);
+            AtomicReference<SseEvent> received = new AtomicReference<>();
+            sse.subscribeChat(request, received::set, complete::countDown, ignored -> { });
+
+            assertTrue(complete.await(3, TimeUnit.SECONDS));
+            assertNotNull(received.get());
+            assertEquals("mystery-7", received.get().getId());
+            assertEquals("mystery.data", received.get().getEvent());
+            assertEquals("{\"foo\":\"bar\"}", received.get().getData());
         }
     }