JobServiceTest.java

/*
 * This file is part of vospace-rest
 * Copyright (C) 2021 Istituto Nazionale di Astrofisica
 * SPDX-License-Identifier: GPL-3.0-or-later
 */
package it.inaf.oats.vospace;

import it.inaf.oats.vospace.datamodel.Views;
import it.inaf.oats.vospace.exception.NodeBusyException;
import it.inaf.oats.vospace.persistence.JobDAO;
import java.util.ArrayList;
import java.util.List;
import javax.servlet.http.HttpServletRequest;
import net.ivoa.xml.uws.v1.ExecutionPhase;
import net.ivoa.xml.uws.v1.JobSummary;
import net.ivoa.xml.vospace.v2.Protocol;
import net.ivoa.xml.vospace.v2.Transfer;
import net.ivoa.xml.vospace.v2.View;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;

@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
public class JobServiceTest {

    @Mock
    private JobDAO jobDAO;

    @Mock
    private UriService uriService;

    @Mock
    private AsyncTransferService asyncTransfService;

    @Mock
    private HttpServletRequest servletRequest;

    @Mock
    private MoveService moveService;
    
    @Mock
    private CopyService copyService;
    
    @Mock
    private FileServiceClient fileServiceClient;

    @InjectMocks
    private JobService jobService;

    @BeforeEach
    public void setUp() {
        when(servletRequest.getRequestURL()).thenReturn(new StringBuffer("http://localhost/vospace/transfer"));
        when(servletRequest.getContextPath()).thenReturn("/vospace");
    }

    @Test
    public void testStartJobDefault() {
        when(uriService.getTransfer(any())).thenReturn(getPullFromVoSpaceHttpTransfer());

        JobSummary job = new JobSummary();
        jobService.setJobPhase(job, "RUN");

        verify(jobDAO, times(2)).updateJob(eq(job), any());
    }

    @Test
    public void testStartJobTape() {
        when(uriService.getTransfer(any())).thenReturn(getTapeTransfer());

        JobSummary job = new JobSummary();
        jobService.setJobPhase(job, "RUN");

        verify(jobDAO, times(2)).updateJob(eq(job), any());
    }

    @Test
    public void testStartJobVoSpaceError() {
        when(uriService.getTransfer(any())).thenReturn(getTapeTransfer());

        when(asyncTransfService.startJob(any())).thenThrow(new NodeBusyException("/foo"));

        JobSummary job = new JobSummary();
        jobService.setJobPhase(job, "RUN");

        verify(jobDAO, times(2)).updateJob(argThat(j -> ExecutionPhase.ERROR.equals(j.getPhase())), any());
    }

    @Test
    public void testStartJobUnexpectedError() {
        when(uriService.getTransfer(any())).thenReturn(getTapeTransfer());

        when(asyncTransfService.startJob(any())).thenThrow(new NullPointerException());

        JobSummary job = new JobSummary();
        jobService.setJobPhase(job, "RUN");

        verify(jobDAO, times(2)).updateJob(argThat(j -> ExecutionPhase.ERROR.equals(j.getPhase())), any());
    }

    @Test
    public void testSyncJobResultVoSpaceError() {
        Transfer transfer = getPullFromVoSpaceHttpTransfer();
        assertFalse(transfer.getProtocols().isEmpty());
        when(uriService.getTransfer(any())).thenReturn(transfer);
        doThrow(new NodeBusyException("/foo")).when(uriService).getNegotiatedTransfer(any(), any());
        jobService.createSyncJobResult(new JobSummary());
        verify(jobDAO, times(1)).createJob(argThat(j -> ExecutionPhase.ERROR.equals(j.getPhase())), any());
        assertTrue(transfer.getProtocols().isEmpty());
    }

    @Test
    public void testSyncJobResultUnexpectedError() {
        Transfer transfer = getPullFromVoSpaceHttpTransfer();
        assertFalse(transfer.getProtocols().isEmpty());
        when(uriService.getTransfer(any())).thenReturn(transfer);
        doThrow(new NullPointerException()).when(uriService).getNegotiatedTransfer(any(), any());
        jobService.createSyncJobResult(new JobSummary());
        verify(jobDAO, times(1)).createJob(argThat(j -> ExecutionPhase.ERROR.equals(j.getPhase())), any());
        assertTrue(transfer.getProtocols().isEmpty());
    }

    @Test
    public void testSyncJobResultErrorAfterNegotiatedTransfer() {
        Transfer transfer = getPullFromVoSpaceHttpTransfer();
        assertFalse(transfer.getProtocols().isEmpty());
        when(uriService.getTransfer(any())).thenReturn(transfer);

        Transfer negotiatedTransfer = getPullFromVoSpaceHttpTransfer();
        assertFalse(negotiatedTransfer.getProtocols().isEmpty());
        when(uriService.getNegotiatedTransfer(any(), any())).thenReturn(negotiatedTransfer);

        doThrow(new NullPointerException()).when(servletRequest).getContextPath();
        jobService.createSyncJobResult(new JobSummary());

        verify(jobDAO, times(1)).createJob(argThat(j -> ExecutionPhase.ERROR.equals(j.getPhase())), any());
        assertTrue(transfer.getProtocols().isEmpty());
        assertTrue(negotiatedTransfer.getProtocols().isEmpty());
    }

    @Test
    public void testStartJobSetQueuedPhaseForAsyncRecall() {

        Transfer asyncRecallTransfer = getTapeTransfer();

        JobSummary job = new JobSummary();
        setJobInfo(job, asyncRecallTransfer);

        when(uriService.getTransfer(any())).thenReturn(asyncRecallTransfer);

        when(asyncTransfService.startJob(any())).thenReturn(job);

        jobService.setJobPhase(job, "RUN");

        // Job will be executed by transfer service
        assertEquals(ExecutionPhase.QUEUED, job.getPhase());
    }

    @Test
    public void testStartJobSetExecutingPhaseForAsyncPullFromVoSpace() {

        Transfer httpTransfer = getPullFromVoSpaceHttpTransfer();

        JobSummary job = new JobSummary();
        setJobInfo(job, httpTransfer);

        when(uriService.getTransfer(any())).thenReturn(httpTransfer);

        jobService.setJobPhase(job, "RUN");

        // Completion will be set by file service
        assertEquals(ExecutionPhase.EXECUTING, job.getPhase());
    }

    @Test
    public void testStartJobMoveNode() {

        Transfer moveNode = new Transfer();
        moveNode.setDirection("vos://example.com!vospace/myfile");

        JobSummary job = new JobSummary();
        setJobInfo(job, moveNode);

        when(uriService.getTransfer(any())).thenReturn(moveNode);

        List<ExecutionPhase> phases = new ArrayList<>();
        doAnswer(invocation -> {
            JobSummary j = invocation.getArgument(0);
            phases.add(j.getPhase());
            return null;
        }).when(jobDAO).updateJob(any(), any());

        jobService.setJobPhase(job, "RUN");

        verify(moveService, timeout(1000).times(1)).processMoveJob(any(), any());

        verify(jobDAO, timeout(1000).times(3)).updateJob(any(), any());

        try {
            Thread.sleep(500);
        } catch (InterruptedException ex) {
        }

        assertEquals(ExecutionPhase.EXECUTING, phases.get(0));
        assertEquals(ExecutionPhase.EXECUTING, phases.get(1));
        assertEquals(ExecutionPhase.COMPLETED, phases.get(2));
    }

    @Test
    public void testStartJobCopyNode() {

        Transfer copyNode = new Transfer();
        copyNode.setDirection("vos://example.com!vospace/myfile");
        copyNode.setKeepBytes(true);

        JobSummary job = new JobSummary();
        setJobInfo(job, copyNode);

        when(uriService.getTransfer(any())).thenReturn(copyNode);

        List<ExecutionPhase> phases = new ArrayList<>();
        doAnswer(invocation -> {
            JobSummary j = invocation.getArgument(0);
            phases.add(j.getPhase());
            return null;
        }).when(jobDAO).updateJob(any(), any());

        when(copyService.processCopyNodes(any(), any(), any())).thenReturn(List.of("source", "dest"));
        
        jobService.setJobPhase(job, "RUN");

        verify(fileServiceClient, timeout(1000).times(1)).startFileCopyJob(eq("source"), eq("dest"), any(), any());

        verify(jobDAO, timeout(1000).times(3)).updateJob(any(), any());

        try {
            Thread.sleep(500);
        } catch (InterruptedException ex) {
        }

        assertEquals(ExecutionPhase.EXECUTING, phases.get(0));
        assertEquals(ExecutionPhase.EXECUTING, phases.get(1));
        assertEquals(ExecutionPhase.EXECUTING, phases.get(2));
    }

    private Transfer getPullFromVoSpaceHttpTransfer() {
        Transfer transfer = new Transfer();
        transfer.setTarget("vos://example.com!vospace/myfile");
        transfer.setDirection("pullFromVoSpace");
        Protocol protocol = new Protocol();
        protocol.setUri("ivo://ivoa.net/vospace/core#httpget");
        transfer.getProtocols().add(protocol);
        return transfer;
    }

    private Transfer getTapeTransfer() {
        Transfer transfer = new Transfer();
        transfer.setDirection("pullToVoSpace");
        View view = new View();
        view.setUri(Views.ASYNC_RECALL_VIEW_URI);
        transfer.setView(view);
        return transfer;
    }

    private void setJobInfo(JobSummary job, Transfer transfer) {
        JobSummary.JobInfo jobInfo = new JobSummary.JobInfo();
        jobInfo.getAny().add(transfer);
        job.setJobInfo(jobInfo);
    }
}