fixup: upgarding test-harnes

Signed-off-by: Simon Schrottner <[email protected]>

Author: aepfli

libs/providers/flagd/src/e2e/constants.ts

@@ -4,7 +4,7 @@ export const FLAGD_NAME = 'flagd';
 export const UNSTABLE_CLIENT_NAME = 'unstable';
 export const UNAVAILABLE_CLIENT_NAME = 'unavailable';
 
-export const GHERKIN_FLAGD = getGherkinTestPath('*.feature');
+export const GHERKIN_FLAGD = getGherkinTestPath('rpc-caching.feature');
 export const GHERKIN_CONFIG = getGherkinTestPath('config.feature');
 export const GHERKIN_FLAGD_FEATURE = getGherkinTestPath('flagd.feature');
 export const GHERKIN_FLAGD_TESTING = getGherkinTestPath('testing.feature');

libs/providers/flagd/src/e2e/step-definitions/configSteps.ts

@@ -38,5 +38,9 @@ export const configSteps: Steps = (state: State) => {
         expect(configElement).toBe(expected);
       },
     );
+
+    then('we should have an error', () => {
+
+    });
   };
 };

libs/providers/flagd/src/e2e/step-definitions/eventSteps.ts

@@ -34,16 +34,25 @@ export const eventSteps: Steps =
     then(/^the (.*) event handler should have been executed$/, async (type: string) => {
       await waitFor(() => expect(state.events.find((value) => value.type == type)).toBeDefined(), { timeout: 20000 });
       expect(state.events.find((value) => value.type == type)).toBeDefined();
-      state.events = state.events.filter((a) => a.type !== type);
+      state.events = [];
     });
 
     then(/^the (.*) event handler should have been executed within (\d+)ms$/, async (type: string, ms: number) => {
       await waitFor(() => expect(state.events.find((value) => value.type == type)).toBeDefined(), { timeout: ms });
       const actual = state.events.find((value) => value.type == type);
       expect(actual).toBeDefined();
-      state.events = state.events.filter((a) => a.type !== type);
+      state.events = [];
       console.error('here bin cih');
     });
 
-    when(/^a (.*) event was fired$/, () => {});
+    when(/^a (.*) event was fired$/, async (type: string) => {
+      await waitFor(() => expect(state.events.find((value) => value.type == type)), { timeout: 1000 });
+      expect(state.events.find((value) => value.type == type)).toBeDefined();
+      state.events = [];
+    });
+
+    then('the flag should be part of the event payload', async () => {
+      await waitFor(() => expect(state.events.find((value) => value.type == 'change')), { timeout: 1000 });
+      state.events = [];
+    });
   };

libs/providers/flagd/src/e2e/step-definitions/flagSteps.ts

@@ -64,15 +64,4 @@ export const flagSteps: Steps =
     then(/^the variant should be "(.*)"$/, (arg0) => {
       expect(state.details?.variant).toBe(arg0);
     });
-    when('the flag was modified', async () => {
-      await waitFor(
-        () =>
-          expect(
-            state.events.find(
-              (value) => value.type == 'change' && (value.details?.flagsChanged as string[]).includes(state.flag!.name),
-            ),
-          ).toBeDefined(),
-        { timeout: 5000 },
-      );
-    });
   };

libs/providers/flagd/src/e2e/step-definitions/providerSteps.ts

@@ -2,64 +2,59 @@ import { OpenFeature } from '@openfeature/server-sdk';
 import { FlagdContainer } from '../tests/flagdContainer';
 import { State, Steps } from './state';
 import { FlagdProvider } from '../../lib/flagd-provider';
-
-type Containers = Record<string, FlagdContainer> & { default: FlagdContainer };
+import { waitFor } from "./utils";
 
 export const providerSteps: Steps =
   (state: State) =>
   ({ given, when, then }) => {
-    const containers: Containers = {
-      default: FlagdContainer.build(),
-    };
+    const container: FlagdContainer = FlagdContainer.build();
     beforeAll(async () => {
       console.log('Setting flagd provider...');
 
-      const promises = [];
-
-      for (const container of Object.values(containers)) {
-        promises.push(container.start());
-      }
-
-      return Promise.all(promises);
+      return container.start();
     }, 50000);
 
     afterAll(async () => {
       await OpenFeature.close();
-      for (const container of Object.values(containers)) {
-        await container.stop();
-      }
+      await container.stop();
     });
 
     beforeEach(async () => {
-      const promises = [];
-
-      for (const container of Object.values(containers)) {
-        promises.push(container.start());
+      if (container.isStarted()) {
+        return container.start();
       }
-
-      return Promise.all(promises);
+      return Promise.resolve();
     }, 50000);
-
-    function getContainer(providerType: string): FlagdContainer {
-      if (Object.hasOwn(containers, providerType)) {
-        return containers[providerType];
+    afterEach(async () => {
+      if (state.client) {
+        await fetch('http://' + container.getLaunchpadUrl() + '/stop');
+        await new Promise((r) => setTimeout(r, 100));
       }
-      return containers.default;
-    }
+      return Promise.resolve();
+    }, 50000);
 
     given(/a (.*) flagd provider/, async (providerType: string) => {
-      const container = getContainer(providerType);
       const flagdOptions: Record<string, unknown> = {
         resolverType: state.resolverType,
+        deadline: 2000,
       };
+      let type = 'default';
       switch (providerType) {
         default:
           flagdOptions['port'] = container.getPort(state.resolverType);
           break;
         case 'unavailable':
           flagdOptions['port'] = 9999;
           break;
+        case 'ssl':
+          // todo: configure properly
+          flagdOptions['port'] = container.getPort(state.resolverType);
+          type = 'ssl';
+          break;
       }
+
+      await fetch('http://' + container.getLaunchpadUrl() + '/start?config=' + type);
+      await new Promise((r) => setTimeout(r, 500));
       if (providerType == 'unavailable') {
         OpenFeature.setProvider(providerType, new FlagdProvider(flagdOptions));
       } else {
@@ -71,8 +66,13 @@ export const providerSteps: Steps =
     });
 
     when(/^the connection is lost for (\d+)s$/, async (time) => {
-      const container = getContainer(state.providerType!);
-      await container.stop();
-      setTimeout(() => container.start(), time * 1000);
+      console.log('stopping flagd');
+      await fetch('http://' + container.getLaunchpadUrl() + '/restart?seconds=' + time);
+      await new Promise((r) => setTimeout(r, 50));
+    });
+
+    when('the flag was modified', async () => {
+      await fetch('http://' + container.getLaunchpadUrl() + '/change');
+      await new Promise((r) => setTimeout(r, 50));
     });
   };

libs/providers/flagd/src/e2e/tests/flagdContainer.ts

@@ -8,8 +8,8 @@ export class FlagdContainer extends GenericContainer {
   private started: StartedTestContainer | undefined;
   private stopped: StoppedTestContainer | undefined;
 
-  public static build(feature: string | undefined = undefined) {
-    return new FlagdContainer(this.generateImageName(feature)).withExposedPorts(8013, 8014, 8015, 8016);
+  public static build() {
+    return new FlagdContainer(this.generateImageName()).withExposedPorts(8080, 8013, 8014, 8015, 8016);
   }
 
   private constructor(image: string) {
@@ -42,15 +42,15 @@ export class FlagdContainer extends GenericContainer {
     return containerPromise;
   }
 
-  private static generateImageName(feature: string | undefined): string {
+  getLaunchpadUrl() {
+    return this.started?.getHost() + ':' + this.started?.getMappedPort(8080);
+  }
+
+  private static generateImageName(): string {
     const image = this.imageBase;
     const file = path.join(__dirname, './../../../../../shared/flagd-core/test-harness/', 'version.txt');
     const version = fs.readFileSync(file, 'utf8').trim();
-    let featurePart = '';
-    if (feature) {
-      featurePart = `-${feature}`;
-    }
-    return `${image}${featurePart}:v${version}`;
+    return `${image}:v${version}`;
   }
 
   getPort(resolverType: ResolverType) {

libs/providers/flagd/src/e2e/tests/rpc.spec.ts

@@ -20,7 +20,8 @@ describe('rpc', () => {
   };
   autoBindSteps(
     loadFeatures(GHERKIN_FLAGD, {
-      tagFilter: '@rpc and not @targetURI and not @customCert and not @events and not @sync  and not @offline and not @grace',
+      tagFilter:
+        '@rpc and not @targetURI and not @customCert and not @events and not @stream and not @grace',
       scenarioNameTemplate: (vars) => {
         return `${vars.scenarioTitle} (${vars.scenarioTags.join(',')} ${vars.featureTags.join(',')})`;
       },

libs/providers/flagd/src/lib/configuration.ts

@@ -18,6 +18,13 @@ export interface Config {
    */
   port: number;
 
+  /**
+   * The deadline for connections.
+   *
+   * @default 500
+   */
+  deadlineMs: number;
+
   /**
    * Determines if TLS should be used.
    *
@@ -79,6 +86,7 @@ export interface Config {
 export type FlagdProviderOptions = Partial<Config>;
 
 const DEFAULT_CONFIG: Omit<Config, 'port' | 'resolverType'> = {
+  deadlineMs: 500,
   host: 'localhost',
   tls: false,
   selector: '',
@@ -93,6 +101,7 @@ const DEFAULT_IN_PROCESS_CONFIG: Config = { ...DEFAULT_CONFIG, resolverType: 'in
 enum ENV_VAR {
   FLAGD_HOST = 'FLAGD_HOST',
   FLAGD_PORT = 'FLAGD_PORT',
+  FLAGD_DEADLINE_MS = 'FLAGD_DEADLINE_MS',
   FLAGD_TLS = 'FLAGD_TLS',
   FLAGD_SOCKET_PATH = 'FLAGD_SOCKET_PATH',
   FLAGD_CACHE = 'FLAGD_CACHE',
@@ -128,6 +137,9 @@ const getEnvVarConfig = (): Partial<Config> => {
     ...(Number(process.env[ENV_VAR.FLAGD_PORT]) && {
       port: Number(process.env[ENV_VAR.FLAGD_PORT]),
     }),
+    ...(Number(process.env[ENV_VAR.FLAGD_DEADLINE_MS]) && {
+      deadlineMs: Number(process.env[ENV_VAR.FLAGD_DEADLINE_MS]),
+    }),
     ...(process.env[ENV_VAR.FLAGD_TLS] && {
       tls: process.env[ENV_VAR.FLAGD_TLS]?.toLowerCase() === 'true',
     }),

libs/providers/flagd/src/lib/service/grpc/grpc-service.ts

@@ -70,6 +70,7 @@ export class GRPCService implements Service {
   private _cache: LRUCache<string, ResolutionDetails<FlagValue>> | undefined;
   private _cacheEnabled = false;
   private _eventStream: ClientReadableStream<EventStreamResponse> | undefined = undefined;
+  private _deadline: number;
   private get _cacheActive() {
     // the cache is "active" (able to be used) if the config enabled it, AND the gRPC stream is live
     return this._cacheEnabled && this._client.getChannel().getConnectivityState(false) === ConnectivityState.READY;
@@ -95,6 +96,7 @@ export class GRPCService implements Service {
           tls ? credentials.createSsl() : credentials.createInsecure(),
           clientOptions,
         );
+    this._deadline = config.deadlineMs;
 
     if (config.cache === 'lru') {
       this._cacheEnabled = true;
@@ -165,11 +167,15 @@ export class GRPCService implements Service {
     // close the previous stream if we're reconnecting
     closeStreamIfDefined(this._eventStream);
 
-    const stream = this._client.eventStream({}, {});
-    stream.on('error', (err: Error) => {
-      rejectConnect?.(err);
-      this.handleError(reconnectCallback, changedCallback, disconnectCallback);
+    this._client.waitForReady(this._deadline, (err) => {
+      console.log('waiting to be ready', err);
+      if (err) {
+        rejectConnect?.(err);
+        this.handleError(reconnectCallback, changedCallback, disconnectCallback);
+      }
     });
+    const stream = this._client.eventStream({}, {});
+
     stream.on('data', (message) => {
       if (message.type === EVENT_PROVIDER_READY) {
         this.logger?.debug(`${FlagdProvider.name}: streaming connection established with flagd`);
@@ -240,6 +246,7 @@ export class GRPCService implements Service {
     const resolver = promisify(promise);
     if (this._cacheActive) {
       const cached = this._cache?.get(flagKey);
+      console.log('cache active and this was in the cache', cached);
       if (cached) {
         return { ...cached, reason: StandardResolutionReasons.CACHED } as ResolutionDetails<T>;
       }