diff --git a/check-code.sh b/check-code.sh
index 3c9f5b3..9e05123 100755
--- a/check-code.sh
+++ b/check-code.sh
@@ -13,5 +13,4 @@ pylint --py3k --reports no ${py_files}
 pep8 ${py_files} --max-line-length 120
 
 # Run tests
-# -> But not those that require the neocommon library.
-py.test fetch test -m 'not with_neocommon'
+py.test fetch test
diff --git a/example-config.yaml b/example-config.yaml
index 3f54a63..1559ba5 100755
--- a/example-config.yaml
+++ b/example-config.yaml
@@ -7,22 +7,9 @@
 # The work directory (for log and lock files):
 directory: /data/fetch
 
-# Notification settings (for errors):
-notify:
-  email: ['jeremy.hooke@ga.gov.au']
-
-# Message bus config, for announcing arrivals. Optional.
-messaging:
-  host: rhe-neo-dev01.dev.lan
-  # virtual_host:
-  username: fetch
-  password: fetch
-
 # Logging level for modules (any python module can be added here)
 log:
   fetch: DEBUG
-  neocommon: DEBUG
-  neocommon.files: INFO
 
 # Download rules:
 rules:
@@ -190,7 +177,9 @@ rules:
     # Convert files to tiff (from netCDF)
     process: !shell
       command: '/usr/local/bin/gdal_translate -a_srs "+proj=latlong +datum=WGS84" {parent_dir}/{filename} {parent_dir}/{file_stem}.tif'
-      expect_file: '{parent_dir}/{file_stem}.tif'
+      upload_dir: '/data/water_vapour'
+      bucket: 'ard-processing-data'
+      prefix: 'ancillary/water_vapour'
 #
 #-------------------------------
 # Examples using the ECMWF API |
diff --git a/fetch/_core.py b/fetch/_core.py
index e2c3d84..d53dc03 100644
--- a/fetch/_core.py
+++ b/fetch/_core.py
@@ -7,20 +7,16 @@
 import datetime
 import errno
 import logging
-import multiprocessing
 import os
 import re
-import smtplib
-import socket
 import subprocess
 import tempfile
-from email.mime.text import MIMEText
-from email.header import Header
 
 from pathlib import Path
 from typing import Callable
 
 from .util import rsync, Uri
+from . import s3
 
 _log = logging.getLogger(__name__)
 
@@ -468,76 +464,6 @@ def on_process_failure(self, process):
         pass
 
 
-class TaskFailureEmailer(TaskFailureListener):
-    """
-    Send failure information via email
-    """
-
-    def __init__(self, addresses):
-        """
-        :type addresses: list of str
-        """
-        self.addresses = addresses
-
-    def on_file_failure(self, process_name, file_uri, summary, body_text):
-        """
-        Send mail on a
-        :param process_name:
-        :param file_uri:
-        :param summary:
-        :param body_text:
-        :return:
-        """
-        self._send_mail(
-            u'uri: {uri}\n{summary}\n\n{body}'.format(
-                uri=file_uri,
-                summary=summary,
-                body=body_text
-            ),
-            process_name
-        )
-
-    def on_process_failure(self, process):
-        """
-        :type process: ScheduledProcess
-        """
-
-        # A negative exit code means it was killed via a signal. Probably by the user.
-        # Not worth emailing.
-        if process.exitcode < 0:
-            return
-
-        with open(process.log_file, 'rt') as f:
-            msg = f.read()
-
-        self._send_mail(msg, process.name)
-
-    def _send_mail(self, body_text, process_name):
-        """
-        :type body_text: str
-        :type process_name: str
-        """
-        hostname = socket.getfqdn()
-        msg = MIMEText(body_text.encode('utf-8'), 'plain', 'utf-8')
-        msg['Subject'] = Header(u'{name} failure on {hostname}'.format(
-            name=process_name,
-            hostname=hostname
-        ).encode('utf-8'), 'utf-8')
-        from_address = 'fetch-{pid}@{hostname}'.format(
-            pid=multiprocessing.current_process().pid,
-            hostname=hostname
-        )
-        msg['from'] = from_address
-        msg['to'] = ", ".join(self.addresses)
-        s = smtplib.SMTP('localhost')
-        s.sendmail(
-            from_address,
-            self.addresses,
-            msg.as_string()
-        )
-        s.quit()
-
-
 class FileProcessor(SimpleObject):
     """
     Any action that will process a file after retrieval. (base class)
@@ -560,20 +486,21 @@ class ShellFileProcessor(FileProcessor):
     :type command: str
     """
 
-    def __init__(self, command=None, expect_file=None, input_files=None):
+    def __init__(self, command, upload_dir, bucket, prefix):
         super(ShellFileProcessor, self).__init__()
         self.command = command
-        self.expect_file = expect_file
-        self.input_files = input_files
+        self.upload_dir = upload_dir
+        self.bucket = bucket
+        self.prefix = prefix
 
-    def _apply_file_pattern(self, pattern, file_path, **keywords):
+    def _apply_file_pattern(self, pattern, file_path):
         """
         Format the given pattern.
         :type file_path: str
 
         :rtype: str
 
-        >>> p = ShellFileProcessor()
+        >>> p = ShellFileProcessor('run command on file {filename}', '/data/upload', 's3-bucket', 's3-prefix')
         >>> p._apply_file_pattern('{file_stem} extension {file_suffix}', '/tmp/something.txt')
         'something extension .txt'
         >>> p._apply_file_pattern('{filename} in {parent_dir}', '/tmp/something.txt')
@@ -582,8 +509,7 @@ def _apply_file_pattern(self, pattern, file_path, **keywords):
         '/tmp'
         >>> p._apply_file_pattern('{parent_dirs[1]}', '/tmp/something.txt')
         '/'
-        >>> p._apply_file_pattern('{base}.hdf', '/tmp/something.hdf',**{'base':'/tmp/something'})
-        '/tmp/something.hdf'
+
         """
         path = Path(file_path)
         return pattern.format(
@@ -598,8 +524,7 @@ def _apply_file_pattern(self, pattern, file_path, **keywords):
             parent_dirs=[str(p) for p in path.parents],
 
             # A more flexible alternative to the above.
-            path=path,
-            **keywords
+            path=path
         )
 
     def process(self, file_path):
@@ -609,33 +534,25 @@ def process(self, file_path):
         :raises: FileProcessError
         """
         command = self.command
-        if self.input_files:
-            path_transform = RegexpOutputPathTransform(self.input_files[0])
-            if not all([os.path.isfile(path_transform.transform_output_path(f, file_path))
-                        for f in self.input_files[1]]):
-                _log.info('Not all of the required_files are present.')
-                # This is used for reporting, so it is returning the file_path.
-                return file_path
-            else:
-                # format the path based on the group from
-                # transform output path
-                # command = path_transform.transform_output_path(command)
-                required_files_formating = path_transform.last_matched_groups
-        else:
-            required_files_formating = {}
-        command = self._apply_file_pattern(command, file_path, **required_files_formating)
+        command = self._apply_file_pattern(command, file_path)
         _log.info('Running %r', command)
 
         # Trigger command
         returned = subprocess.call(command, shell=True)
         if returned != 0:
             raise FileProcessError('Return code %r from command %r' % (returned, command))
+        else:
+            _log.debug("Command completed successfully.")
 
         # Check that output exists
-        expected_path = self._apply_file_pattern(self.expect_file, file_path, **required_files_formating)
+        expect_file = self.upload_dir + '/{file_stem}.h5'
+        expected_path = self._apply_file_pattern(expect_file, file_path)
 
         if not os.path.exists(expected_path):
             raise FileProcessError('Expected output not found {!r} for command {!r}'.format(expected_path, command))
 
+        s3.upload(expected_path, self.bucket, self.prefix)
+
         _log.debug('File available %r', expected_path)
+
         return expected_path
diff --git a/fetch/auto.py b/fetch/auto.py
index 7bdf393..6bf5b1c 100644
--- a/fetch/auto.py
+++ b/fetch/auto.py
@@ -25,7 +25,7 @@
 from croniter import croniter
 
 from . import load
-from ._core import ResultHandler, TaskFailureEmailer, RemoteFetchException, mkdirs
+from ._core import ResultHandler, RemoteFetchException, mkdirs
 
 # setproctitle is only supported on some platforms (Linux).
 try:
@@ -61,22 +61,7 @@ def _attempt_lock(lock_file):
     return True
 
 
-def _redirect_output(log_file):
-    """
-    Redirect all output to the given file.
-
-    :type log_file: str
-    """
-    output = open(log_file, 'w')
-    sys.stdout = output
-    sys.stderr = output
-    logging_clear()
-    handler = logging.StreamHandler(stream=output)
-    handler.setFormatter(_LOG_FORMATTER)
-    logging.getLogger().addHandler(handler)
-
-
-def _run_item(reporter, item, scheduled_time, log_directory, lock_directory):
+def _run_item(reporter, item, scheduled_time, lock_directory):
     """
     Run the given module in a subprocess
     :type reporter: ResultHandler
@@ -84,11 +69,11 @@ def _run_item(reporter, item, scheduled_time, log_directory, lock_directory):
     :rtype: ScheduledProcess
     """
     p = ScheduledProcess(
-        reporter, item, scheduled_time, log_directory, lock_directory
+        reporter, item, scheduled_time, lock_directory
     )
 
     _log.debug('Module info %r', item.module)
-    _log.info('Starting %r. Log %r, Lock %r', p.name, p.log_file, p.lock_file)
+    _log.info('Starting %r. Lock %r', p.name, p.lock_file)
     p.start()
     return p
 
@@ -98,22 +83,21 @@ class ScheduledProcess(multiprocessing.Process):
     A subprocess to run a module.
     """
 
-    def __init__(self, reporter, item, scheduled_time, log_directory, lock_directory, epoch_to_time=time.localtime):
+    def __init__(self, reporter, item, scheduled_time, lock_directory, epoch_to_time=time.localtime):
         """
         :type reporter: fetch.ResultHandler
         :type item: fetch.load.ScheduledItem
         :type scheduled_time: float
-        :type log_directory: str
         :type lock_directory: str
 
         >>> from ._core import EmptySource
         >>> item = load.ScheduledItem('LS7 CPF', '* * * * *', EmptySource())
         >>> # 04:36 UTC time
         >>> scheduled_time = 1416285412.541422
-        >>> log, lock = '/tmp/test-log', '/tmp/test-lock'
-        >>> s = ScheduledProcess(None, item, scheduled_time, log, lock, epoch_to_time=time.gmtime)
-        >>> (s.name, s.log_file, s.lock_file)
-        ('fetch-0436-ls7-cpf', '/tmp/test-log/0436-ls7-cpf.log', '/tmp/test-lock/ls7-cpf.lck')
+        >>> lock = '/tmp/test-lock'
+        >>> s = ScheduledProcess(None, item, scheduled_time, lock, epoch_to_time=time.gmtime)
+        >>> (s.name, s.lock_file)
+        ('fetch-0436-ls7-cpf', '/tmp/test-lock/ls7-cpf.lck')
         """
         super(ScheduledProcess, self).__init__()
         id_ = item.sanitized_name
@@ -122,15 +106,7 @@ def __init__(self, reporter, item, scheduled_time, log_directory, lock_directory
             '{id}.lck'.format(id=id_)
         )
         scheduled_time_st = time.strftime('%H%M', epoch_to_time(scheduled_time))
-        log_file = os.path.join(
-            log_directory,
-            '{time}-{id}.log'.format(
-                id=id_,
-                time=scheduled_time_st
-            )
-        )
 
-        self.log_file = log_file
         self.lock_file = lock_file
         self.name = 'fetch-{}-{}'.format(scheduled_time_st, id_)
         self.scheduled_time = scheduled_time
@@ -143,7 +119,6 @@ def run(self):
         Configure the environment and run our module.
         """
         _init_signals()
-        _redirect_output(self.log_file)
 
         if not _attempt_lock(self.lock_file):
             _log.debug('Lock is activated. Skipping run. %r', self.name)
@@ -235,8 +210,8 @@ def _on_child_finish(child, notifiers):
 
     if exit_code != 0:
         _log.error(
-            'Error return code %s from %r. Output logged to %r',
-            exit_code, child.name, child.log_file
+            'Error return code %s from %r.',
+            exit_code, child.name,
         )
 
         for n in notifiers:
@@ -263,30 +238,6 @@ def _filter_finished_children(running_children, notifiers):
     return still_running
 
 
-def get_day_log_dir(log_directory, time_secs):
-    """
-    Get log directory for this day.
-    :type log_directory: str
-    :type time_secs: float
-    :rtype: str
-
-    >>> get_day_log_dir('/tmp/day-dir-test', 1416285412.541422)
-    '/tmp/day-dir-test/2014/11-18'
-    """
-    # We use localtime because the cron scheduling uses localtime.
-    t = time.localtime(time_secs)
-
-    day_log_dir = os.path.join(
-        log_directory,
-        time.strftime('%Y', t),
-        time.strftime('%m-%d', t)
-    )
-    if not os.path.exists(day_log_dir):
-        mkdirs(day_log_dir)
-
-    return day_log_dir
-
-
 def _on_shutdown(running_children, notifiers):
     """
     :type running_children: set of ScheduledProcess
@@ -359,14 +310,10 @@ def __init__(self, config_path):
         self.schedule = None
         # : type: str
         self.base_directory = None
-        # : type: str
-        self.log_directory = None
         #: type: str
         self.lock_directory = None
         #: :type: list of fetch.TaskFailureListener
         self.notifiers = []
-        #: :type: dict of (str, str)
-        self.messaging_settings = None
         # Key-values are log names and levels.
         #: :type: dict of (str, str)
         self.log_levels = None
@@ -379,14 +326,6 @@ def load(self):
 
         self.schedule = Schedule(config.rules)
         self.base_directory = config.directory
-        self.messaging_settings = config.messaging_settings
-
-        _log.info('%s messaging configuration.', 'Loaded' if config.messaging_settings else 'No')
-
-        self.notifiers = []
-        if config.notify_addresses:
-            self.notifiers.append(TaskFailureEmailer(config.notify_addresses))
-        _log.info('%s addresses for error notification: %s', len(config.notify_addresses), config.notify_addresses)
 
         if not os.path.exists(self.base_directory):
             raise ValueError('Configured base folder does not exist: %r' % self.base_directory)
@@ -398,11 +337,6 @@ def load(self):
             if not os.path.exists(self.lock_directory):
                 mkdirs(self.lock_directory)
 
-        self.log_directory = os.path.join(self.base_directory, 'log')
-        _log.info('Using log directory %s', self.log_directory)
-        if not os.path.exists(self.log_directory):
-            mkdirs(self.log_directory)
-
         if config.log_levels != self.log_levels:
             _set_logging_levels(config.log_levels)
             self.log_levels = config.log_levels
@@ -421,34 +355,6 @@ def __init__(self, config, job_id):
         self.config = config
         self.job_id = job_id
 
-    def _announce_files_complete(self, source_uri, paths, msg_metadata=None):
-        """
-        Announce on the message bus that files are complete.
-
-        No-op if there is no messaging configuration.
-        :type source_uri: str
-        :type paths: list of str
-        """
-        md = msg_metadata or {}
-        md.update({
-            'source-uri': source_uri
-        })
-
-        _log.info('Completed %r -> %r', source_uri, paths)
-        if self.config.messaging_settings:
-            # Optional library.
-            #: pylint: disable=import-error
-            from neocommon import message, Uri as NeoUri
-            uris = [NeoUri.parse(path) for path in paths]
-            with message.NeoMessenger(message.MessengerConnection(**self.config.messaging_settings)) as msg:
-                msg.announce_ancillary(
-                    message.AncillaryUpdate(
-                        ancillary_type=self.job_id,
-                        uris=uris,
-                        properties=md
-                    )
-                )
-
     def files_complete(self, source_uri, paths, msg_metadata=None):
         """
         Call on completion of multiple files.
@@ -459,7 +365,7 @@ def files_complete(self, source_uri, paths, msg_metadata=None):
         :type msg_metadata: dict of (str, str)
         :return:
         """
-        self._announce_files_complete(source_uri, paths, msg_metadata=msg_metadata)
+        _log.info('Completed %r -> %r', source_uri, paths)
 
     def file_complete(self, source_uri, path, msg_metadata=None):
         """
@@ -468,7 +374,7 @@ def file_complete(self, source_uri, path, msg_metadata=None):
         :type msg_metadata: dict of (str, str)
         :type path: str
         """
-        self._announce_files_complete(source_uri, [path], msg_metadata=msg_metadata)
+        _log.info('Completed %r -> %r', source_uri, path)
 
     def file_error(self, uri, summary, body):
         """
@@ -541,8 +447,6 @@ def run_loop(o):
                 reporter,
                 scheduled_item,
                 scheduled_time=scheduled_time,
-                # Use a unique log directory for each day
-                log_directory=get_day_log_dir(o.log_directory, scheduled_time),
                 lock_directory=o.lock_directory
             )
             running_children.add(p)
@@ -602,8 +506,6 @@ def run_items(o, *item_names):
             NotifyResultHandler(o, chosen_item.sanitized_name),
             chosen_item,
             scheduled_time=scheduled_time,
-            # Use a unique log directory for each day
-            log_directory=get_day_log_dir(o.log_directory, scheduled_time),
             lock_directory=o.lock_directory
         )
         running_children.add(p)
@@ -662,6 +564,6 @@ def _set_logging_levels(levels):
         _log.info('Set log level %s to %s', name, level)
 
 
-_LOG_HANDLER = logging.StreamHandler(stream=sys.stderr)
+_LOG_HANDLER = logging.StreamHandler(sys.stdout)
 _LOG_FORMATTER = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
 _LOG_HANDLER.setFormatter(_LOG_FORMATTER)
diff --git a/fetch/ecmwf.py b/fetch/ecmwf.py
index dfa20a5..6a5a145 100644
--- a/fetch/ecmwf.py
+++ b/fetch/ecmwf.py
@@ -40,7 +40,6 @@ def retrieve(self, req):
             c = APIRequest(
                 self.url,
                 "datasets/%s" % (dataset, ),
-                self.email,
                 self.key,
                 self.trace,
                 verbose=self.verbose
diff --git a/fetch/load.py b/fetch/load.py
index c270a16..2c6d69d 100644
--- a/fetch/load.py
+++ b/fetch/load.py
@@ -10,7 +10,6 @@
 import os
 
 import yaml
-import yaml.resolver
 from croniter import croniter
 
 from . import ftp, http, ecmwf
@@ -137,7 +136,7 @@ class Config(object):
     Configuration.
     """
 
-    def __init__(self, directory, rules, notify_addresses, messaging_settings=None, log_levels=None):
+    def __init__(self, directory, rules, log_levels=None):
         """
         :type directory: str
         :type rules: list of ScheduledItem
@@ -150,15 +149,6 @@ def __init__(self, directory, rules, notify_addresses, messaging_settings=None,
         # Empty list of rules is ok: they may be added after startup (a config reload/SIGHUP).
         self.rules = rules
 
-        self.notify_addresses = notify_addresses
-
-        if messaging_settings:
-            # Optional library.
-            #: pylint: disable=import-error
-            from neocommon.message import MessengerConnection
-            verify_can_construct(MessengerConnection, messaging_settings, identifier='messaging settings')
-
-        self.messaging_settings = messaging_settings
         self.log_levels = log_levels
 
     @classmethod
@@ -171,15 +161,8 @@ def from_dict(cls, config):
         """
 
         directory = config.get('directory')
-        messaging_settings = config.get('messaging')
         log_levels = config.get('log')
 
-        notify_email_addresses = []
-        if 'notify' in config:
-            notify_config = config['notify']
-            if 'email' in notify_config:
-                notify_email_addresses = notify_config['email']
-
         rules = []
         if 'rules' in config:
             for name, fields in config['rules'].items():
@@ -190,8 +173,6 @@ def from_dict(cls, config):
         return Config(
             directory,
             rules,
-            notify_email_addresses,
-            messaging_settings=messaging_settings,
             log_levels=log_levels
         )
 
@@ -202,11 +183,7 @@ def to_dict(self):
         """
         return remove_nones({
             'directory': self.directory,
-            'notify': {
-                'email': self.notify_addresses
-            },
             'log': self.log_levels,
-            'messaging': self.messaging_settings,
             'rules': dict([
                 (
                     r.name, remove_nones({
diff --git a/fetch/s3.py b/fetch/s3.py
new file mode 100644
index 0000000..8a40538
--- /dev/null
+++ b/fetch/s3.py
@@ -0,0 +1,41 @@
+import logging
+import os
+import boto3
+
+_log = logging.getLogger(__name__)
+_s3 = boto3.client("s3")
+
+
+class FileUploadError(Exception):
+    """
+    An error in file processing.
+    """
+    pass
+
+
+def upload(filepath: str, bucket: str, prefix: str):
+    """
+    Upload all files in input_file_dir to S3.
+
+    :raises: FileUploadError
+    """
+    message = f"Uploading file {filepath} to s3://{bucket}/{prefix}"
+    _log.info(message)
+
+    if not os.path.exists(filepath):
+        raise FileUploadError(f"File does not exist: {filepath}")
+
+    directory, filename = os.path.split(filepath)
+
+    s3_key = f"{prefix}/{filename}"
+
+    try:
+        message = f"Uploading {filename} to s3://{bucket}/{s3_key}"
+        _log.debug(message)
+        _s3.upload_file(filepath, bucket, s3_key)
+    except Exception as e:
+        message = f"Failed to upload {filename} to s3://{bucket}/{s3_key}"
+        raise FileUploadError(message) from e
+
+    message = f"Successfully Uploaded {filename} to s3://{bucket}/{s3_key}"
+    _log.info(message)
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..d98ee87
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,40 @@
+[build-system]
+requires = [
+    "setuptools>=61",
+    "wheel",
+    "versioneer[toml]"
+]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "fetch"
+description = "Automatic data"
+requires-python = ">=3.7"
+
+authors = [
+    { name = "Jeremy Hooke", email = "jeremy.hooke@ga.gov.au" }
+]
+
+dynamic = ["version"]
+
+dependencies = [
+    "arrow",
+    "croniter",
+    "feedparser",
+    "lxml",
+    "pyyaml",
+    "requests>=2.21.0",
+    "boto3",
+    "pytest",
+    "mock",
+    "pylint",
+    "pep8",
+    "setproctitle; sys_platform == 'linux'",
+]
+
+[project.optional-dependencies]
+ecmwf = ["ecmwf-api-client"]
+
+[project.scripts]
+fetch-service = "fetch.scripts.service:main"
+fetch-now = "fetch.scripts.now:main"
diff --git a/test/test_auto.py b/test/test_auto.py
index 46422db..9b67698 100644
--- a/test/test_auto.py
+++ b/test/test_auto.py
@@ -12,7 +12,6 @@ def __init__(self, name='p', exitcode=None, pid=None):
                 self.exitcode = exitcode
                 self.name = name
                 self.pid = pid
-                self.log_file = '/tmp/test.log'
 
         running_proc = MockProcess(exitcode=None)
         failed_proc = MockProcess(exitcode=1)
diff --git a/test/test_ecmwf.py b/test/test_ecmwf.py
index 9485f4c..0be53b7 100644
--- a/test/test_ecmwf.py
+++ b/test/test_ecmwf.py
@@ -134,9 +134,6 @@ def _make_ecmwf_config(ancillary_data_root='/tmp/anc'):
     """
     schedule = {
         'directory': '/tmp/anc-fetch',
-        'notify': {
-            'email': ['test@ga.gov.au']
-        },
         'log': {
             'fetch': 'DEBUG'
         },
diff --git a/test/test_load.py b/test/test_load.py
index a211e45..19e3b29 100644
--- a/test/test_load.py
+++ b/test/test_load.py
@@ -3,13 +3,10 @@
 
 import tempfile
 
-import pytest
 from pathlib import Path
 
 from fetch import load
 
-with_neocommon = pytest.mark.with_neocommon
-
 
 def _fail_with_diff(reparsed_config, source_config):
     print('-' * 20)
@@ -44,20 +41,6 @@ def test_dump_load_obj_full():
     _check_load_dump_config(_make_config)
 
 
-@with_neocommon
-def test_dump_load_obj_with_messaging():
-    def make_config_no_messaging():
-        c = _make_config()
-        c['messaging'] = {
-            'host': 'rhe-pma-test08.test.lan',
-            'username': 'fetch',
-            'password': 'fetch'
-        }
-        return c
-
-    _check_load_dump_config(make_config_no_messaging)
-
-
 def print_simple_obj_diff(dict1, dict2):
     if type(dict2) in (int, float, str, text):
         print('-   {!r}'.format(dict1))
@@ -116,9 +99,6 @@ def _make_config():
     anc_data = '/tmp/anc'
     schedule = {
         'directory': '/tmp/anc-fetch',
-        'notify': {
-            'email': ['jeremy.hooke@ga.gov.au']
-        },
         'log': {
             'fetch': 'DEBUG'
         },
@@ -193,7 +173,9 @@ def _make_config():
                 'process': ShellFileProcessor(
                     command='/usr/local/bin/gdal_translate -a_srs "+proj=latlong +datum=WGS84" '
                             '{parent_dir}/{file_stem}.nc {parent_dir}/{file_stem}.tif',
-                    expect_file='{parent_dir}/{file_stem}.tif'
+                    upload_dir='{parent_dir}',
+                    bucket='ard-processing-data',
+                    prefix='ancillary/water_vapour'
                 )
             },
             'NPP GDAS-forecast': {
diff --git a/test/test_s3.py b/test/test_s3.py
new file mode 100644
index 0000000..512979a
--- /dev/null
+++ b/test/test_s3.py
@@ -0,0 +1,70 @@
+"""
+A package for testing the code in s3.py
+"""
+from fetch import s3
+from fetch.s3 import upload, FileUploadError
+
+import pytest
+from unittest.mock import patch
+
+# Constants for tests
+filepath = "/tmp/test.txt"
+bucket = "my-bucket"
+prefix = "my/prefix"
+
+
+def test_upload_success():
+    # Set up mocks
+    with (
+        patch("os.path.exists", return_value=True),
+        patch.object(s3, "_s3") as mock_s3,
+        patch.object(s3, "_log") as mock_log,
+    ):
+        # Call code
+        upload(filepath, bucket, prefix)
+
+        # Expected logged messages
+        mock_log.info.assert_any_call("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix")
+        mock_log.debug.assert_any_call("Uploading test.txt to s3://my-bucket/my/prefix/test.txt")
+        mock_log.info.assert_any_call("Successfully Uploaded test.txt to s3://my-bucket/my/prefix/test.txt")
+
+        # expected write to S3
+        mock_s3.upload_file.assert_called_once_with(
+            filepath,
+            bucket,
+            "my/prefix/test.txt",
+        )
+
+
+def test_given_file_not_exist():
+    # Set up mocks
+    with (
+        patch("os.path.exists", return_value=False),
+        patch.object(s3, "_s3") as mock_s3,
+        patch.object(s3, "_log") as mock_log,
+    ):
+        # Run code and confirm we get an expection back, checking type & message
+        with pytest.raises(FileUploadError, match="File does not exist: /tmp/test.txt"):
+            upload(filepath, bucket, prefix)
+
+        # Expected logged messages
+        mock_log.info.assert_called_once_with("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix")
+        mock_s3.upload_file.assert_not_called()
+
+
+def test_write_to_s3_failed():
+    # Set up mocks
+    with (
+        patch("os.path.exists", return_value=True),
+        patch.object(s3, "_s3") as mock_s3,
+        patch.object(s3, "_log") as mock_log,
+    ):
+        mock_s3.upload_file.side_effect = Exception("poop")
+
+        # Run code and confirm we get an expection back, checking type & message
+        with pytest.raises(FileUploadError, match="Failed to upload test.txt to s3://my-bucket/my/prefix/test.txt"):
+            upload(filepath, bucket, prefix)
+
+        # Expected logged messages
+        mock_log.info.assert_called_once_with("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix")
+        mock_log.debug.assert_called_once_with("Uploading test.txt to s3://my-bucket/my/prefix/test.txt")
diff --git a/test/test_shellfileprocessor.py b/test/test_shellfileprocessor.py
index db0b0ce..2335782 100644
--- a/test/test_shellfileprocessor.py
+++ b/test/test_shellfileprocessor.py
@@ -1,31 +1,104 @@
+"""
+A package for testing the code in the ShellFileProcessor class, in _core.py
+"""
+import pytest
+from unittest.mock import patch, call
 
-import os
-from pathlib import Path
+from fetch import s3
+from fetch import _core
+from fetch._core import ShellFileProcessor, FileProcessError
 
-from fetch._core import ShellFileProcessor
+# Test Constants
+command = "run command on file {filename}"
+upload_dir = "/data/upload"
+bucket = "s3-bucket"
+prefix = "s3-prefix"
+input_file = "/data/staging/file.txt"
 
 
-def test_shellfilepro_required_files_there():
-    command = 'ls {base}.py'
-    # required_files = (r'^(?P<base>.*test.+)\.py$',['{base}.py','{base}.py'])
-    required_files = ('^(?P<base>.*test.+)\\.py$', ['{base}.py', '{base}.py'])
-    file_path = os.path.abspath(__file__)
-    expect_file = '{base}.py'
-    sfp = ShellFileProcessor(command=command, expect_file=expect_file, input_files=required_files)
-    results = sfp.process(file_path)
+def test_apply_file_pattern_basic():
+    p = ShellFileProcessor(
+        command=command,
+        upload_dir=upload_dir,
+        bucket=bucket,
+        prefix=prefix,
+    )
 
-    assert results == Path(__file__).absolute().as_posix()
+    result = p._apply_file_pattern(command, "/tmp/something.txt")
 
+    assert result == "run command on file something.txt"
 
-def test_shellfilepro_required_files_not_there():
-    command = 'ls {base}.py'
-    required_files = (r'^(?P<base>.*test.+)\.py$', ['{base}.py', '/this/is/not/here/please.py'])
-    file_path = os.path.abspath(__file__)
-    sfp = ShellFileProcessor(command=command, expect_file=file_path, input_files=required_files)
-    results = sfp.process(file_path)
-    # A rather toothless assert given required_files_there would return the same result...
-    assert file_path == results
-    # Future option is to ask for a tmp_path, and make your command touch {tmp_path / 'test_file.txt'}.
-    # Then if the file exists, you know the command was run. It could work in both tests.
 
-# Add a test when no required_files parameter is supplied at all
+def test_process_success():
+    # Set up mocks
+    with (
+        patch("subprocess.call", return_value=0),
+        patch("os.path.exists", return_value=True),
+        patch.object(s3, "_s3"),
+        patch.object(_core, "_log") as mock_log
+    ):
+        p = ShellFileProcessor(
+            command=command,
+            upload_dir=upload_dir,
+            bucket=bucket,
+            prefix=prefix,
+        )
+
+        # assert success using return value (and no raised errors)
+        assert p.process(input_file) == "/data/upload/file.h5"
+
+        # confirm the correct messages are logged
+        mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt')
+        mock_log.debug.assert_has_calls([
+            call("Command completed successfully."),
+            call('File available %r', '/data/upload/file.h5')
+        ])
+
+
+def test_process_command_failed():
+    # Set up mocks
+    with (
+        patch("subprocess.call", return_value=1),
+        patch("os.path.exists", return_value=True),
+        patch.object(s3, "_s3"),
+        patch.object(_core, "_log") as mock_log
+    ):
+        p = ShellFileProcessor(
+            command=command,
+            upload_dir=upload_dir,
+            bucket=bucket,
+            prefix=prefix,
+        )
+
+        # Run code and confirm we get an expection back, checking type & message
+        with pytest.raises(FileProcessError, match="Return code 1 from command 'run command on file file.txt'"):
+            p.process(input_file)
+
+        # confirm the correct messages are logged
+        mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt')
+        mock_log.debug.assert_not_called()
+
+
+def test_process_output_file_missing():
+    # Set up mocks
+    with (
+        patch("subprocess.call", return_value=0),
+        patch("os.path.exists", return_value=False),
+        patch.object(s3, "_s3"),
+        patch.object(_core, "_log") as mock_log
+    ):
+        p = ShellFileProcessor(
+            command=command,
+            upload_dir=upload_dir,
+            bucket=bucket,
+            prefix=prefix,
+        )
+
+        # Run code and confirm we get an expection back, checking type & message
+        message = "Expected output not found '/data/upload/file.h5' for command 'run command on file file.txt'"
+        with pytest.raises(FileProcessError, match=message):
+            p.process(input_file)
+
+        # confirm the correct messages are logged
+        mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt')
+        mock_log.debug.assert_called_once_with("Command completed successfully.")