Add unit tests for Subscription service

Author: Ugo Plouviez

src/main/java/com/iexec/worker/pubsub/SubscriptionService.java

@@ -36,12 +36,11 @@
 @Service
 public class SubscriptionService extends StompSessionHandlerAdapter {
 
-    private final String coreHost;
-    private final int corePort;
-    private final String workerWalletAddress;
     private RestTemplate restTemplate;
     // external services
     private TaskExecutorService taskExecutorService;
+    private CoreConfigurationService coreConfigurationService;
+    private WorkerConfigurationService workerConfigurationService;
 
     // internal components
     private StompSession session;
@@ -54,22 +53,23 @@ public SubscriptionService(CoreConfigurationService coreConfigurationService,
                                TaskExecutorService taskExecutorService,
                                RestTemplate restTemplate) {
         this.taskExecutorService = taskExecutorService;
-
-        this.coreHost = coreConfigurationService.getHost();
-        this.corePort = coreConfigurationService.getPort();
-        this.workerWalletAddress = workerConfigurationService.getWorkerWalletAddress();
         this.restTemplate = restTemplate;
-
+        this.coreConfigurationService = coreConfigurationService;
+        this.workerConfigurationService = workerConfigurationService;
         chainTaskIdToSubscription = new ConcurrentHashMap<>();
-        url = "http://" + coreHost + ":" + corePort + "/connect";
+
     }
 
     @PostConstruct
-    private void run() {
+    void init() {
+        String coreHost = coreConfigurationService.getHost();
+        int corePort = coreConfigurationService.getPort();
+        this.url = "http://" + coreHost + ":" + corePort + "/connect";
+
         this.restartStomp();
     }
 
-    private void restartStomp() {
+    void restartStomp() {
         log.info("Starting STOMP");
         if (isConnectEndpointUp()) {
             WebSocketClient webSocketClient = new StandardWebSocketClient();
@@ -90,7 +90,7 @@ private boolean isConnectEndpointUp() {
         if (checkConnectionEntity.getStatusCode().is2xxSuccessful()) {
             return true;
         }
-        log.error("isConnectEndpointUp failed (will retry) [url:{}, status:]", url, checkConnectionEntity.getStatusCode());
+        log.error("isConnectEndpointUp failed (will retry) [url:{}, status:{}]", url, checkConnectionEntity.getStatusCode());
         try {
             Thread.sleep(1000);
         } catch (InterruptedException e) {
@@ -173,7 +173,7 @@ public void unsubscribeFromTopic(String chainTaskId) {
     }
 
     public void handleTaskNotification(TaskNotification notif) {
-        if (notif.getWorkersAddress().contains(workerWalletAddress)
+        if (notif.getWorkersAddress().contains(workerConfigurationService.getWorkerWalletAddress())
                 || notif.getWorkersAddress().isEmpty()) {
             log.info("Received notification [notification:{}]", notif);
 

src/test/java/com/iexec/worker/amnesia/AmnesiaRecoveryServiceTests.java

@@ -6,7 +6,6 @@
 import com.iexec.common.notification.TaskNotificationType;
 import com.iexec.common.task.TaskDescription;
 import com.iexec.worker.chain.IexecHubService;
-import com.iexec.worker.executor.TaskExecutorService;
 import com.iexec.worker.feign.CustomFeignClient;
 import com.iexec.worker.pubsub.SubscriptionService;
 import com.iexec.worker.result.ResultService;

src/test/java/com/iexec/worker/pubsub/SubscriptionServiceTests.java

@@ -0,0 +1,123 @@
+package com.iexec.worker.pubsub;
+
+import com.iexec.common.chain.ContributionAuthorization;
+import com.iexec.common.notification.TaskNotification;
+import com.iexec.common.notification.TaskNotificationExtra;
+import com.iexec.common.notification.TaskNotificationType;
+import com.iexec.worker.config.WorkerConfigurationService;
+import com.iexec.worker.executor.TaskExecutorService;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import java.util.Collections;
+
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.when;
+
+public class SubscriptionServiceTests {
+
+    @Mock
+    private WorkerConfigurationService workerConfigurationService;
+    @Mock
+    private TaskExecutorService taskExecutorService;
+
+    @InjectMocks
+    SubscriptionService subscriptionService;
+
+    private static final String WORKER_WALLET_ADDRESS = "0x1234";
+    private static final String CHAIN_TASK_ID = "chaintaskid";
+
+    private TaskNotification notifTemplate = TaskNotification.builder()
+            .workersAddress(Collections.singletonList(WORKER_WALLET_ADDRESS))
+            .taskNotificationExtra(TaskNotificationExtra.builder()
+                    .build())
+            .chainTaskId(CHAIN_TASK_ID)
+            .build();
+
+    @Before
+    public void init() {
+        MockitoAnnotations.initMocks(this);
+
+        when(workerConfigurationService.getWorkerWalletAddress()).thenReturn(WORKER_WALLET_ADDRESS);
+
+
+    }
+
+
+    @Test
+    public void shouldNotHandleNotificationSinceWorkerWalletNotInNotif() {
+
+        TaskNotification notif = TaskNotification.builder()
+                .workersAddress(Collections.singletonList("0xabcd"))
+                .chainTaskId(CHAIN_TASK_ID)
+                .build();
+
+        subscriptionService.handleTaskNotification(notif);
+
+        Mockito.verifyZeroInteractions(taskExecutorService);
+    }
+
+    @Test
+    public void shouldNotContributeSinceNoContributionAuthorization() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_CONTRIBUTE);
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verifyZeroInteractions(taskExecutorService);
+    }
+
+    @Test
+    public void shouldTryToContribute() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_CONTRIBUTE);
+        notifTemplate.setTaskNotificationExtra(TaskNotificationExtra.builder()
+                        .contributionAuthorization(new ContributionAuthorization())
+                        .build());
+
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verify(taskExecutorService, Mockito.times(1)).tryToContribute(any());
+    }
+
+    @Test
+    public void shouldAbortOnContributionTimeout() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_ABORT_CONTRIBUTION_TIMEOUT);
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verify(taskExecutorService, Mockito.times(1)).abortContributionTimeout(CHAIN_TASK_ID);
+    }
+
+    @Test
+    public void shouldAbortOnConsensusReached() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_ABORT_CONSENSUS_REACHED);
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verify(taskExecutorService, Mockito.times(1)).abortConsensusReached(CHAIN_TASK_ID);
+    }
+
+    @Test
+    public void shouldReveal() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_REVEAL);
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verify(taskExecutorService, Mockito.times(1)).reveal(eq(CHAIN_TASK_ID), anyLong());
+    }
+
+    @Test
+    public void shouldUpload() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_UPLOAD);
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verify(taskExecutorService, Mockito.times(1)).uploadResult(CHAIN_TASK_ID);
+    }
+
+    @Test
+    public void shouldComplete() {
+        notifTemplate.setTaskNotificationType(TaskNotificationType.PLEASE_COMPLETE);
+        subscriptionService.handleTaskNotification(notifTemplate);
+
+        Mockito.verify(taskExecutorService, Mockito.times(1)).completeTask(CHAIN_TASK_ID);
+    }
+}