SP-183, BIS-91: done including unit tests.

SVN: 26085

datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/Copier.java

@@ -29,6 +29,7 @@ import ch.systemsx.cisd.common.exceptions.Status;
 import ch.systemsx.cisd.common.filesystem.BooleanStatus;
 import ch.systemsx.cisd.common.filesystem.FileUtilities;
 import ch.systemsx.cisd.common.filesystem.HostAwareFile;
+import ch.systemsx.cisd.common.filesystem.IImmutableCopier;
 import ch.systemsx.cisd.common.filesystem.IPathCopier;
 import ch.systemsx.cisd.common.filesystem.ssh.ISshCommandExecutor;
 import ch.systemsx.cisd.common.highwatermark.HostAwareFileWithHighwaterMark;
@@ -50,6 +51,8 @@ public class Copier implements Serializable, IPostRegistrationDatasetHandler
 
     private final File sshExecutable;
 
+    private File lnExecutable;
+    
     private final String hostFile;
 
     private final String rsyncPasswordFile;
@@ -58,18 +61,36 @@ public class Copier implements Serializable, IPostRegistrationDatasetHandler
 
     private final ISshCommandExecutorFactory sshCommandExecutorFactory;
 
+    private final IImmutableCopierFactory immutableCopierFactory;
+    
     private final boolean renameToDataSetCode;
 
+    private final boolean hardLinkCopy;
+
     public Copier(Properties properties, IPathCopierFactory pathCopierFactory,
-            ISshCommandExecutorFactory sshCommandExecutorFactory)
+            ISshCommandExecutorFactory sshCommandExecutorFactory,
+            IImmutableCopierFactory immutableCopierFactory)
     {
         this.pathCopierFactory = pathCopierFactory;
         this.sshCommandExecutorFactory = sshCommandExecutorFactory;
+        this.immutableCopierFactory = immutableCopierFactory;
         rsyncPasswordFile = properties.getProperty(DataSetCopier.RSYNC_PASSWORD_FILE_KEY);
         rsyncExecutable = getExecutable(properties, DataSetCopier.RSYNC_EXEC);
         sshExecutable = getExecutable(properties, DataSetCopier.SSH_EXEC);
         hostFile = PropertyUtils.getMandatoryProperty(properties, DataSetCopier.DESTINATION_KEY);
-        renameToDataSetCode = PropertyUtils.getBoolean(properties, "rename-to-dataset-code", false);
+        renameToDataSetCode = PropertyUtils.getBoolean(properties, DataSetCopier.RENAME_TO_DATASET_CODE_KEY, false);
+        hardLinkCopy = PropertyUtils.getBoolean(properties, DataSetCopier.HARD_LINK_COPY_KEY, false);
+        if (hardLinkCopy)
+        {
+            String host = HostAwareFileWithHighwaterMark.create(hostFile, -1).tryGetHost();
+            if (host != null)
+            {
+                throw new ConfigurationFailureException(
+                        "Hard link copying not possible on an unmounted destination on host '"
+                                + host + "'.");
+            }
+            lnExecutable = getExecutable(properties, DataSetCopier.LN_EXEC);
+        }
     }
 
     protected String transformHostFile(String originalHostFile,
@@ -100,24 +121,37 @@ public class Copier implements Serializable, IPostRegistrationDatasetHandler
         }
         File destination = hostAwareFile.getFile();
         File destinationFile = new File(destination, originalData.getName());
+        String dataSetCode = dataSetInformation.getDataSetCode();
+        File finalDestinationFile = new File(destination, renameToDataSetCode ? dataSetCode : originalData.getName());
         BooleanStatus destinationExists =
-                checkDestinationFileExistence(destinationFile, host, sshCommandExecutor);
+                checkDestinationFileExistence(finalDestinationFile, host, sshCommandExecutor);
 
         if (destinationExists.isSuccess())
         {
-            operationLog.error("Destination file/directory '" + destinationFile.getPath()
+            operationLog.error("Destination file/directory '" + finalDestinationFile.getPath()
                     + "' already exists - dataset files will not be copied.");
             return Status.createError(DataSetCopier.ALREADY_EXIST_MSG);
         } else if (destinationExists.isError())
         {
-            operationLog.error("Could not test destination file/directory '" + destinationFile
+            operationLog.error("Could not test destination file/directory '" + finalDestinationFile
                     + "' existence" + (host != null ? " on host '" + host + "'" : "") + ": "
                     + destinationExists.tryGetMessage());
             return Status.createError(DataSetCopier.COPYING_FAILED_MSG);
         }
-        Status status =
-                copier.copyToRemote(originalData, destination, host, rsyncModule, rsyncPasswordFile);
-        String dataSetCode = dataSetInformation.getDataSetCode();
+        Status status;
+        if (hardLinkCopy)
+        {
+            IImmutableCopier hardLinkMaker = immutableCopierFactory.create(rsyncExecutable,
+            lnExecutable);
+            status =
+                    hardLinkMaker.copyImmutably(originalData, destination,
+                            renameToDataSetCode ? dataSetCode : null);
+        } else
+        {
+            status =
+                    copier.copyToRemote(originalData, destination, host, rsyncModule,
+                            rsyncPasswordFile);
+        }
         if (status.isError())
         {
             operationLog.error("Could not copy data set " + dataSetCode
@@ -127,20 +161,19 @@ public class Copier implements Serializable, IPostRegistrationDatasetHandler
                     + status);
             return Status.createError(DataSetCopier.COPYING_FAILED_MSG);
         }
-        if (renameToDataSetCode)
+        if (hardLinkCopy == false && renameToDataSetCode)
         {
-            File newFile = new File(destination, dataSetCode);
             if (host == null)
             {
-                if (destinationFile.renameTo(newFile))
+                if (destinationFile.renameTo(finalDestinationFile) == false)
                 {
                     operationLog.error("Moving of '" + destinationFile.getPath() + "' to '"
-                            + newFile + "' failed.");
-                    return Status.createError("couldn't move");
+                            + finalDestinationFile + "' failed.");
+                    return Status.createError("couldn't rename");
                 }
             } else
             {
-                String newFilePath = newFile.getPath();
+                String newFilePath = finalDestinationFile.getPath();
                 ProcessResult result =
                     sshCommandExecutor.executeCommandRemotely("mv " + destinationFile.getPath()
                             + " " + newFilePath, DataSetCopier.SSH_TIMEOUT_MILLIS);
@@ -148,7 +181,7 @@ public class Copier implements Serializable, IPostRegistrationDatasetHandler
                 {
                     operationLog.error("Remote move of '" + destinationFile.getPath() + "' to '"
                             + newFilePath + "' failed with exit value: " + result.getExitValue());
-                    return Status.createError("couldn't move");
+                    return Status.createError("couldn't rename");
                 }
             }
         }

datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/DataSetCopier.java

@@ -44,6 +44,10 @@ public class DataSetCopier extends AbstractDropboxProcessingPlugin
 
     public static final String RSYNC_PASSWORD_FILE_KEY = "rsync-password-file";
 
+    public static final String RENAME_TO_DATASET_CODE_KEY = "rename-to-dataset-code";
+    
+    public static final String HARD_LINK_COPY_KEY = "hard-link-copy";
+    
     @Private
     static final String ALREADY_EXIST_MSG = "already exist";
 
@@ -54,6 +58,8 @@ public class DataSetCopier extends AbstractDropboxProcessingPlugin
 
     public static final String RSYNC_EXEC = "rsync";
 
+    public static final String LN_EXEC = "ln";
+    
     public static final String SSH_EXEC = "ssh";
 
     public static final long SSH_TIMEOUT_MILLIS = 15 * 1000; // 15s
@@ -61,15 +67,16 @@ public class DataSetCopier extends AbstractDropboxProcessingPlugin
     public DataSetCopier(Properties properties, File storeRoot)
     {
         this(properties, storeRoot, new RsyncCopierFactory(), new SshCommandExecutorFactory(),
-                SystemTimeProvider.SYSTEM_TIME_PROVIDER);
+                new ImmutableCopierFactory(), SystemTimeProvider.SYSTEM_TIME_PROVIDER);
     }
 
     @Private
     DataSetCopier(Properties properties, File storeRoot, IPathCopierFactory pathCopierFactory,
-            ISshCommandExecutorFactory sshCommandExecutorFactory, ITimeProvider timeProvider)
+            ISshCommandExecutorFactory sshCommandExecutorFactory,
+            IImmutableCopierFactory immutableCopierFactory, ITimeProvider timeProvider)
     {
         super(properties, storeRoot, new Copier(properties, pathCopierFactory,
-                sshCommandExecutorFactory), timeProvider);
+                sshCommandExecutorFactory, immutableCopierFactory), timeProvider);
     }
 
     public DataSetCopier(Properties properties, File storeRoot,

datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/DataSetCopierForUsers.java

@@ -37,16 +37,16 @@ public class DataSetCopierForUsers extends DataSetCopier
 
     public DataSetCopierForUsers(Properties properties, File storeRoot)
     {
-        this(properties, storeRoot, new RsyncCopierFactory(), new SshCommandExecutorFactory());
+        this(properties, storeRoot, new RsyncCopierFactory(), new SshCommandExecutorFactory(), new ImmutableCopierFactory());
     }
 
     @Private
     DataSetCopierForUsers(Properties properties, File storeRoot,
             IPathCopierFactory pathCopierFactory,
-            ISshCommandExecutorFactory sshCommandExecutorFactory)
+            ISshCommandExecutorFactory sshCommandExecutorFactory, IImmutableCopierFactory immutableCopierFactory)
     {
         super(properties, storeRoot, new Copier(properties, pathCopierFactory,
-                sshCommandExecutorFactory)
+                sshCommandExecutorFactory, immutableCopierFactory)
             {
                 private static final long serialVersionUID = 1L;
 

datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/IImmutableCopierFactory.java

+/*
+ * Copyright 2012 ETH Zuerich, CISD
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard;
+
+import java.io.File;
+
+import ch.systemsx.cisd.common.filesystem.IImmutableCopier;
+
+/**
+ * Factory for creating an {@link IImmutableCopier} instance.
+ *
+ * @author Franz-Josef Elmer
+ */
+public interface IImmutableCopierFactory
+{
+    public IImmutableCopier create(File rsyncExecutable, File lnExecutable);
+}

datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/ImmutableCopierFactory.java

+/*
+ * Copyright 2012 ETH Zuerich, CISD
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard;
+
+import java.io.File;
+import java.io.Serializable;
+
+import ch.systemsx.cisd.common.filesystem.FastRecursiveHardLinkMaker;
+import ch.systemsx.cisd.common.filesystem.IImmutableCopier;
+
+/**
+ * Factory for {@link IImmutableCopier} instances based on {@link FastRecursiveHardLinkMaker}.
+ *
+ * @author Franz-Josef Elmer
+ */
+public class ImmutableCopierFactory implements IImmutableCopierFactory, Serializable
+{
+
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public IImmutableCopier create(File rsyncExecutable, File lnExecutable)
+    {
+        return FastRecursiveHardLinkMaker.create(rsyncExecutable, lnExecutable);
+    }
+
+}

datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/DataSetCopierForUsersTest.java

@@ -66,6 +66,8 @@ public class DataSetCopierForUsersTest extends AbstractFileSystemTestCase
 
     private ISshCommandExecutor sshCommandExecutor;
 
+    private IImmutableCopierFactory hardLinkMakerFactory;
+    
     private File storeRoot;
 
     private File sshExecutableDummy;
@@ -86,6 +88,7 @@ public class DataSetCopierForUsersTest extends AbstractFileSystemTestCase
         context = new Mockery();
         pathFactory = context.mock(IPathCopierFactory.class);
         sshFactory = context.mock(ISshCommandExecutorFactory.class);
+        hardLinkMakerFactory = context.mock(IImmutableCopierFactory.class);
         copier = context.mock(IPathCopier.class);
         sshCommandExecutor = context.mock(ISshCommandExecutor.class);
         storeRoot = new File(workingDirectory, "store");
@@ -135,7 +138,8 @@ public class DataSetCopierForUsersTest extends AbstractFileSystemTestCase
                 }
             });
         DataSetCopier dataSetCopier =
-                new DataSetCopierForUsers(properties, storeRoot, pathFactory, sshFactory);
+                new DataSetCopierForUsers(properties, storeRoot, pathFactory, sshFactory,
+                        hardLinkMakerFactory);
 
         ProcessingStatus processingStatus =
                 dataSetCopier.process(Arrays.asList(ds), dataSetProcessingContext);
@@ -159,7 +163,8 @@ public class DataSetCopierForUsersTest extends AbstractFileSystemTestCase
                 }
             });
         DataSetCopier dataSetCopier =
-                new DataSetCopierForUsers(properties, storeRoot, pathFactory, sshFactory);
+                new DataSetCopierForUsers(properties, storeRoot, pathFactory, sshFactory,
+                        hardLinkMakerFactory);
 
         ProcessingStatus processingStatus =
                 dataSetCopier.process(Arrays.asList(ds), dataSetProcessingContext);

datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/server/plugins/standard/DataSetCopierTest.java

@@ -17,6 +17,8 @@
 package ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard;
 
 import static ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard.DataSetCopier.DESTINATION_KEY;
+import static ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard.DataSetCopier.HARD_LINK_COPY_KEY;
+import static ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard.DataSetCopier.RENAME_TO_DATASET_CODE_KEY;
 import static ch.systemsx.cisd.openbis.dss.generic.server.plugins.standard.DataSetCopier.RSYNC_PASSWORD_FILE_KEY;
 
 import java.io.File;
@@ -39,6 +41,7 @@ import ch.systemsx.cisd.base.tests.AbstractFileSystemTestCase;
 import ch.systemsx.cisd.common.exceptions.ConfigurationFailureException;
 import ch.systemsx.cisd.common.exceptions.Status;
 import ch.systemsx.cisd.common.filesystem.BooleanStatus;
+import ch.systemsx.cisd.common.filesystem.IImmutableCopier;
 import ch.systemsx.cisd.common.filesystem.IPathCopier;
 import ch.systemsx.cisd.common.filesystem.ssh.ISshCommandExecutor;
 import ch.systemsx.cisd.common.mail.EMailAddress;
@@ -56,6 +59,8 @@ import ch.systemsx.cisd.openbis.generic.shared.dto.DatasetDescription;
     { DataSetCopier.class, AbstractDropboxProcessingPlugin.class })
 public class DataSetCopierTest extends AbstractFileSystemTestCase
 {
+    private static final String RSYNC_EXECUTABLE = "rsync-executable";
+
     private static final String SHARE_ID = "42";
 
     private static final String USER_EMAIL = "a@bc.de";
@@ -83,6 +88,8 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
     private File sshExecutableDummy;
 
     private File rsyncExecutableDummy;
+    
+    private File lnExecutableDummy;
 
     private Properties properties;
 
@@ -108,6 +115,10 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
 
     private IMailClient mailClient;
 
+    private IImmutableCopier hardLinkMaker;
+
+    private IImmutableCopierFactory hardLinkMakerFactory;
+
     @BeforeMethod
     public void beforeMethod() throws IOException
     {
@@ -117,6 +128,8 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
         pathFactory = context.mock(IPathCopierFactory.class);
         sshFactory = context.mock(ISshCommandExecutorFactory.class);
         copier = context.mock(IPathCopier.class);
+        hardLinkMakerFactory = context.mock(IImmutableCopierFactory.class);
+        hardLinkMaker = context.mock(IImmutableCopier.class);
         sshCommandExecutor = context.mock(ISshCommandExecutor.class);
         storeRoot = new File(workingDirectory, "store");
         storeRoot.mkdirs();
@@ -124,9 +137,12 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
         sshExecutableDummy.createNewFile();
         rsyncExecutableDummy = new File(workingDirectory, "my-rsync");
         rsyncExecutableDummy.createNewFile();
+        lnExecutableDummy = new File(workingDirectory, "my-ln");
+        lnExecutableDummy.createNewFile();
         properties = new Properties();
         properties.setProperty("ssh-executable", sshExecutableDummy.getPath());
-        properties.setProperty("rsync-executable", rsyncExecutableDummy.getPath());
+        properties.setProperty(RSYNC_EXECUTABLE, rsyncExecutableDummy.getPath());
+        properties.setProperty("ln-executable", lnExecutableDummy.getPath());
         ds1 = createDataSetDescription("ds1", DS1_LOCATION, true);
         File share = new File(storeRoot, SHARE_ID);
         File ds1Folder = new File(share, DS1_LOCATION + "/original");
@@ -462,22 +478,21 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
     }
 
     @Test
-    public void testCopyLocallyFails()
+    public void testCopyLocallyFailsBecauseDestinationExists()
     {
-        properties.setProperty(DESTINATION_KEY, "tmp/test");
-
-        File existingDestinationDir = new File("tmp/test/existing");
+        final File destination = new File(workingDirectory, "tmp/test");
+        properties.setProperty(DESTINATION_KEY, destination.getPath());
+        File existingDestinationDir = new File(destination, "existing");
         existingDestinationDir.mkdirs();
-
         prepareCreateAndCheckCopier(null, null, 4, true);
         context.checking(new Expectations()
             {
                 {
-                    one(copier).copyToRemote(ds1Data, getCanonicalFile("tmp/test"), null, null,
+                    one(copier).copyToRemote(ds1Data, getCanonicalFile(destination), null, null,
                             null);
                     will(returnValue(Status.createError("error message")));
 
-                    one(copier).copyToRemote(ds2Data, getCanonicalFile("tmp/test"), null, null,
+                    one(copier).copyToRemote(ds2Data, getCanonicalFile(destination), null, null,
                             null);
                     will(returnValue(Status.OK));
                 }
@@ -495,7 +510,25 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
         Status alreadyExistStatus = Status.createError(DataSetCopier.ALREADY_EXIST_MSG);
         assertError(processingStatus, alreadyExistStatus, ds3, ds4);
 
-        existingDestinationDir.delete();
+        context.assertIsSatisfied();
+    }
+
+    @Test
+    public void testCopyLocallyWithRenamingFailsBecauseDestinationExists()
+    {
+        final File destination = new File(workingDirectory, "tmp/test");
+        properties.setProperty(DESTINATION_KEY, destination.getPath());
+        properties.setProperty(RENAME_TO_DATASET_CODE_KEY, "true");
+        File existingDestinationDir = new File(destination, ds1.getDataSetCode());
+        existingDestinationDir.mkdirs();
+        prepareCreateAndCheckCopier(null, null, 1, true);
+        DataSetCopier dataSetCopier = createCopier();
+
+        ProcessingStatus processingStatus =
+                dataSetCopier.process(Arrays.asList(ds1), dummyContext);
+
+        Status errorStatus = Status.createError(DataSetCopier.ALREADY_EXIST_MSG);
+        assertError(processingStatus, errorStatus, ds1);
 
         context.assertIsSatisfied();
     }
@@ -631,6 +664,113 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
 
         context.assertIsSatisfied();
     }
+    
+    @Test
+    public void testHardLinkCopyingNotPossibleForRemoteDestinations()
+    {
+        properties.setProperty(DESTINATION_KEY, "host:tmp/test");
+        properties.setProperty(HARD_LINK_COPY_KEY, "true");
+        
+        try
+        {
+            createCopier();
+            fail("ConfigurationFailureException expected");
+        } catch (ConfigurationFailureException ex)
+        {
+            assertEquals("Hard link copying not possible on an unmounted destination "
+                    + "on host 'host'.", ex.getMessage());
+        }
+
+        context.assertIsSatisfied();
+    }
+    
+    @Test
+    public void testHardLinkCopyingSucessfully()
+    {
+        properties.setProperty(DESTINATION_KEY, "tmp/test");
+        properties.setProperty(HARD_LINK_COPY_KEY, "true");
+        properties.setProperty(AbstractDropboxProcessingPlugin.SEND_DETAILED_EMAIL_KEY, "true");
+        prepareCreateAndCheckCopier(null, null, 1, true);
+        final RecordingMatcher<String> subjectRecorder = new RecordingMatcher<String>();
+        final RecordingMatcher<String> contentRecorder = new RecordingMatcher<String>();
+        final RecordingMatcher<EMailAddress[]> recipientsRecorder =
+                new RecordingMatcher<EMailAddress[]>();
+        context.checking(new Expectations()
+            {
+                {
+                    one(hardLinkMakerFactory).create(rsyncExecutableDummy, lnExecutableDummy);
+                    will(returnValue(hardLinkMaker));
+                    
+                    one(hardLinkMaker).copyImmutably(ds1Data, getCanonicalFile("tmp/test"), null);
+                    will(returnValue(Status.OK));
+                    
+                    one(mailClient).sendEmailMessage(with(subjectRecorder), with(contentRecorder),
+                            with(new IsNull<EMailAddress>()), with(new IsNull<EMailAddress>()),
+                            with(recipientsRecorder));
+                }
+            });
+        DataSetCopier dataSetCopier = createCopier();
+        
+        ProcessingStatus processingStatus =
+                dataSetCopier.process(Arrays.asList(ds1), dummyContext);
+        
+        assertNoErrors(processingStatus);
+        assertSuccessful(processingStatus, ds1);
+        assertEquals(USER_EMAIL, recipientsRecorder.recordedObject()[0].tryGetEmailAddress());
+        assertEquals("Data set ds1 [MY_DATA] successfully processed",
+                subjectRecorder.recordedObject());
+        assertEquals("Successfully processed data set ds1 [MY_DATA].\n\n" + "Processing details:\n"
+                + "Description: Copy to tmp/test\n" + "Experiment: /g/p/e [MY_EXPERIMENT]\n"
+                + "Sample: /g/s [MY_SAMPLE]\n" + "Started: 1970-01-01 01:00:00 +0100.\n"
+                + "Finished: 1970-01-01 01:00:00 +0100.", contentRecorder.recordedObject());
+
+        context.assertIsSatisfied();
+    }
+
+    
+    @Test
+    public void testHardLinkCopyingWithRenamingSucessfully()
+    {
+        properties.setProperty(DESTINATION_KEY, "tmp/test");
+        properties.setProperty(HARD_LINK_COPY_KEY, "true");
+        properties.setProperty(RENAME_TO_DATASET_CODE_KEY, "true");
+        properties.setProperty(AbstractDropboxProcessingPlugin.SEND_DETAILED_EMAIL_KEY, "true");
+        prepareCreateAndCheckCopier(null, null, 1, true);
+        final RecordingMatcher<String> subjectRecorder = new RecordingMatcher<String>();
+        final RecordingMatcher<String> contentRecorder = new RecordingMatcher<String>();
+        final RecordingMatcher<EMailAddress[]> recipientsRecorder =
+                new RecordingMatcher<EMailAddress[]>();
+        context.checking(new Expectations()
+            {
+                {
+                    one(hardLinkMakerFactory).create(rsyncExecutableDummy, lnExecutableDummy);
+                    will(returnValue(hardLinkMaker));
+
+                    one(hardLinkMaker).copyImmutably(ds1Data, getCanonicalFile("tmp/test"),
+                            ds1.getDataSetCode());
+                    will(returnValue(Status.OK));
+
+                    one(mailClient).sendEmailMessage(with(subjectRecorder), with(contentRecorder),
+                            with(new IsNull<EMailAddress>()), with(new IsNull<EMailAddress>()),
+                            with(recipientsRecorder));
+                }
+            });
+        DataSetCopier dataSetCopier = createCopier();
+
+        ProcessingStatus processingStatus = dataSetCopier.process(Arrays.asList(ds1), dummyContext);
+
+        assertNoErrors(processingStatus);
+        assertSuccessful(processingStatus, ds1);
+        assertEquals(USER_EMAIL, recipientsRecorder.recordedObject()[0].tryGetEmailAddress());
+        assertEquals("Data set ds1 [MY_DATA] successfully processed",
+                subjectRecorder.recordedObject());
+        assertEquals("Successfully processed data set ds1 [MY_DATA].\n\n" + "Processing details:\n"
+                + "Description: Copy to tmp/test\n" + "Experiment: /g/p/e [MY_EXPERIMENT]\n"
+                + "Sample: /g/s [MY_SAMPLE]\n" + "Started: 1970-01-01 01:00:00 +0100.\n"
+                + "Finished: 1970-01-01 01:00:00 +0100.", contentRecorder.recordedObject());
+
+        context.assertIsSatisfied();
+    }
 
     private void prepareCreateAndCheckCopier(final String hostOrNull,
             final String rsyncModuleOrNull, final int numberOfExpectedCreations,
@@ -683,7 +823,9 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
     private void assertError(ProcessingStatus processingStatus, Status errorStatus,
             DatasetDescription... datasets)
     {
-        processingStatus.getErrorStatuses().contains(errorStatus);
+        List<Status> errorStatuses = processingStatus.getErrorStatuses();
+        assertEquals("Error statuses " + errorStatuses + " dosn't contain " + errorStatus, true,
+                errorStatuses.contains(errorStatus));
         checkStatus(processingStatus, errorStatus, datasets);
     }
 
@@ -706,10 +848,15 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
     }
 
     private File getCanonicalFile(String fileName)
+    {
+        return getCanonicalFile(new File(fileName));
+    }
+
+    private File getCanonicalFile(File file)
     {
         try
         {
-            return new File(fileName).getCanonicalFile();
+            return file.getCanonicalFile();
         } catch (IOException ex)
         {
             throw CheckedExceptionTunnel.wrapIfNecessary(ex);
@@ -718,7 +865,8 @@ public class DataSetCopierTest extends AbstractFileSystemTestCase
 
     private DataSetCopier createCopier()
     {
-        return new DataSetCopier(properties, storeRoot, pathFactory, sshFactory, timeProvider);
+        return new DataSetCopier(properties, storeRoot, pathFactory, sshFactory,
+                hardLinkMakerFactory, timeProvider);
     }
 
 }