test: finish Bearer Token test case

Deleted NimbusTokenProvider.BearerTokenFlow class, no longer used.
Added a wrapper for WeaviateClient that can tie client's lifetime
to that of the Testcontainer that created it. That way the client,
once opened, is only closed when the container it belonged to is stopped.

Author: bevzzz

src/it/java/io/weaviate/containers/Container.java

@@ -17,16 +17,6 @@ public class Container {
   public static final Contextionary CONTEXTIONARY = Contextionary.createDefault();
   public static final Img2VecNeural IMG2VEC_NEURAL = Img2VecNeural.createDefault();
 
-  static {
-    startAll();
-  }
-
-  /** Start all shared Testcontainers. */
-  // TODO: start lazily!
-  static void startAll() {
-    // WEAVIATE.start();
-  }
-
   /**
    * Stop all shared Testcontainers created in {@link #startAll}.
    * <p>

src/it/java/io/weaviate/containers/Weaviate.java

@@ -15,11 +15,11 @@
 import io.weaviate.client6.v1.internal.ObjectBuilder;
 
 public class Weaviate extends WeaviateContainer {
-  private WeaviateClient clientInstance;
-
   public static final String VERSION = "1.29.1";
   public static final String DOCKER_IMAGE = "semitechnologies/weaviate";
 
+  private volatile SharedClient clientInstance;
+
   public WeaviateClient getClient() {
     return getClient(ObjectBuilder.identity());
   }
@@ -30,24 +30,29 @@ public WeaviateClient getClient() {
    * this is not made thread-safe.
    */
   public WeaviateClient getClient(Function<Config.Custom, ObjectBuilder<Config>> fn) {
-    // FIXME: control from containers?
     if (!isRunning()) {
       start();
     }
-    if (clientInstance == null) {
-      var host = getHost();
-      var customFn = ObjectBuilder.partial(fn,
-          conn -> conn
-              .scheme("http")
-              .httpHost(host)
-              .grpcHost(host)
-              .httpPort(getMappedPort(8080))
-              .grpcPort(getMappedPort(50051)));
-      try {
-        clientInstance = WeaviateClient.custom(customFn);
-        // clientInstance = WeaviateClient.local();
-      } catch (Exception e) {
-        throw new RuntimeException("create WeaviateClient for Weaviate container", e);
+    if (clientInstance != null) {
+      return clientInstance;
+    }
+
+    synchronized (this) {
+      if (clientInstance == null) {
+        var host = getHost();
+        var customFn = ObjectBuilder.partial(fn,
+            conn -> conn
+                .scheme("http")
+                .httpHost(host)
+                .grpcHost(host)
+                .httpPort(getMappedPort(8080))
+                .grpcPort(getMappedPort(50051)));
+        var config = customFn.apply(new Config.Custom()).build();
+        try {
+          clientInstance = new SharedClient(config, this);
+        } catch (Exception e) {
+          throw new RuntimeException("create WeaviateClient for Weaviate container", e);
+        }
       }
     }
     return clientInstance;
@@ -149,10 +154,32 @@ public void stop() {
     if (clientInstance == null) {
       return;
     }
-    try {
-      clientInstance.close();
-    } catch (IOException e) {
-      // TODO: log error
+    synchronized (this) {
+      try {
+        clientInstance.close(this);
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
+
+  /** SharedClient's lifetime is tied to that of it's parent container. */
+  private class SharedClient extends WeaviateClient {
+    private final Weaviate parent;
+
+    private SharedClient(Config config, Weaviate parent) {
+      super(config);
+      this.parent = parent;
+    }
+
+    private void close(Weaviate caller) throws IOException {
+      if (caller == parent) {
+        super.close();
+      }
+    }
+
+    @Override
+    public void close() throws IOException {
     }
   }
 }

src/it/java/io/weaviate/integration/OIDCSupportITest.java

@@ -11,6 +11,8 @@
 
 import io.weaviate.ConcurrentTest;
 import io.weaviate.client6.v1.api.Authorization;
+import io.weaviate.client6.v1.internal.TokenProvider;
+import io.weaviate.client6.v1.internal.rest.RestTransport;
 import io.weaviate.containers.Weaviate;
 
 /**
@@ -45,9 +47,13 @@ public class OIDCSupportITest extends ConcurrentTest {
       .build();
 
   @Test
-  public void test_bearerToken() {
+  public void test_bearerToken() throws IOException {
     Assume.assumeTrue("WCS_DUMMY_CI_PW is not set", WCS_DUMMY_CI_PW != null);
     Assume.assumeTrue("no internet connection", hasInternetConnection());
+
+    var t = TokenInterceptor.stealToken();
+    var authz = Authorization.bearerToken(t.accessToken(), t.refreshToken(), t.expiresIn());
+    pingWeaviate(wcsContainer, authz);
   }
 
   @Test
@@ -66,6 +72,7 @@ public void test_clientCredentials() throws IOException {
 
     var authz = Authorization.clientCredentials(OKTA_CLIENT_ID, OKTA_CLIENT_SECRET, List.of());
     pingWeaviate(oktaContainer, authz);
+    pingWeaviate(oktaContainer, authz);
   }
 
   private static void pingWeaviate(final Weaviate container, Authorization authz) throws IOException {
@@ -87,4 +94,32 @@ private static boolean ping(String site) {
       return false;
     }
   }
+
+  private static class TokenInterceptor implements Authorization, TokenProvider {
+    /** Exchange resource owner password for a token and return it. */
+    public static Token stealToken() throws IOException {
+      var authz = Authorization.resourceOwnerPassword(WCS_DUMMY_CI_USERNAME, WCS_DUMMY_CI_PW, List.of());
+      var spy = new TokenInterceptor(authz);
+      pingWeaviate(wcsContainer, spy);
+      return spy.getToken();
+    }
+
+    private Authorization authorization;
+    private TokenProvider tokenProvider;
+
+    private TokenInterceptor(Authorization actual) {
+      this.authorization = actual;
+    }
+
+    @Override
+    public TokenProvider getTokenProvider(RestTransport transport) {
+      tokenProvider = authorization.getTokenProvider(transport);
+      return this;
+    }
+
+    @Override
+    public Token getToken() {
+      return tokenProvider.getToken();
+    }
+  }
 }

src/main/java/io/weaviate/client6/v1/internal/oidc/nimbus/HttpResponseParser.java

@@ -75,8 +75,6 @@ private static ErrorDetails fromErrorResponse(ErrorResponse response) {
   }
 
   private static ErrorDetails fromHttpResponse(HTTPResponse response) {
-    System.out.println(response.getBody());
-    System.out.println(JSON.deserialize(response.getBody(), ErrorDetails.class));
     return JSON.deserialize(response.getBody(), ErrorDetails.class);
   }
 }

src/main/java/io/weaviate/client6/v1/internal/oidc/nimbus/NimbusTokenProvider.java

@@ -4,13 +4,10 @@
 
 import javax.annotation.concurrent.NotThreadSafe;
 
-import com.nimbusds.oauth2.sdk.AuthorizationGrant;
 import com.nimbusds.oauth2.sdk.ParseException;
-import com.nimbusds.oauth2.sdk.RefreshTokenGrant;
 import com.nimbusds.oauth2.sdk.Scope;
 import com.nimbusds.oauth2.sdk.TokenRequest;
 import com.nimbusds.oauth2.sdk.id.ClientID;
-import com.nimbusds.oauth2.sdk.token.RefreshToken;
 import com.nimbusds.openid.connect.sdk.op.OIDCProviderMetadata;
 import com.nimbusds.openid.connect.sdk.token.OIDCTokens;
 
@@ -102,8 +99,8 @@ public Token getToken() {
         ? Token.expireAfter(accessToken.getValue(), accessToken.getLifetime())
         : Token.expireAfter(accessToken.getValue(), refreshToken.getValue(), accessToken.getLifetime());
 
-    if (flow instanceof BearerTokenFlow btf) {
-      btf.setToken(newToken);
+    if (flow instanceof RefreshTokenFlow rtf) {
+      rtf.setToken(newToken);
     }
 
     return newToken;
@@ -121,22 +118,4 @@ private static OIDCProviderMetadata _parseProviderMetadata(String providerMetada
       throw new WeaviateOAuthException("parse provider metadata: ", ex);
     }
   }
-
-  @NotThreadSafe
-  final class BearerTokenFlow implements Flow {
-    private Token t;
-
-    BearerTokenFlow(Token t) {
-      this.t = t;
-    }
-
-    @Override
-    public AuthorizationGrant getAuthorizationGrant() {
-      return new RefreshTokenGrant(new RefreshToken(t.refreshToken()));
-    }
-
-    public void setToken(Token t) {
-      this.t = t;
-    }
-  }
 }