Skip to content

Commit

Permalink
feat(tasks-manager): assert execOrder queue before consume it
Browse files Browse the repository at this point in the history 
fix(@leav/utils): fix Mockify
  • Loading branch information
@philippechevieux
philippechevieux committed Aug 30, 2024
1 parent 3e458ea commit 4401dad
Show file tree
Hide file tree
Showing 16 changed files with 68 additions and 49 deletions.
60 changes: 35 additions & 25 deletions apps/core/src/domain/tasksManager/tasksManagerDomain.spec.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,15 @@
// License text available at https://www.gnu.org/licenses/lgpl-3.0.txt
import {IAmqpService} from '@leav/message-broker';
import * as amqp from 'amqplib';
import {IEventsManagerDomain} from 'domain/eventsManager/eventsManagerDomain';
import {ITaskRepo} from 'infra/task/taskRepo';
import {IUtils} from 'utils/utils';
import {IConfig} from '_types/config';
import {IEventsManagerDomain} from '../../domain/eventsManager/eventsManagerDomain';
import {ITaskRepo} from '../../infra/task/taskRepo';
import {IUtils} from '../../utils/utils';
import {IConfig} from '../../_types/config';
import {TaskCallbackStatus, TaskStatus} from '../../_types/tasksManager';
import {mockCtx} from '../../__tests__/mocks/shared';
import {mockTask} from '../../__tests__/mocks/task';
import tasksManager from './tasksManagerDomain';
import {Mockify} from '@leav/utils';

const mockAmqpChannel: Mockify<amqp.ConfirmChannel> = {
assertExchange: jest.fn(),
Expand Down Expand Up @@ -79,8 +80,8 @@ describe('Tasks Manager', () => {

await tm.createTask(mockTask, mockCtx);

expect(mockTaskRepo.createTask).toBeCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toBeCalledTimes(1);
expect(mockTaskRepo.createTask).toHaveBeenCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toHaveBeenCalledTimes(1);
});

test('Delete task', async () => {
Expand All @@ -97,7 +98,7 @@ describe('Tasks Manager', () => {

await tm.deleteTasks([mockTask], mockCtx);

expect(mockTaskRepo.deleteTask).toBeCalledTimes(1);
expect(mockTaskRepo.deleteTask).toHaveBeenCalledTimes(1);
});

test('Archive task', async () => {
Expand All @@ -114,8 +115,8 @@ describe('Tasks Manager', () => {

await tm.deleteTasks([{...mockTask, archive: true}], mockCtx);

expect(mockTaskRepo.updateTask).toBeCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toBeCalledTimes(1);
expect(mockTaskRepo.updateTask).toHaveBeenCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toHaveBeenCalledTimes(1);
});

test('Cancel task', async () => {
Expand All @@ -137,8 +138,8 @@ describe('Tasks Manager', () => {

await tm.cancelTask(mockTask, mockCtx);

expect(mockTaskRepo.updateTask).toBeCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toBeCalledTimes(1);
expect(mockTaskRepo.updateTask).toHaveBeenCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toHaveBeenCalledTimes(1);
});

test('Get tasks', async () => {
Expand All @@ -153,7 +154,7 @@ describe('Tasks Manager', () => {

await tm.getTasks({params: {}, ctx: mockCtx});

expect(mockTaskRepo.getTasks).toBeCalledTimes(1);
expect(mockTaskRepo.getTasks).toHaveBeenCalledTimes(1);
});

test('Init Master / Task to execute', async () => {
Expand Down Expand Up @@ -192,8 +193,8 @@ describe('Tasks Manager', () => {

await new Promise(r => setTimeout(r, conf.tasksManager.checkingInterval + 1));

expect(mockAmqpService.consumer.channel.assertQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.assertQueue).toHaveBeenCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toHaveBeenCalledTimes(1);

expect(mockTaskRepo.updateTask).toBeCalledWith(
{id: mockTask.id, status: TaskStatus.PENDING},
Expand Down Expand Up @@ -243,8 +244,8 @@ describe('Tasks Manager', () => {

await new Promise(r => setTimeout(r, conf.tasksManager.checkingInterval + 1));

expect(mockAmqpService.consumer.channel.assertQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.assertQueue).toHaveBeenCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toHaveBeenCalledTimes(1);

expect(mockAmqpService.publish).toBeCalled();

Expand Down Expand Up @@ -290,8 +291,8 @@ describe('Tasks Manager', () => {

await new Promise(r => setTimeout(r, conf.tasksManager.checkingInterval + 1));

expect(mockAmqpService.consumer.channel.assertQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.assertQueue).toHaveBeenCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toHaveBeenCalledTimes(1);

expect(mockTaskRepo.updateTask).toBeCalledWith(
{id: mockTask.id, callbacks: [{...mockTask.callbacks[0], status: TaskCallbackStatus.RUNNING}]},
Expand Down Expand Up @@ -321,9 +322,18 @@ describe('Tasks Manager', () => {

await tm.initWorker();

expect(mockAmqpService.consume).toBeCalledTimes(2);
expect(mockAmqpService.consumer.channel.assertQueue).toBeCalledTimes(1);
expect(mockAmqpService.consumer.channel.bindQueue).toBeCalledTimes(1);
expect(mockAmqpService.consume).toHaveBeenCalledTimes(2);
expect(mockAmqpService.consumer.channel.assertQueue).toHaveBeenCalledTimes(2);
expect(mockAmqpService.consumer.channel.assertQueue).toHaveBeenNthCalledWith(
1,
conf.tasksManager?.queues.execOrders
);
expect(mockAmqpService.consumer.channel.assertQueue).toHaveBeenNthCalledWith(
2,
expect.stringMatching(conf.tasksManager?.queues.cancelOrders),
{autoDelete: true, durable: false, exclusive: true}
);
expect(mockAmqpService.consumer.channel.bindQueue).toHaveBeenCalledTimes(1);
});

test('Update progress', async () => {
Expand All @@ -344,12 +354,12 @@ describe('Tasks Manager', () => {
mockCtx
);

expect(mockTaskRepo.getTasks).toBeCalledTimes(1);
expect(mockTaskRepo.getTasks).toHaveBeenCalledTimes(1);
expect(mockTaskRepo.updateTask).toBeCalledWith(
{id: mockTask.id, progress: {percent: 55, description: {fr: 'description', en: 'description'}}},
mockCtx
);
expect(mockEventsManager.sendPubSubEvent).toBeCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toHaveBeenCalledTimes(1);

await tm.updateProgress(
mockTask.id,
Expand Down Expand Up @@ -377,8 +387,8 @@ describe('Tasks Manager', () => {

await tm.setLink(mockTask.id, {name: 'name', url: 'url'}, mockCtx);

expect(mockTaskRepo.getTasks).toBeCalledTimes(1);
expect(mockTaskRepo.getTasks).toHaveBeenCalledTimes(1);
expect(mockTaskRepo.updateTask).toBeCalledWith({id: mockTask.id, link: {name: 'name', url: 'url'}}, mockCtx);
expect(mockEventsManager.sendPubSubEvent).toBeCalledTimes(1);
expect(mockEventsManager.sendPubSubEvent).toHaveBeenCalledTimes(1);
});
});
3 changes: 2 additions & 1 deletion apps/core/src/domain/tasksManager/tasksManagerDomain.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -457,6 +457,8 @@ export default function ({
};

const _listenExecOrders = async () => {
await amqpService.consumer.channel.assertQueue(config.tasksManager.queues.execOrders);

await amqpService.consume(
config.tasksManager.queues.execOrders,
config.tasksManager.routingKeys.execOrders,
Expand Down Expand Up @@ -499,7 +501,6 @@ export default function ({
config.amqp.exchange,
config.tasksManager.routingKeys.execOrders
);

return _monitorTasks({
userId: config.defaultUserId,
queryId: 'TasksManagerDomain'
Expand Down
2 changes: 1 addition & 1 deletion libs/ui/package.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@
"dependencies": {
"@ckeditor/ckeditor5-build-inline": "39.0.1",
"@ckeditor/ckeditor5-react": "6.1.0",
"@leav/utils": "0.0.2",
"@leav/utils": "0.0.3",
"dompurify": "3.0.5",
"html-react-parser": "4.2.2",
"i18next-resources-to-backend": "1.1.4",
Expand Down
1 change: 1 addition & 0 deletions libs/utils/dist/cjs/types/forms.d.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ export declare enum TabsDirection {
export interface ICommonFieldsSettings {
label?: Record<string, string | null>;
attribute?: string;
useAttributeLabel?: boolean;
}
export interface IRequiredFieldsSettings extends ICommonFieldsSettings {
required?: boolean;
Expand Down
2 changes: 1 addition & 1 deletion libs/utils/dist/cjs/types/forms.js.map
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 3 additions & 1 deletion libs/utils/dist/cjs/types/helpers.d.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,5 +10,7 @@ export declare type WithTypename<T> = {
readonly __typename?: string;
};
export declare type Mockify<T> = {
[P in keyof T]?: T[P] extends (...args: any) => any ? jest.Mock : T[P];
[P in keyof T]?: T[P] extends (...args: any[]) => any
? jest.Mock<ReturnType<T[P]> extends never ? never : any>
: T[P];
};
11 changes: 5 additions & 6 deletions libs/utils/dist/cjs/utils.js
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading

0 comments on commit 4401dad

Please sign in to comment.