TransferControllerTest.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.ia2.aa.data.User;
import static it.inaf.oats.vospace.VOSpaceXmlTestUtil.loadDocument;
import it.inaf.oats.vospace.datamodel.NodeProperties;
import it.inaf.oats.vospace.datamodel.Views;
import it.inaf.oats.vospace.exception.ErrorSummaryFactory;
import it.inaf.oats.vospace.exception.PermissionDeniedException;
import it.inaf.oats.vospace.persistence.JobDAO;
import it.inaf.oats.vospace.persistence.LocationDAO;
import it.inaf.oats.vospace.persistence.NodeDAO;
import it.inaf.oats.vospace.persistence.model.Location;
import it.inaf.oats.vospace.persistence.model.LocationType;
import it.inaf.oats.vospace.persistence.model.Storage;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Optional;
import net.ivoa.xml.uws.v1.ExecutionPhase;
import net.ivoa.xml.uws.v1.JobSummary;
import net.ivoa.xml.uws.v1.Jobs;
import net.ivoa.xml.uws.v1.ShortJobDescription;
import net.ivoa.xml.vospace.v2.DataNode;
import net.ivoa.xml.vospace.v2.Node;
import net.ivoa.xml.vospace.v2.Property;
import net.ivoa.xml.vospace.v2.Protocol;
import net.ivoa.xml.vospace.v2.Transfer;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.matchesPattern;
import static org.junit.jupiter.api.Assertions.assertEquals;
import org.junit.jupiter.api.Test;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.http.MediaType;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.web.servlet.MockMvc;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
import org.w3c.dom.Document;
import java.util.List;
import net.ivoa.xml.uws.v1.ErrorSummary;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.junit.jupiter.api.BeforeEach;
import org.mockito.ArgumentCaptor;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.doAnswer;

@SpringBootTest
@AutoConfigureMockMvc
@ContextConfiguration(classes = {TokenFilterConfig.class})
@TestPropertySource(properties = {"spring.main.allow-bean-definition-overriding=true", "file-service-url=http://file-service"})
public class TransferControllerTest {

    @MockBean
    private JobDAO jobDao;

    @MockBean
    private NodeDAO nodeDao;

    @MockBean
    private LocationDAO locationDao;

    @MockBean
    private AsyncTransferService asyncTransfService;

    @Autowired
    private MockMvc mockMvc;

    @BeforeEach
    public void init() {
        Location asyncLocation = new Location();
        asyncLocation.setType(LocationType.ASYNC);
        asyncLocation.setId(1);
        when(locationDao.getNodeLocation(eq("/mynode"))).thenReturn(Optional.of(asyncLocation));

        Location portalLocation = new Location();
        portalLocation.setType(LocationType.PORTAL);
        portalLocation.setId(2);
        Storage portalStorage = new Storage();
        portalStorage.setHostname("archive.lbto.org");
        portalStorage.setBaseUrl("/files");
        portalLocation.setSource(portalStorage);
        when(locationDao.getNodeLocation(eq("/portalnode"))).thenReturn(Optional.of(portalLocation));
        when(locationDao.findPortalLocation(any())).thenReturn(Optional.of(portalLocation));
    }

    @Test
    public void testPullFromVoSpaceAsync() throws Exception {
        // job completion will be set by file service
        String endpoint = testAsyncTransferNegotiation("/mynode",
                getResourceFileContent("pullFromVoSpace.xml"), ExecutionPhase.EXECUTING);
        assertTrue(endpoint.startsWith("http://file-service/mynode?jobId="));
    }

    @Test
    public void testPullFromVoSpaceSync() throws Exception {

        Node node = mockPublicDataNode();
        when(nodeDao.listNode(eq("/mynode"))).thenReturn(Optional.of(node));

        String requestBody = getResourceFileContent("pullFromVoSpace.xml");

        String redirect = mockMvc.perform(post("/synctrans")
                .content(requestBody)
                .contentType(MediaType.APPLICATION_XML)
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().is3xxRedirection())
                .andReturn().getResponse().getHeader("Location");

        assertThat(redirect, matchesPattern("^/transfers/.*/results/transferDetails"));

        verify(jobDao, times(1)).createJob(argThat(j -> {
            return ExecutionPhase.COMPLETED == j.getPhase()
                    && j.getResults().get(0).getHref().contains("/transferDetails");
        }), argThat(t -> {
            return t.getProtocols().get(0).getEndpoint().startsWith("http://file-service/mynode?jobId=");
        }));
    }

    @Test
    public void testPullToVoSpaceTape() throws Exception {
        testVoSpaceAsyncTransfer("/mynode", getResourceFileContent("pullToVoSpace-tape.xml"));

        verify(asyncTransfService, times(1)).startJob(any());

        verify(jobDao, times(2)).updateJob(argThat(j -> ExecutionPhase.QUEUED == j.getPhase()), any());
    }

    @Test
    public void testPullToVoSpacePortal() throws Exception {

        when(nodeDao.getNodeOsName(eq("/portalnode"))).thenReturn("file.fits");

        String endpoint = testAsyncTransferNegotiation("/portalnode",
                getResourceFileContent("pullToVoSpace-portal.xml"), ExecutionPhase.COMPLETED);

        assertTrue(endpoint.startsWith("http://archive.lbto.org"));

        verify(nodeDao, times(1)).setNodeLocation(eq("/portalnode"), eq(2), eq("lbcr.20130512.060722.fits.gz"));
    }

    @Test
    public void testPushToVoSpace() throws Exception {
        // job completion will be set by file service
        String endpoint = testAsyncTransferNegotiation("/uploadedfile",
                getResourceFileContent("pushToVoSpace.xml"), ExecutionPhase.EXECUTING);
        assertTrue(endpoint.startsWith("http://file-service/uploadedfile?jobId="));
    }

    private String testAsyncTransferNegotiation(String path, String requestBody, ExecutionPhase endPhase) throws Exception {

        // detect phase updates
        List<ExecutionPhase> phases = new ArrayList<>();
        List<Transfer> negotiatedTransfers = new ArrayList<>();
        doAnswer(invocation -> {
            phases.add(((JobSummary) invocation.getArgument(0)).getPhase());
            negotiatedTransfers.add(invocation.getArgument(1));
            return null;
        }).when(jobDao).updateJob(any(), any());

        testVoSpaceAsyncTransfer(path, requestBody);

        ArgumentCaptor<JobSummary> jobCaptor = ArgumentCaptor.forClass(JobSummary.class);
        verify(jobDao, times(2)).updateJob(jobCaptor.capture(), any());

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

        JobSummary job = jobCaptor.getAllValues().get(1);
        assertEquals(endPhase, job.getPhase());
        assertTrue(job.getResults().get(0).getHref().contains("/transferDetails"));

        assertNull(negotiatedTransfers.get(0));
        Transfer negotiatedTransfer = negotiatedTransfers.get(1);
        return negotiatedTransfer.getProtocols().get(0).getEndpoint();
    }

    private void testVoSpaceAsyncTransfer(String path, String requestBody) throws Exception {

        Node node = mockPublicDataNode();
        when(nodeDao.listNode(eq(path))).thenReturn(Optional.of(node));

        String redirect = mockMvc.perform(post("/transfers?PHASE=RUN")
                .header("Authorization", "Bearer user1_token")
                .content(requestBody)
                .contentType(MediaType.APPLICATION_XML)
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().is3xxRedirection())
                .andReturn().getResponse().getHeader("Location");

        assertThat(redirect, matchesPattern("^/transfers/.*"));
    }

    @Test
    public void testSetJobPhase() throws Exception {

        Node node = mockPublicDataNode();
        when(nodeDao.listNode(eq("/mynode"))).thenReturn(Optional.of(node));

        JobSummary job = getFakePendingJob();
        when(jobDao.getJob(eq("123"))).thenReturn(Optional.of(job));

        User user = new User();
        user.setUserId("ownerId");

        String redirect = mockMvc.perform(post("/transfers/123/phase")
                .header("Authorization", "Bearer user1_token")
                .param("PHASE", "RUN")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().is3xxRedirection())
                .andReturn().getResponse().getHeader("Location");

        verify(jobDao, times(2)).updateJob(any(), any());

        assertThat(redirect, matchesPattern("^/transfers/.*"));
    }

    @Test
    public void testGetTransferDetails() throws Exception {

        JobSummary job = getFakePendingJob();
        when(jobDao.getJob(eq("123"))).thenReturn(Optional.of(job));
        
        when(jobDao.getTransferDetails(eq("123"))).thenReturn(new Transfer());

        mockMvc.perform(get("/transfers/123/results/transferDetails")
                .header("Authorization", "Bearer user1_token")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().isOk());
    }

    @Test
    public void testGetJobPhase() throws Exception {

        JobSummary job = getFakePendingJob();
        when(jobDao.getJob(eq("123"))).thenReturn(Optional.of(job));

        String phase = mockMvc.perform(get("/transfers/123/phase")
                .header("Authorization", "Bearer user1_token"))
                .andDo(print())
                .andExpect(status().isOk())
                .andReturn().getResponse().getContentAsString();

        assertEquals("PENDING", phase);
    }

    private Node mockPublicDataNode() {
        Node node = new DataNode();
        Property property = new Property();
        property.setUri("ivo://ivoa.net/vospace/core#publicread");
        property.setValue("true");
        node.getProperties().add(property);

        Property ownerProp = new Property();
        ownerProp.setUri(NodeProperties.CREATOR_URI);
        ownerProp.setValue("user1");
        node.getProperties().add(ownerProp);

        Property groupProp = new Property();
        groupProp.setUri(NodeProperties.GROUP_WRITE_URI);
        groupProp.setValue("group1");
        node.getProperties().add(groupProp);

        return node;
    }

    @Test
    public void testGetJob() throws Exception {

        JobSummary job = new JobSummary();

        when(jobDao.getJob(eq("123"))).thenReturn(Optional.of(job));

        String xml = mockMvc.perform(get("/transfers/123")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().isOk())
                .andReturn().getResponse().getContentAsString();

        Document doc = loadDocument(xml);
        assertEquals("uws:job", doc.getDocumentElement().getNodeName());

        verify(jobDao, times(1)).getJob(eq("123"));
    }

    @Test
    public void testErrorEndpoint() throws Exception {
        JobSummary job = new JobSummary();
        job.setJobId("123");
        job.setPhase(ExecutionPhase.EXECUTING);
        ErrorSummary e = ErrorSummaryFactory.newErrorSummary(
                new PermissionDeniedException("/pippo1/pippo2")
        );
        job.setErrorSummary(e);

        when(jobDao.getJob(eq("123"))).thenReturn(Optional.of(job));

        String response = mockMvc.perform(get("/transfers/123/error")
                .accept(MediaType.TEXT_PLAIN_VALUE))
                .andDo(print())
                .andExpect(status().isOk())
                .andReturn().getResponse().getContentAsString();

        assertEquals("Job is not in ERROR phase", response);

        job.setPhase(ExecutionPhase.ERROR);

        response = mockMvc.perform(get("/transfers/123/error")
                .accept(MediaType.TEXT_PLAIN_VALUE))
                .andDo(print())
                .andExpect(status().isOk())
                .andReturn().getResponse().getContentAsString();

        assertEquals(e.getDetailMessage(), response);

        e.setHasDetail(false);

        response = mockMvc.perform(get("/transfers/123/error")
                .accept(MediaType.TEXT_PLAIN_VALUE))
                .andDo(print())
                .andExpect(status().isOk())
                .andReturn().getResponse().getContentAsString();

        assertEquals("No error details available", response);

        when(jobDao.getJob(eq("124"))).thenReturn(Optional.ofNullable(null));

        mockMvc.perform(get("/transfers/124/error")
                .accept(MediaType.TEXT_PLAIN_VALUE))
                .andDo(print())
                .andExpect(status().is4xxClientError());
    }

    @Test
    public void testGetJobs() throws Exception {

        when(jobDao.getJobs(eq("user1"), any(), any(), any(), any(), any()))
                .thenReturn(this.getFakeJobs());

        mockMvc.perform(get("/transfers")
                .header("Authorization", "Bearer user1_token")
                .param("LAST", "-3")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().is4xxClientError());

        mockMvc.perform(get("/transfers")
                .header("Authorization", "Bearer user1_token")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().isOk());

        // direction query parameter
        mockMvc.perform(get("/transfers")
                .param("direction", "pullFromVoSpace")
                .header("Authorization", "Bearer user1_token")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().isOk());

        verify(jobDao, times(1)).getJobs(eq("user1"), any(), argThat(v -> {
            return v.size() == 1 && v.contains(JobService.JobDirection.pullFromVoSpace);
        }), any(), any(), any());

        // PHASE query parameter
        mockMvc.perform(get("/transfers")
                .param("PHASE", ExecutionPhase.EXECUTING.value())
                .header("Authorization", "Bearer user1_token")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().isOk());

        verify(jobDao, times(1)).getJobs(eq("user1"), argThat(v -> {
            return v.size() == 1 && v.contains(ExecutionPhase.EXECUTING);
        }), any(), any(), any(), any());

        // VIEW query parameters
        mockMvc.perform(get("/transfers")
                .param("VIEW", Views.TAR_VIEW_URI)
                .param("VIEW", Views.ZIP_VIEW_URI)
                .header("Authorization", "Bearer user1_token")
                .accept(MediaType.APPLICATION_XML))
                .andDo(print())
                .andExpect(status().isOk());

        verify(jobDao, times(1)).getJobs(eq("user1"), any(), any(), argThat(v -> {
            return v.size() == 2 && v.contains(Views.TAR_VIEW_URI) && v.contains(Views.ZIP_VIEW_URI);
        }), any(), any());
    }

    @Test
    public void testSyncTransferUrlParamsMode() throws Exception {

        Node node = mockPublicDataNode();
        when(nodeDao.listNode(eq("/mynode"))).thenReturn(Optional.of(node));

        mockMvc.perform(get("/synctrans")
                .header("Authorization", "Bearer user1_token")
                .param("TARGET", "vos://example.com!vospace/mynode")
                .param("DIRECTION", "pullFromVoSpace")
                // testing duplicated protocol (CADC client)
                .param("PROTOCOL", "ivo://ivoa.net/vospace/core#httpget")
                .param("PROTOCOL", "ivo://ivoa.net/vospace/core#httpget"))
                .andExpect(status().is3xxRedirection());
    }

    private Jobs getFakeJobs() {
        Jobs jobs = new Jobs();
        jobs.setVersion("1.1");
        List<ShortJobDescription> sjdList = jobs.getJobref();
        sjdList.add(getFakeSJD1());

        return jobs;
    }

    private ShortJobDescription getFakeSJD1() {
        ShortJobDescription sjd = new ShortJobDescription();
        sjd.setId("pippo1");
        sjd.setPhase(ExecutionPhase.QUEUED);
        sjd.setOwnerId("user1");
        sjd.setType(JobService.JobDirection.pullFromVoSpace.toString());

        LocalDateTime now = LocalDateTime.now();
        Timestamp ts = Timestamp.valueOf(now);
        sjd.setCreationTime(JobDAO.toXMLGregorianCalendar(ts));

        return sjd;
    }

    private JobSummary getFakePendingJob() {
        JobSummary job = new JobSummary();
        job.setPhase(ExecutionPhase.PENDING);
        job.setOwnerId("user1");

        Transfer transfer = new Transfer();
        transfer.setDirection("pullFromVoSpace");
        transfer.setTarget(Arrays.asList("vos://example.com!vospace/mynode"));
        Protocol protocol = new Protocol();
        protocol.setUri("ivo://ivoa.net/vospace/core#httpget");
        transfer.getProtocols().add(protocol);

        JobSummary.JobInfo jobInfo = new JobSummary.JobInfo();
        jobInfo.getAny().add(transfer);

        job.setJobInfo(jobInfo);
        return job;
    }

    protected static String getResourceFileContent(String fileName) throws Exception {
        try ( InputStream in = TransferControllerTest.class.getClassLoader().getResourceAsStream(fileName)) {
            return new String(in.readAllBytes(), StandardCharsets.UTF_8);
        }
    }
}