@Test
  public void testFileCopiedLogSegmentIsValid() throws Exception {
    // Arrange -
    // 1. Clean up source and target directories
    // 2. Create a log segment file in the target directory of size 25MB.
    // 3. Set up Spy on FCHandler with FileCopyGetMetaDataResponse to return the log segment file.
    // 4. Set up Spy on FCHandler with FileCopyGetChunkResponse to return the log segment file.
    cleanUpDirectories(Arrays.asList(sourcePartitionDir, targetPartitionDir));

    LogSegmentName logSegmentName = getRandomLogSegmentName();
    String logSegmentFileName = logSegmentName + "_log";

    File logSegment = new File(targetPartitionDir, logSegmentFileName);
    createSampleFile(logSegment, SEGMENT_CAPACITY); // 25 MB
    String targetLogSegment = fileStore.resetCompactionCycleIndexInFileName(logSegmentFileName);
    StoreLogInfo storeLogInfo = new StoreLogInfo(new StoreFileInfo(logSegmentName.toString(), logSegment.length()),
        new ArrayList<>(), new ArrayList<>());
    FileCopyGetMetaDataResponse fileCopyGetMetaDataResponse = new FileCopyGetMetaDataResponse(
        FileCopyGetMetaDataResponse.FILE_COPY_PROTOCOL_METADATA_RESPONSE_VERSION_V_1, 0, "clientId",
        1, Collections.singletonList(storeLogInfo), "snapshotId", ServerErrorCode.NoError, null);

    StoreFileCopyHandler spyHandler = spy(super.handler);
    doReturn(fileCopyGetMetaDataResponse)
        .when(spyHandler)
        .getFileCopyGetMetaDataResponse(any());

    List<FileCopyGetChunkResponse> chunkResponses = new ArrayList<>();
    int chunksInLogSegment = (int) Math.ceil((double) logSegment.length() / fileCopyBasedReplicationConfig.getFileCopyHandlerChunkSize);

    for (int i = 0; i < chunksInLogSegment; i++) {
      int startOffset = i * fileCopyBasedReplicationConfig.getFileCopyHandlerChunkSize;
      int sizeInBytes = (int)Math.min(fileCopyBasedReplicationConfig.getFileCopyHandlerChunkSize, logSegment.length() - startOffset);

      chunkResponses.add(getFileCopyGetChunkResponse(logSegment, startOffset, sizeInBytes));
    }

    // Expecting 3 chunks for the log segment (log segment size = 25MB, chunk size = 10MB)
    doReturn(chunkResponses.get(0)).
    doReturn(chunkResponses.get(1)).
    doReturn(chunkResponses.get(2))
        .when(spyHandler)
        .getFileCopyGetChunkResponse(any(), any(), any(), any(), anyBoolean());

    // Act - Run FCHandler.copy using Spy handler
    // TODO: Temporary directory should have been created before FileCopyHandler.copy() is called.
    File fileCopyTempDirectory = new File(sourcePartitionDir, "fileCopyTempDirectory");
    assertFalse("Temporary directory already exists!", fileCopyTempDirectory.exists());
    assertTrue(fileCopyTempDirectory.mkdirs());
    assertEquals(SEGMENT_COUNT, getReservePoolRemainingSegmentCount(STORE_ID, SEGMENT_CAPACITY));
    spyHandler.copy(fileCopyInfo);
    assertTrue(fileCopyTempDirectory.delete());

    // Assert -
    // 1. Check that no sub-directories are created in the source directory
    // 2. Check that the log segment file is copied correctly to the source directory
    // 3. Check that the number of log segment files in the source directory is 1
    assertNoSubDirectories(sourcePartitionDir);
    assertCopiedFileIsValid(logSegment, new File(sourcePartitionDir, targetLogSegment));
    assertEquals(1, getNumberOfFilesMatchingSuffix(sourcePartitionDir, "_log"));
    assertEquals(SEGMENT_COUNT-1, getReservePoolRemainingSegmentCount(STORE_ID, SEGMENT_CAPACITY));

    // cleanup the file to give it back to DiskSpaceAllocator
    fileStore.cleanLogFile(logSegment.getPath(), STORE_ID);
    assertEquals(SEGMENT_COUNT, getReservePoolRemainingSegmentCount(STORE_ID, SEGMENT_CAPACITY));
  }
Tests that the file copy handler can copy log segments from source to target.
It creates a log segment file in the target directory,
and then verifies that the file is copied correctly to the source directory.
ambry-file-transfer/src/test/java/com/github/ambry/filetransfer/handler/StoreFileCopyHandlerIntegTest.java