import { of } from 'rxjs'; import { InputsService } from './inputs.service'; import DadosferaLogger from '@dadosfera/dadosfera-logs/dist'; import { CreateCdcInputReq } from './dtos/input.model'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; const info = { customer_id: 'cid', user_id: 'u' } as unknown as Info; describe('InputsService.createCdc', () => { let service: InputsService; let inputCreateCdcMock: jest.Mock; beforeEach(async () => { inputCreateCdcMock = jest .fn() .mockImplementation((req) => of({ input: req.input })); const grpcClient: any = { getService: jest.fn().mockReturnValue({ InputCreateCdc: inputCreateCdcMock, }), }; service = new InputsService(new DadosferaLogger(), grpcClient); await service.onModuleInit(); }); it('forwards destination and per-table iceberg_table_name to the gRPC request', async () => { const body: CreateCdcInputReq = { name: 'CDC Iceberg Test', plugin: 'mysql_cdc', read_only: true, destination: { iceberg: { namespace: 'cdc_raw' } }, tables: [ { name: 'orders', table_schema: 'mydb', primary_keys: ['id'], iceberg_table_name: 'cdc_raw.mydb__orders', }, ], }; await service.createCdc({ body, info }); expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); const sentRequest = inputCreateCdcMock.mock.calls[0][0]; expect(sentRequest.input).toEqual( expect.objectContaining({ destination: { iceberg: { namespace: 'cdc_raw' } }, }), ); expect(sentRequest.input.tables[0]).toEqual( expect.objectContaining({ iceberg_table_name: 'cdc_raw.mydb__orders', }), ); }); it('forwards per-table columns to the gRPC request', async () => { const body: CreateCdcInputReq = { name: 'CDC Columns Test', plugin: 'mysql_cdc', read_only: true, tables: [ { name: 'orders', table_schema: 'mydb', primary_keys: ['id'], columns: [ { name: 'id', type: 'int', is_primary_key: true }, { name: 'descr', type: 'varchar(255)', is_primary_key: false }, ], }, ], }; await service.createCdc({ body, info }); expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); const sentRequest = inputCreateCdcMock.mock.calls[0][0]; expect(sentRequest.input.tables[0].columns).toEqual([ { name: 'id', type: 'int', is_primary_key: true }, { name: 'descr', type: 'varchar(255)', is_primary_key: false }, ]); }); it('back-compat: a body with no destination sends destination undefined, not an error', async () => { const body: CreateCdcInputReq = { name: 'CDC Legacy Test', plugin: 'mysql_cdc', read_only: true, tables: [ { name: 'pedidos', table_schema: 'cadastros', primary_keys: ['id'], }, ], }; const result = await service.createCdc({ body, info }); expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); const sentRequest = inputCreateCdcMock.mock.calls[0][0]; expect(sentRequest.input.destination).toBeUndefined(); expect(sentRequest.input.tables[0].iceberg_table_name).toBeUndefined(); expect(result.input).toBeDefined(); }); });