testing storage connector contract

Signed-off-by: Kavindu Dodanduwa <[email protected]>

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessResolver.java

@@ -61,7 +61,7 @@ public void init() throws Exception {
                         case STALE:
                             // todo set stale state
                         default:
-                            log.log(Level.INFO, String.format("Storage emitted unknown status: %s", storageState));
+                            log.log(Level.INFO, String.format("Storage emitted unhandled status: %s", storageState));
                     }
                 }
             } catch (InterruptedException e) {

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStore.java

@@ -6,6 +6,7 @@
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayload;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import lombok.extern.java.Log;
+
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
@@ -119,6 +120,8 @@ private void streamerListener(final Connector connector) throws InterruptedExcep
                     log.log(Level.INFO, String.format("Payload with unknown type: %s", take.getType()));
             }
         }
+
+        log.log(Level.INFO, "Shutting down store stream listener");
     }
 
 }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/TestUtils.java

@@ -0,0 +1,25 @@
+package dev.openfeature.contrib.providers.flagd.resolver.process;
+
+import dev.openfeature.contrib.providers.flagd.resolver.process.model.FlagParser;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+public class TestUtils {
+    public static final String VALID_SIMPLE = "flagConfigurations/valid-simple.json";
+    public static final String VALID_LONG = "flagConfigurations/valid-long.json";
+    public static final String INVALID_FLAG = "flagConfigurations/invalid-flag.json";
+    public static final String INVALID_CFG = "flagConfigurations/invalid-configuration.json";
+
+
+    public static String getFlagsFromResource(final String file) throws IOException {
+        final URL url = FlagParser.class.getClassLoader().getResource(file);
+        if (url == null) {
+            throw new IllegalStateException(String.format("Resource %s not found", file));
+        } else {
+            return new String(Files.readAllBytes(Paths.get(url.getPath())));
+        }
+    }
+}

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/model/FlagParserTest.java

@@ -3,20 +3,18 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Paths;
 import java.util.Map;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.INVALID_CFG;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.INVALID_FLAG;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.VALID_LONG;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.VALID_SIMPLE;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.getFlagsFromResource;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 class FlagParserTest {
-
-    private static final String VALID_SIMPLE = "flagConfigurations/valid-simple.json";
-    private static final String VALID_LONG = "flagConfigurations/valid-long.json";
-    private static final String INVALID_FLAG = "flagConfigurations/invalid-flag.json";
-    private static final String INVALID_CFG = "flagConfigurations/invalid-configuration.json";
-
     @Test
     public void validJsonConfigurationParsing() throws IOException {
         Map<String, FeatureFlag> flagMap = FlagParser.parseString(getFlagsFromResource(VALID_SIMPLE));
@@ -65,13 +63,4 @@ public void invalidConfigurationsThrowsError() {
             FlagParser.parseString(getFlagsFromResource(INVALID_CFG));
         });
     }
-
-    private static String getFlagsFromResource(final String file) throws IOException {
-        final URL url = FlagParser.class.getClassLoader().getResource(file);
-        if (url == null) {
-            throw new IllegalStateException(String.format("Resource %s not found", file));
-        } else {
-            return new String(Files.readAllBytes(Paths.get(url.getPath())));
-        }
-    }
 }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStoreTest.java

@@ -0,0 +1,72 @@
+package dev.openfeature.contrib.providers.flagd.resolver.process.storage;
+
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayload;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayloadType;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.INVALID_FLAG;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.VALID_LONG;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.VALID_SIMPLE;
+import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.getFlagsFromResource;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTimeout;
+
+class FlagStoreTest {
+
+    @Test
+    public void connectorHandling() {
+        final BlockingQueue<StreamPayload> payload = new LinkedBlockingQueue<>();
+        FlagStore store = new FlagStore(new MockConnector(payload));
+
+        store.init();
+        final BlockingQueue<StorageState> states = store.getStateQueue();
+
+        // OK for simple flag
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            payload.offer(new StreamPayload(StreamPayloadType.DATA, getFlagsFromResource(VALID_SIMPLE)));
+        });
+
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            assertEquals(StorageState.OK,  states.take());
+        });
+
+        // STALE for invalid flag
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            payload.offer(new StreamPayload(StreamPayloadType.DATA, getFlagsFromResource(INVALID_FLAG)));
+        });
+
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            assertEquals(StorageState.STALE,  states.take());
+        });
+
+        // OK again for next payload
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            payload.offer(new StreamPayload(StreamPayloadType.DATA, getFlagsFromResource(VALID_LONG)));
+        });
+
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            assertEquals(StorageState.OK,  states.take());
+        });
+
+        // ERROR is propagated correctly
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            payload.offer(new StreamPayload(StreamPayloadType.ERROR, null));
+        });
+
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            assertEquals(StorageState.ERROR,  states.take());
+        });
+
+        // Shutdown handling
+        store.shutdown();
+
+        assertTimeout(Duration.ofMillis(200), ()-> {
+            assertEquals(StorageState.ERROR,  states.take());
+        });
+    }
+
+}

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/MockConnector.java

@@ -0,0 +1,34 @@
+package dev.openfeature.contrib.providers.flagd.resolver.process.storage;
+
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.Connector;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayload;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayloadType;
+import lombok.extern.java.Log;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.logging.Level;
+
+@Log
+public class MockConnector implements Connector {
+
+    private BlockingQueue<StreamPayload> mockQueue;
+
+    MockConnector(final BlockingQueue<StreamPayload> mockQueue){
+        this.mockQueue = mockQueue;
+    }
+
+    public void init() {
+        // no-op
+    }
+
+    public BlockingQueue<StreamPayload> getStream() {
+        return mockQueue;
+    }
+
+    public void shutdown() {
+        // Emit error mocking closed connection scenario
+        if(!mockQueue.offer(new StreamPayload(StreamPayloadType.ERROR, "shutdown invoked"))){
+            log.log(Level.WARNING, "Failed to offer shutdown status");
+        }
+    }
+}