diff --git a/src/modules/pipelinesV2/pipelines.service.spec.ts b/src/modules/pipelinesV2/pipelines.service.spec.ts new file mode 100644 index 0000000..564639e --- /dev/null +++ b/src/modules/pipelinesV2/pipelines.service.spec.ts @@ -0,0 +1,95 @@ +// These imported modules pull in gRPC client-config / service modules that read +// process.env at load time; mock them (hoisted before imports) so the spec needs +// no runtime env. Each mock severs an entire import subtree while still providing +// a class usable as a value/DI token. Mirrors platform-api.controller.spec.ts. +jest.mock('./pipelines-client', () => ({ PipelinesClientConfiguration: class {} })); +jest.mock('../connector/client.service', () => ({ ConnectorClientService: class {} })); +jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} })); +jest.mock('../transformations/transformations.service', () => ({ TransformationsService: class {} })); +jest.mock('../platform-api/platform-api.service', () => ({ PlatformApiService: class {} })); +jest.mock('src/services/nimbus/nimbus.service', () => ({ NimbusService: class {} })); +jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} })); + +import { PipelinesService } from './pipelines.service'; + +const logger = { + info: jest.fn(), + error: jest.fn(), +}; + +const cdcOldInput = { input: { plugin: 'mysql_cdc', tables: [] } }; +const batchOldInput = { input: { plugin: 'mysql', type: 'database', tables: [] } }; + +const updateResponse = { + input: { type: 'database' }, + tablesUpdate: [], + dataAssetUpdate: [], +}; + +const user: any = { customer_modules: [] }; +const updateInputDTO: any = { tables: [] }; +const info: any = { customer: 'cust' }; +const metadata: any = {}; + +function buildService(oldInput: any) { + const inputsService: any = { + findOne: jest.fn().mockResolvedValue(oldInput), + update: jest.fn().mockResolvedValue(updateResponse), + rollbackUpdate: jest.fn().mockResolvedValue({}), + }; + const nimbusService: any = { renameTable: jest.fn().mockResolvedValue({}) }; + + const service = new PipelinesService( + { logger } as any, // dadosferaLogger + {} as any, // grpcClient + {} as any, // connectorService + inputsService, // inputsService + {} as any, // transformationsService + {} as any, // platformAPI + nimbusService, // nimbusService + {} as any, // catalogService + ); + + const updatePlatformJobsSpy = jest + .spyOn(service, 'updatePlatformJobs') + .mockResolvedValue(undefined as any); + + return { service, inputsService, updatePlatformJobsSpy }; +} + +describe('PipelinesService - updatePipelineInput', () => { + afterEach(() => jest.clearAllMocks()); + + it('CDC input skips updatePlatformJobs', async () => { + const { service, inputsService, updatePlatformJobsSpy } = buildService(cdcOldInput); + + const result = await service.updatePipelineInput( + 'pipeline-id', + 'input-id', + updateInputDTO, + info, + user, + metadata, + ); + + expect(updatePlatformJobsSpy).not.toHaveBeenCalled(); + expect(inputsService.update).toHaveBeenCalled(); + expect(result).toBe(updateResponse); + }); + + it('batch input calls updatePlatformJobs', async () => { + const { service, inputsService, updatePlatformJobsSpy } = buildService(batchOldInput); + + await service.updatePipelineInput( + 'pipeline-id', + 'input-id', + updateInputDTO, + info, + user, + metadata, + ); + + expect(updatePlatformJobsSpy).toHaveBeenCalled(); + expect(inputsService.update).toHaveBeenCalled(); + }); +}); diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index b875cb7..992077f 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -433,6 +433,7 @@ export class PipelinesService implements OnModuleInit { }); this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput)); + const isCdc = !!oldInput.plugin?.endsWith('_cdc'); const pipelineIdFormat = pipelineId.split('-').join('_'); const rollback: RollbackPromise[] = []; @@ -493,17 +494,21 @@ export class PipelinesService implements OnModuleInit { } } - try { - await this.updatePlatformJobs( - pipelineIdFormat, - updateInputResponse.input.type, - updateInputDTO, - user - ); - } catch (error) { - this.logger.error(error); - await this.executeRenameRollback(rollback) - throw new Error("Error Platform API updating jobs"); + if (!isCdc) { + try { + await this.updatePlatformJobs( + pipelineIdFormat, + updateInputResponse.input.type, + updateInputDTO, + user + ); + } catch (error) { + this.logger.error(error); + await this.executeRenameRollback(rollback) + throw new Error("Error Platform API updating jobs"); + } + } else { + this.logger.info('CDC input: skipping updatePlatformJobs (batch sync_mode/memory do not apply to CDC jobs)'); } return updateInputResponse;