Use mqtt service instead of calling GET /status every 5 seconds.

framework/python/src/common/session.py

@@ -17,6 +17,7 @@
 import pytz
 import json
 import os
+from fastapi.encoders import jsonable_encoder
 from common import util, logger, mqtt
 from common.risk_profile import RiskProfile
 from net_orc.ip_control import IPControl
@@ -52,7 +53,10 @@ def wrapper(self, *args, **kwargs):
     result = method(self, *args, **kwargs)
 
     if self.get_status() != 'Idle':
-      self.get_mqtt_client().send_message(STATUS_TOPIC, self.to_json())
+      self.get_mqtt_client().send_message(
+                                        STATUS_TOPIC,
+                                        jsonable_encoder(self.to_json())
+                                        )
 
     return result
   return wrapper
@@ -344,7 +348,7 @@ def get_report_tests(self):
     return {'total': self.get_total_tests(), 'results': test_results}
 
   def add_test_result(self, result):
-    LOGGER.info('------adding resul----t')
+
     updated = False
 
     # Check if test has already been added

modules/ui/src/app/model/topic.ts

@@ -1,6 +1,7 @@
 export enum Topic {
   NetworkAdapters = 'events/adapter',
   InternetConnection = 'events/internet',
+  Status = 'status',
 }
 
 export interface InternetConnection {

modules/ui/src/app/services/test-run-mqtt.service.spec.ts

@@ -8,6 +8,7 @@ import { of } from 'rxjs';
 import { MOCK_ADAPTERS } from '../mocks/settings.mock';
 import { Topic } from '../model/topic';
 import { MOCK_INTERNET } from '../mocks/topic.mock';
+import { MOCK_PROGRESS_DATA_IN_PROGRESS } from '../mocks/testrun.mock';
 
 describe('TestRunMqttService', () => {
   let service: TestRunMqttService;
@@ -69,6 +70,28 @@ describe('TestRunMqttService', () => {
     });
   });
 
+  describe('getStatus', () => {
+    beforeEach(() => {
+      mockService.observe.and.returnValue(
+        of(getResponse(MOCK_PROGRESS_DATA_IN_PROGRESS))
+      );
+    });
+
+    it('should subscribe the topic', done => {
+      service.getStatus().subscribe(() => {
+        expect(mockService.observe).toHaveBeenCalledWith(Topic.Status);
+        done();
+      });
+    });
+
+    it('should return object of type', done => {
+      service.getStatus().subscribe(res => {
+        expect(res).toEqual(MOCK_PROGRESS_DATA_IN_PROGRESS);
+        done();
+      });
+    });
+  });
+
   function getResponse<Type>(response: Type): IMqttMessage {
     const enc = new TextEncoder();
     const message = enc.encode(JSON.stringify(response));

modules/ui/src/app/services/test-run-mqtt.service.ts

@@ -3,6 +3,7 @@ import { IMqttMessage, MqttService } from 'ngx-mqtt';
 import { catchError, Observable, of } from 'rxjs';
 import { map } from 'rxjs/operators';
 import { Adapters } from '../model/setting';
+import { TestrunStatus } from '../model/testrun-status';
 import { InternetConnection, Topic } from '../model/topic';
 
 @Injectable({
@@ -19,6 +20,10 @@ export class TestRunMqttService {
     return this.topic<InternetConnection>(Topic.InternetConnection);
   }
 
+  getStatus(): Observable<TestrunStatus> {
+    return this.topic<TestrunStatus>(Topic.Status);
+  }
+
   private topic<Type>(topicName: string): Observable<Type> {
     return this.mqttService.observe(topicName).pipe(
       map(

modules/ui/src/app/store/effects.spec.ts

@@ -42,6 +42,7 @@ import {
 } from '../mocks/testrun.mock';
 import {
   fetchSystemStatus,
+  fetchSystemStatusSuccess,
   setReports,
   setStatus,
   setTestrunStatus,
@@ -52,6 +53,7 @@ import { throwError } from 'rxjs/internal/observable/throwError';
 import { HttpErrorResponse } from '@angular/common/http';
 import { IDLE_STATUS } from '../model/testrun-status';
 import { HISTORY } from '../mocks/reports.mock';
+import { TestRunMqttService } from '../services/test-run-mqtt.service';
 
 describe('Effects', () => {
   let actions$ = new Observable<Action>();
@@ -64,6 +66,8 @@ describe('Effects', () => {
       'dismissWithTimout',
       'openSnackBar',
     ]);
+  const mockMqttService: jasmine.SpyObj<TestRunMqttService> =
+    jasmine.createSpyObj('mockMqttService', ['getStatus']);
 
   beforeEach(() => {
     testRunServiceMock = jasmine.createSpyObj('testRunServiceMock', [
@@ -85,11 +89,16 @@ describe('Effects', () => {
     testRunServiceMock.fetchProfiles.and.returnValue(of([]));
     testRunServiceMock.getHistory.and.returnValue(of([]));
 
+    mockMqttService.getStatus.and.returnValue(
+      of(MOCK_PROGRESS_DATA_IN_PROGRESS)
+    );
+
     TestBed.configureTestingModule({
       providers: [
         AppEffects,
         { provide: TestRunService, useValue: testRunServiceMock },
         { provide: NotificationService, useValue: notificationServiceMock },
+        { provide: TestRunMqttService, useValue: mockMqttService },
         provideMockActions(() => actions$),
         provideMockStore({}),
       ],
@@ -399,14 +408,15 @@ describe('Effects', () => {
         );
       });
 
-      it('should call fetchSystemStatus for status "in progress"', fakeAsync(() => {
+      it('should call fetchSystemStatus for status "in progress"', () => {
         effects.onFetchSystemStatusSuccess$.subscribe(() => {
-          tick(5000);
-
-          expect(dispatchSpy).toHaveBeenCalledWith(fetchSystemStatus());
-          discardPeriodicTasks();
+          expect(dispatchSpy).toHaveBeenCalledWith(
+            fetchSystemStatusSuccess({
+              systemStatus: MOCK_PROGRESS_DATA_IN_PROGRESS,
+            })
+          );
         });
-      }));
+      });
 
       it('should dispatch status and systemStatus', done => {
         effects.onFetchSystemStatusSuccess$.subscribe(() => {
@@ -451,14 +461,15 @@ describe('Effects', () => {
         );
       });
 
-      it('should call fetchSystemStatus for status "waiting for device"', fakeAsync(() => {
+      it('should call fetchSystemStatus for status "waiting for device"', () => {
         effects.onFetchSystemStatusSuccess$.subscribe(() => {
-          tick(5000);
-
-          expect(dispatchSpy).toHaveBeenCalledWith(fetchSystemStatus());
-          discardPeriodicTasks();
+          expect(dispatchSpy).toHaveBeenCalledWith(
+            fetchSystemStatusSuccess({
+              systemStatus: MOCK_PROGRESS_DATA_IN_PROGRESS,
+            })
+          );
         });
-      }));
+      });
 
       it('should open snackbar when waiting for device is too long', fakeAsync(() => {
         effects.onFetchSystemStatusSuccess$.subscribe(() => {

modules/ui/src/app/store/effects.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { Injectable, NgZone } from '@angular/core';
+import { Injectable } from '@angular/core';
 import { Actions, createEffect, ofType } from '@ngrx/effects';
 import { Store } from '@ngrx/store';
 import { map, switchMap, tap, withLatestFrom } from 'rxjs/operators';
@@ -25,12 +25,12 @@ import { TestRunService } from '../services/test-run.service';
 import {
   filter,
   combineLatest,
-  interval,
   Subject,
   timer,
   take,
   catchError,
   EMPTY,
+  Subscription,
 } from 'rxjs';
 import {
   selectIsOpenWaitSnackBar,
@@ -46,6 +46,7 @@ import {
 } from '../model/testrun-status';
 import {
   fetchSystemStatus,
+  fetchSystemStatusSuccess,
   setReports,
   setStatus,
   setTestrunStatus,
@@ -54,13 +55,13 @@ import {
 import { takeUntil } from 'rxjs/internal/operators/takeUntil';
 import { NotificationService } from '../services/notification.service';
 import { Profile } from '../model/profile';
+import { TestRunMqttService } from '../services/test-run-mqtt.service';
 
 const WAIT_TO_OPEN_SNACKBAR_MS = 60 * 1000;
 
 @Injectable()
 export class AppEffects {
-  private startInterval = false;
-  private destroyInterval$: Subject<boolean> = new Subject<boolean>();
+  private statusSubscription: Subscription | undefined;
   private destroyWaitDeviceInterval$: Subject<boolean> = new Subject<boolean>();
 
   checkInterfacesInConfig$ = createEffect(() =>
@@ -206,8 +207,7 @@ export class AppEffects {
       return this.actions$.pipe(
         ofType(AppActions.stopInterval),
         tap(() => {
-          this.startInterval = false;
-          this.destroyInterval$.next(true);
+          this.statusSubscription?.unsubscribe();
         })
       );
     },
@@ -219,10 +219,7 @@ export class AppEffects {
       return this.actions$.pipe(
         ofType(AppActions.fetchSystemStatusSuccess),
         tap(({ systemStatus }) => {
-          if (
-            this.testrunService.testrunInProgress(systemStatus.status) &&
-            !this.startInterval
-          ) {
+          if (this.testrunService.testrunInProgress(systemStatus.status)) {
             this.pullingSystemStatusData();
           } else if (
             !this.testrunService.testrunInProgress(systemStatus.status)
@@ -251,24 +248,20 @@ export class AppEffects {
         tap(([{ systemStatus }, , status]) => {
           // for app - requires only status
           if (systemStatus.status !== status?.status) {
-            this.ngZone.run(() => {
-              this.store.dispatch(setStatus({ status: systemStatus.status }));
-              this.store.dispatch(
-                setTestrunStatus({ systemStatus: systemStatus })
-              );
-            });
+            this.store.dispatch(setStatus({ status: systemStatus.status }));
+            this.store.dispatch(
+              setTestrunStatus({ systemStatus: systemStatus })
+            );
           } else if (
             systemStatus.finished !== status?.finished ||
             (systemStatus.tests as TestsData)?.results?.length !==
               (status?.tests as TestsData)?.results?.length ||
             (systemStatus.tests as IResult[])?.length !==
               (status?.tests as IResult[])?.length
           ) {
-            this.ngZone.run(() => {
-              this.store.dispatch(
-                setTestrunStatus({ systemStatus: systemStatus })
-              );
-            });
+            this.store.dispatch(
+              setTestrunStatus({ systemStatus: systemStatus })
+            );
           }
         })
       );
@@ -353,22 +346,23 @@ export class AppEffects {
   }
 
   private pullingSystemStatusData(): void {
-    this.ngZone.runOutsideAngular(() => {
-      this.startInterval = true;
-      interval(5000)
-        .pipe(
-          takeUntil(this.destroyInterval$),
-          tap(() => this.store.dispatch(fetchSystemStatus()))
-        )
-        .subscribe();
-    });
+    if (
+      this.statusSubscription === undefined ||
+      this.statusSubscription?.closed
+    ) {
+      this.statusSubscription = this.testrunMqttService
+        .getStatus()
+        .subscribe(systemStatus => {
+          this.store.dispatch(fetchSystemStatusSuccess({ systemStatus }));
+        });
+    }
   }
 
   constructor(
     private actions$: Actions,
     private testrunService: TestRunService,
+    private testrunMqttService: TestRunMqttService,
     private store: Store<AppState>,
-    private ngZone: NgZone,
     private notificationService: NotificationService
   ) {}
 }