Adding missing tests

Author: CarlosGamero

packages/kafka/test/publisher/PermissionPublisher.spec.ts

@@ -1,5 +1,11 @@
 import { InternalError } from '@lokalise/node-core'
-import { type PermissionAdded, type PermissionRemoved, TOPICS } from '../utils/permissionSchemas.js'
+import {
+  PERMISSION_ADDED_SCHEMA,
+  PERMISSION_GENERAL_TOPIC,
+  type PermissionAdded,
+  type PermissionRemoved,
+  TOPICS,
+} from '../utils/permissionSchemas.js'
 import { type TestContext, createTestContext } from '../utils/testContext.js'
 import { PermissionPublisher } from './PermissionPublisher.js'
 
@@ -96,14 +102,22 @@ describe('PermissionPublisher - init', () => {
         expect(emittedEvent.message).toMatchObject({ id: '1', type: 'added' })
       },
     )
+
+    it('should fail if message type is not supported and a topic has different schemas', () => {
+      // Given
+      expect(
+        () => new PermissionPublisher(testContext.cradle, { supportMessageTypes: false }),
+      ).toThrowErrorMatchingInlineSnapshot(
+        '[Error: if messageTypeField is not provided, messageSchemas must have a single schema]',
+      )
+    })
   })
 
   describe('publish', () => {
-    beforeEach(() => {
+    it('should fail if topic is not supported', async () => {
+      // Given
       publisher = new PermissionPublisher(testContext.cradle)
-    })
 
-    it('should fail if topic is not supported', async () => {
       // When
       await expect(
         publisher.publish('bad topic' as any, {} as any), // Intentionally bad topic to force the error
@@ -114,6 +128,8 @@ describe('PermissionPublisher - init', () => {
 
     it('should fail if there is no schema for message type', async () => {
       // Given
+      publisher = new PermissionPublisher(testContext.cradle)
+
       const message = {
         id: '1',
         type: 'bad' as any, // Intentionally bad type to force the error
@@ -128,6 +144,8 @@ describe('PermissionPublisher - init', () => {
 
     it('should fail if message does not match schema', async () => {
       // Given
+      publisher = new PermissionPublisher(testContext.cradle)
+
       const message = {
         id: 1 as unknown as string,
         type: 'added',
@@ -154,6 +172,8 @@ describe('PermissionPublisher - init', () => {
 
     it('should publish messages', async () => {
       // Given
+      publisher = new PermissionPublisher(testContext.cradle)
+
       const message1 = {
         id: '1',
         type: 'added',
@@ -176,5 +196,39 @@ describe('PermissionPublisher - init', () => {
       const emittedEvent2 = await publisher.handlerSpy.waitForMessageWithId('2', 'published')
       expect(emittedEvent2.message).toMatchObject(message2)
     })
+
+    it('should publish only messages meeting schema', async () => {
+      // Given
+      publisher = new PermissionPublisher(testContext.cradle, {
+        supportMessageTypes: false,
+        topicsConfig: [
+          {
+            topic: PERMISSION_GENERAL_TOPIC,
+            schemas: [PERMISSION_ADDED_SCHEMA],
+          },
+        ] as any, // we are not adding the other topics intentionally
+      })
+
+      const messageValid = {
+        id: '1',
+        type: 'added',
+        permissions: [],
+      } satisfies PermissionAdded
+      const messageInvalid = {
+        id: '2',
+        type: 'removed',
+        permissions: [],
+      } satisfies PermissionRemoved
+
+      // When&Then - valid
+      await publisher.publish(PERMISSION_GENERAL_TOPIC, messageValid)
+      const emittedEvent = await publisher.handlerSpy.waitForMessageWithId('1', 'published')
+      expect(emittedEvent.message).toMatchObject(messageValid)
+
+      // When&Then - invalid
+      await expect(
+        publisher.publish(PERMISSION_GENERAL_TOPIC, messageInvalid),
+      ).rejects.toThrowErrorMatchingInlineSnapshot('[Error: Unsupported message type: removed]')
+    })
   })
 })

packages/kafka/test/publisher/PermissionPublisher.ts

@@ -11,7 +11,10 @@ type PermissionPublisherOptions = Partial<
     KafkaPublisherOptions<typeof PERMISSION_TOPIC_MESSAGES_CONFIG>,
     'handlerSpy' | 'kafka' | 'autocreateTopics' | 'topicsConfig'
   >
->
+> & {
+  supportMessageTypes?: boolean
+}
+
 export class PermissionPublisher extends AbstractKafkaPublisher<
   typeof PERMISSION_TOPIC_MESSAGES_CONFIG
 > {
@@ -22,8 +25,8 @@ export class PermissionPublisher extends AbstractKafkaPublisher<
       autocreateTopics: options.autocreateTopics ?? true,
       handlerSpy: options?.handlerSpy ?? true,
       logMessages: true,
-      messageIdField: 'id',
-      messageTypeField: 'type',
+      messageIdField: options.supportMessageTypes === false ? undefined : 'id',
+      messageTypeField: options.supportMessageTypes === false ? undefined : 'type',
     })
   }
 }

packages/kafka/test/utils/permissionSchemas.ts

@@ -16,9 +16,9 @@ export const PERMISSION_REMOVED_SCHEMA = BASE_SCHEMA.extend({
 })
 export type PermissionRemoved = z.infer<typeof PERMISSION_REMOVED_SCHEMA>
 
-const PERMISSION_ADDED_TOPIC = 'permission-added'
-const PERMISSION_REMOVED_TOPIC = 'permission-removed'
-const PERMISSION_GENERAL_TOPIC = 'permission-general'
+export const PERMISSION_ADDED_TOPIC = 'permission-added'
+export const PERMISSION_REMOVED_TOPIC = 'permission-removed'
+export const PERMISSION_GENERAL_TOPIC = 'permission-general'
 export const TOPICS = [PERMISSION_ADDED_TOPIC, PERMISSION_REMOVED_TOPIC, PERMISSION_GENERAL_TOPIC]
 
 export const PERMISSION_TOPIC_MESSAGES_CONFIG = [