Untitled

package com.pep.pcaf.spp.salemetric.processor;

import com.azure.storage.blob.BlobClient;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobServiceClient;
import com.azure.storage.blob.models.BlobItem;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pcaf.shp.clf.model.CommonLoggingDto;
import com.pep.pcaf.spp.salemetric.config.AppProperties;
import com.pep.pcaf.spp.salemetric.config.AzureBlobConstants;
import com.pep.pcaf.spp.salemetric.config.AzureStorageBlobServiceConfig;
import com.pep.pcaf.spp.salemetric.processor.dto.SuccessResponseDto;
import com.pep.pcaf.spp.salemetric.repo.CouchbaseIdempotentRepository;
import com.pep.pcaf.spp.salemetric.service.DefaultChannelService;
import com.pep.pcaf.spp.salemetric.service.FailureAzureBlobService;
import com.pep.pcaf.spp.salemetric.util.ClfLogUtil;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.impl.DefaultMessage;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.util.StreamUtils;

import java.nio.charset.Charset;
import java.time.Instant;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;

public class AzureBlobProcessorTest {

    @InjectMocks
    private AzureBlobProcessor azureBlobProcessor;

    @Mock
    private BlobServiceClient blobServiceClient;

    @Mock
    private CouchbaseIdempotentRepository couchbaseIdempotentRepository;

    @Mock
    private DefaultChannelService channelService;

    @Mock
    private ClfLogUtil clfLogUtil;

    @Mock
    private AppProperties appProperties;

    @Mock
    private FailureAzureBlobService failureAzureBlobService;

    @Mock
    private BlobContainerClient blobContainerClient;

    @Mock
    private BlobClient blobClient;

    @Captor
    private ArgumentCaptor<String> stringCaptor;

    @BeforeEach
    public void setUp() {
        MockitoAnnotations.openMocks(this);
    }

    @Test
    public void testProcess() throws Exception {
        // Setup
        Exchange exchange = new DefaultExchange(null);
        Message in = new DefaultMessage(null);
        exchange.setIn(in);
        AzureStorageBlobServiceConfig serviceConfig = new AzureStorageBlobServiceConfig();
        serviceConfig.setFileContainer("test-container");
        serviceConfig.setRequestContainer("request-container");
        serviceConfig.setBucketName("bucket-name");
        in.setHeader(AzureBlobConstants.AZURE_STORAGE_BLOB_SERVICE_CONFIG, serviceConfig);
        BlobItem blobItem = mock(BlobItem.class);
        when(blobItem.getName()).thenReturn("test.json");
        List<BlobItem> blobItemList = Collections.singletonList(blobItem);
        in.setBody(blobItemList);

        when(blobServiceClient.getBlobContainerClient(anyString())).thenReturn(blobContainerClient);
        when(blobContainerClient.getBlobClient(anyString())).thenReturn(blobClient);
        when(blobClient.openInputStream()).thenReturn(StreamUtils.emptyInput());
        when(appProperties.isAuditLogEnabled()).thenReturn(true);
        when(appProperties.isCLELogEnabled()).thenReturn(true);
        when(couchbaseIdempotentRepository.add(anyString(), anyString(), anyString())).thenReturn(true);

        // Exercise
        azureBlobProcessor.process(exchange);

        // Verify
        verify(blobClient, times(1)).openInputStream();
        verify(couchbaseIdempotentRepository, times(1)).add(anyString(), anyString(), anyString());
        verify(channelService, times(1)).buildDistrictChannel(anyString(), anyString(), anyString(), anyString());
    }

    @Test
    public void testStartLockTheBlobFile() throws Exception {
        // Setup
        when(couchbaseIdempotentRepository.getKeyValue(anyString())).thenReturn("key");
        when(couchbaseIdempotentRepository.add(anyString(), anyString(), anyString())).thenReturn(true);
        when(blobClient.getBlobName()).thenReturn("blobName");
        AzureStorageBlobServiceConfig serviceConfig = new AzureStorageBlobServiceConfig();

        // Exercise
        boolean result = azureBlobProcessor.startLockTheBlobFile(blobClient, serviceConfig);

        // Verify
        assertTrue(result);
        verify(couchbaseIdempotentRepository, times(1)).add(anyString(), anyString(), anyString());
    }

    @Test
    public void testInjectDocumentIntoCB() {
        // Setup
        String blobContentAsString = "[{\"$Type\":\"DstrctSlsMtrc\",\"RgnId\":\"1\"}]";
        AzureStorageBlobServiceConfig serviceConfig = new AzureStorageBlobServiceConfig();
        serviceConfig.setBucketName("bucket-name");
        when(appProperties.isCLELogEnabled()).thenReturn(true);

        // Exercise
        azureBlobProcessor.injectDocumentIntoCB(blobContentAsString, blobClient, serviceConfig);

        // Verify
        verify(couchbaseIdempotentRepository, times(1)).upsertDoc(anyString(), anyString(), anyString());
    }

    @Test
    public void testStoreSuccessCount() throws Exception {
        // Setup
        when(couchbaseIdempotentRepository.getKeyValue(anyString())).thenReturn("key");
        when(couchbaseIdempotentRepository.retrieve(anyString(), anyString())).thenReturn("{\"status\":\"processing\"}");
        when(blobClient.getBlobName()).thenReturn("blobName");
        AzureStorageBlobServiceConfig serviceConfig = new AzureStorageBlobServiceConfig();
        AtomicInteger counter = new AtomicInteger(5);

        // Exercise
        azureBlobProcessor.storeSuccessCount(blobClient, serviceConfig, counter);

        // Verify
        verify(couchbaseIdempotentRepository, times(1)).confirm(anyString(), anyString(), anyString());
    }
}