refactor tests for stability (#32)

.gitignore

@@ -5,4 +5,5 @@ yarn-error.log
 /bazel-*
 .ijwb
 .idea
-target/
+target/
+*.log

package.json

@@ -21,6 +21,7 @@
     },
     "dependencies": {
         "@osskit/wiremock-client": "^4.3.1",
+        "@testcontainers/kafka": "^10.2.2",
         "@types/jest": "^24.0.12",
         "@types/lodash": "^4.14.149",
         "@types/node": "^18.16.3",
@@ -47,7 +48,7 @@
         "p-retry": "^5.1.2",
         "prettier": "^2.3.2",
         "prettier-plugin-java": "^1.3.0",
-        "testcontainers": "^9.6.0",
+        "testcontainers": "^10.2.2",
         "ts-node": "^10.9.1"
     },
     "husky": {

tests/jest.config.ts

@@ -21,6 +21,8 @@ const config: JestConfigWithTsJest = {
         // eslint-disable-next-line @typescript-eslint/naming-convention
         '^(\\.{1,2}/.*)\\.js$': '$1',
     },
+    setupFilesAfterEnv: ['<rootDir>/setupFilesAfterEnv.ts'],
+    testTimeout: 1800000,
 };
 
 export default config;

tests/services/consume.ts

@@ -0,0 +1,17 @@
+import {Kafka, KafkaMessage} from 'kafkajs';
+
+export const consume = async (kafka: Kafka, topic: string, parse = true) => {
+    const consumer = kafka.consumer({groupId: 'orchestrator'});
+    await consumer.subscribe({topic: topic, fromBeginning: true});
+    const consumedMessage = await new Promise<KafkaMessage>((resolve) => {
+        consumer.run({
+            eachMessage: async ({message}) => resolve(message),
+        });
+    });
+    await consumer.disconnect();
+    const value = parse ? JSON.parse(consumedMessage.value?.toString() ?? '{}') : consumedMessage.value?.toString();
+    const headers = Object.fromEntries(
+        Object.entries(consumedMessage.headers!).map(([key, value]) => [key, value?.toString()])
+    );
+    return {value, headers};
+};

tests/setupFilesAfterEnv.ts

@@ -0,0 +1,3 @@
+import {jest} from '@jest/globals';
+
+jest.retryTimes(3, {logErrorsBeforeRetry: true});

tests/specs/__snapshots__/produce.spec.ts.snap

@@ -0,0 +1,18 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`tests produce 1`] = `
+{
+  "headers": {
+    "x-b3-flags": "1",
+    "x-b3-parentspanid": "101112",
+    "x-b3-sampled": "1",
+    "x-b3-spanid": "789",
+    "x-b3-traceid": "456",
+    "x-ot-span-context": "foo",
+    "x-request-id": "123",
+  },
+  "value": {
+    "data": "foo",
+  },
+}
+`;

tests/specs/__snapshots__/produceFailure.spec.ts.snap

@@ -0,0 +1,3 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`tests produce failure 1`] = `[TypeError: fetch failed]`;

tests/specs/produce.spec.ts

@@ -0,0 +1,51 @@
+import delay from 'delay';
+
+import type {Orchestrator} from '../testcontainers/orchestrator.js';
+import {start} from '../testcontainers/orchestrator.js';
+import {consume} from '../services/consume.js';
+
+const topic = 'my-topic';
+
+describe('tests', () => {
+    let orchestrator: Orchestrator;
+
+    beforeEach(async () => {
+        orchestrator = await start(
+            {
+                KAFKA_BROKER: 'kafka:9092',
+                MAX_BLOCK_MS: '1000',
+            },
+            [topic]
+        );
+    }, 5 * 60 * 1000);
+
+    afterEach(async () => {
+        if (!orchestrator) {
+            return;
+        }
+        await orchestrator.stop();
+    });
+
+    it('produce', async () => {
+        orchestrator.dafkaProducer.produce([
+            {
+                topic,
+                key: 'thekey',
+                value: {data: 'foo'},
+                headers: {
+                    'x-request-id': '123',
+                    'x-b3-traceid': '456',
+                    'x-b3-spanid': '789',
+                    'x-b3-parentspanid': '101112',
+                    'x-b3-sampled': '1',
+                    'x-b3-flags': '1',
+                    'x-ot-span-context': 'foo',
+                },
+            },
+        ]);
+
+        await delay(5000);
+
+        await expect(consume(orchestrator.kafkaClient, topic)).resolves.toMatchSnapshot();
+    });
+});

tests/specs/produceFailure.spec.ts

@@ -0,0 +1,33 @@
+import type {Orchestrator} from '../testcontainers/orchestrator.js';
+import {start} from '../testcontainers/orchestrator.js';
+
+describe('tests', () => {
+    let orchestrator: Orchestrator;
+
+    beforeEach(async () => {
+        orchestrator = await start(
+            {
+                KAFKA_BROKER: 'kafka:9092',
+            },
+            []
+        );
+    }, 5 * 60 * 1000);
+
+    afterEach(async () => {
+        if (!orchestrator) {
+            return;
+        }
+        await orchestrator.stop();
+    });
+
+    it('produce failure', async () => {
+        expect(() =>
+            orchestrator.dafkaProducer.produce([
+                {
+                    topic: 'not exists',
+                    value: {data: 'foo'},
+                },
+            ])
+        ).rejects.toMatchSnapshot();
+    });
+});

tests/testcontainers/dafka.ts

@@ -0,0 +1,43 @@
+import {StartedNetwork, Wait} from 'testcontainers';
+import {GenericContainer} from 'testcontainers';
+import fs from 'node:fs';
+import {withThrow} from '@osskit/fetch-enhancers';
+
+const enhanchedFetch = withThrow(fetch);
+
+export interface ServiceClient {
+    produce: (payload: any) => Promise<Response>;
+}
+
+export const dafka = async (network: StartedNetwork, env: Record<string, string>) => {
+    const container = await new GenericContainer('bazel/src:image')
+        .withExposedPorts(8080)
+        .withNetwork(network)
+        .withEnvironment(env)
+        .withStartupTimeout(parseInt(process.env.STARTUP_TIMEOUT ?? '60000'))
+        .withWaitStrategy(Wait.forHttp('/ready', 8080).forStatusCode(200))
+        .start();
+
+    if (process.env.DEBUG) {
+        try {
+            fs.truncateSync('service.log', 0);
+        } catch (err) {
+            fs.writeFileSync('service.log', '', {flag: 'wx'});
+        }
+        await container.logs().then((logs) => logs.pipe(fs.createWriteStream('service.log')));
+    }
+
+    const baseUrl = `http://localhost:${container.getMappedPort(8080)}`;
+
+    return {
+        stop: () => container.stop(),
+        client: {
+            produce: (payload: any) =>
+                enhanchedFetch(`${baseUrl}/produce`, {
+                    method: 'post',
+                    body: JSON.stringify(payload),
+                    headers: {'content-type': 'application/json'},
+                }),
+        },
+    };
+};