E2e Tests: allow to pass google json sa

langstream-e2e-tests/src/test/java/ai/langstream/tests/PythonAgentsIT.java

@@ -83,7 +83,9 @@ public void testProcessor() throws Exception {
         deleteAppAndAwaitCleanup(tenant, applicationId);
 
         final List<String> topics = getAllTopics();
-        Assertions.assertEquals(List.of("ls-test-topic0"), topics);
+        log.info("all topics: {}", topics);
+        Assertions.assertTrue(topics.contains("ls-test-topic0"));
+        Assertions.assertFalse(topics.contains("ls-test-topic1"));
     }
 
     @Test

langstream-e2e-tests/src/test/java/ai/langstream/tests/TextCompletionsIT.java

@@ -51,7 +51,7 @@ public static Map<String, String> getAppEnvForAIServiceProvider() {
             return getAppEnvMapFromSystem(
                     List.of(
                             "VERTEX_AI_URL",
-                            "VERTEX_AI_TOKEN",
+                            "VERTEX_AI_SERVICE_ACCOUNT_JSON",
                             "VERTEX_AI_REGION",
                             "VERTEX_AI_PROJECT"));
         }
@@ -69,7 +69,7 @@ public void test() throws Exception {
         final String sessionId = UUID.randomUUID().toString();
 
         executeCommandOnClient(
-                "bin/langstream gateway produce %s produce-input -v 'Translate \"Good morning\" to French.' -p sessionId=%s"
+                "bin/langstream gateway produce %s produce-input -v 'Translate \"Apple\" to French, lowercase.' -p sessionId=%s"
                         .formatted(applicationId, sessionId)
                         .split(" "));
 
@@ -81,6 +81,6 @@ public void test() throws Exception {
                                 .formatted(sessionId)
                                 .split(" "));
         log.info("Output: {}", message);
-        Assertions.assertTrue(message.getAnswerFromChatCompletionsValue().contains("Bonjour"));
+        Assertions.assertTrue(message.getAnswerFromChatCompletionsValue().contains("pomme"));
     }
 }

langstream-e2e-tests/src/test/java/ai/langstream/tests/WebCrawlerToVectorIT.java

@@ -56,6 +56,15 @@ public static void checkCredentials() {
 
     @Test
     public void test() throws Exception {
+        if (!codeStorageConfig.type().equals("s3")) {
+            throw new IllegalStateException(
+                    "This test can only run with S3 code storage, but got: "
+                            + codeStorageConfig.type());
+        }
+
+        appEnv.put("S3_ENDPOINT", codeStorageConfig.configuration().get("endpoint"));
+        appEnv.put("S3_ACCESS_KEY", codeStorageConfig.configuration().get("access-key"));
+        appEnv.put("S3_SECRET_KEY", codeStorageConfig.configuration().get("secret-key"));
         installLangStreamCluster(true);
         final String tenant = "ten-" + System.currentTimeMillis();
         setupTenant(tenant);

langstream-e2e-tests/src/test/java/ai/langstream/tests/util/BaseEndToEndTest.java

@@ -526,8 +526,7 @@ private static KubeCluster getKubeCluster() {
     public void setupSingleTest() {
         // cleanup previous runs
         cleanupAllEndToEndTestsNamespaces();
-        codeStorageProvider.cleanup();
-        streamingClusterProvider.cleanup();
+        cleanupEnv();
 
         namespace = "ls-test-" + UUID.randomUUID().toString().substring(0, 8);
 
@@ -541,18 +540,29 @@ public void setupSingleTest() {
                 .serverSideApply();
     }
 
+    private void cleanupEnv() {
+        if (codeStorageProvider != null) {
+            codeStorageProvider.cleanup();
+        }
+        if (streamingClusterProvider != null) {
+            streamingClusterProvider.cleanup();
+        }
+    }
+
     @AfterEach
     public void cleanupAfterEach() {
         cleanupAllEndToEndTestsNamespaces();
-        streamingClusterProvider.cleanup();
+        cleanupEnv();
     }
 
     private static void cleanupAllEndToEndTestsNamespaces() {
-        client.namespaces().withLabel("app", "ls-test").delete();
-        client.namespaces().list().getItems().stream()
-                .map(ns -> ns.getMetadata().getName())
-                .filter(ns -> ns.startsWith(TENANT_NAMESPACE_PREFIX))
-                .forEach(ns -> deleteTenantNamespace(ns));
+        if (client != null) {
+            client.namespaces().withLabel("app", "ls-test").delete();
+            client.namespaces().list().getItems().stream()
+                    .map(ns -> ns.getMetadata().getName())
+                    .filter(ns -> ns.startsWith(TENANT_NAMESPACE_PREFIX))
+                    .forEach(ns -> deleteTenantNamespace(ns));
+        }
     }
 
     private static void deleteTenantNamespace(String ns) {
@@ -1095,7 +1105,12 @@ private static void deployLocalApplicationAndAwaitReady(
         if (env != null && !env.isEmpty()) {
             beforeCmd =
                     env.entrySet().stream()
-                            .map(e -> "export \"%s\"=\"%s\"".formatted(e.getKey(), e.getValue()))
+                            .map(
+                                    e ->
+                                            "export '%s'='%s'"
+                                                    .formatted(
+                                                            e.getKey(),
+                                                            e.getValue().replace("'", "''")))
                             .collect(Collectors.joining(" && "));
             beforeCmd += " && ";
         }
@@ -1122,11 +1137,10 @@ private static void deployLocalApplicationAndAwaitReady(
         } else {
             podUids = "";
         }
-        executeCommandOnClient(
-                (beforeCmd
-                                + "bin/langstream apps %s %s -app /tmp/app -i /tmp/instance.yaml -s /tmp/secrets.yaml")
-                        .formatted(isUpdate ? "update" : "deploy", applicationId)
-                        .split(" "));
+        final String command =
+                "bin/langstream apps %s %s -app /tmp/app -i /tmp/instance.yaml -s /tmp/secrets.yaml"
+                        .formatted(isUpdate ? "update" : "deploy", applicationId);
+        executeCommandOnClient((beforeCmd + command).split(" "));
 
         awaitApplicationReady(applicationId, expectedNumExecutors);
         Awaitility.await()

langstream-e2e-tests/src/test/resources/apps/chat-completions/configuration.yaml

@@ -29,6 +29,6 @@ configuration:
       id: "vertex"
       configuration:
         url: "{{ secrets.vertex-ai.url }}"
-        token: "{{ secrets.vertex-ai.token }}"
+        serviceAccountJson: "{{ secrets.vertex-ai.service-json }}"
         region: "{{ secrets.vertex-ai.region }}"
         project: "{{ secrets.vertex-ai.project }}"

langstream-e2e-tests/src/test/resources/apps/text-completions/configuration.yaml

@@ -29,6 +29,6 @@ configuration:
       id: "vertex"
       configuration:
         url: "{{ secrets.vertex-ai.url }}"
-        token: "{{ secrets.vertex-ai.token }}"
+        serviceAccountJson: "{{ secrets.vertex-ai.service-json }}"
         region: "{{ secrets.vertex-ai.region }}"
         project: "{{ secrets.vertex-ai.project }}"

langstream-e2e-tests/src/test/resources/apps/webcrawler-to-vector/chatbot.yaml

@@ -34,7 +34,8 @@ pipeline:
   - name: "compute-embeddings"
     type: "compute-ai-embeddings"
     configuration:
-      model: "{{{secrets.open-ai.embeddings-model}}}"
+      ai-service: "{{{secrets.embeddings.service}}}"
+      model: "{{{secrets.embeddings.model}}}"
       embeddings-field: "value.question_embeddings"
       text: "{{% value.question }}"
       flush-interval: 0

langstream-e2e-tests/src/test/resources/apps/webcrawler-to-vector/configuration.yaml

@@ -38,6 +38,6 @@ configuration:
     id: "vertex"
     configuration:
       url: "{{ secrets.vertex-ai.url }}"
-      token: "{{ secrets.vertex-ai.token }}"
+      serviceAccountJson: "{{ secrets.vertex-ai.service-json }}"
       region: "{{ secrets.vertex-ai.region }}"
       project: "{{ secrets.vertex-ai.project }}"

langstream-e2e-tests/src/test/resources/apps/webcrawler-to-vector/crawler.yaml

@@ -39,9 +39,9 @@ pipeline:
       handle-cookies: true
       max-unflushed-pages: 100
       bucketName: "langstream-test-crawler-to-vector"
-      endpoint: http://minio.minio-dev.svc.cluster.local:9000
-      access-key: minioadmin
-      secret-key: minioadmin
+      endpoint: "{{{secrets.s3.endpoint}}}"
+      access-key: "{{{secrets.s3.access-key}}}"
+      secret-key: "{{{secrets.s3.secret-key}}}"
   - name: "Extract text"
     type: "text-extractor"
   - name: "Normalise text"

langstream-e2e-tests/src/test/resources/secrets/secret1.yaml

@@ -52,7 +52,7 @@ secrets:
   - id: vertex-ai
     data:
       url: "${VERTEX_AI_URL:-}"
-      token: "${VERTEX_AI_TOKEN:-}"
+      service-json: "${VERTEX_AI_SERVICE_ACCOUNT_JSON:-}"
       region: "${VERTEX_AI_REGION:-}"
       project: "${VERTEX_AI_PROJECT:-}"
   - id: astra
@@ -64,4 +64,10 @@ secrets:
       environment: "${ASTRA_ENVIRONMENT}"
   - id: kafka
     data:
-      bootstrap-servers: "${KAFKA_BOOTSTRAP_SERVERS}"
+      bootstrap-servers: "${KAFKA_BOOTSTRAP_SERVERS}"
+
+  - id: s3
+    data:
+      endpoint: "${S3_ENDPOINT}"
+      access-key: "${S3_ACCESS_KEY}"
+      secret-key: "${S3_SECRET_KEY}"