Repository: pytroll/trollmoves (language: python)

Issue:
s3downloader: download data from an s3 (object store) to localhost
## Feature Request

Add a s3download daemon/module.

This listens to messages from eg s3stalker in pytroll-collectors, and then starts a transfere of these files to local host.

Optionally publish a resulting message with the downloaded file.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 58efeae8..538f6c8c 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -27,7 +27,7 @@ jobs:
 
       - name: Install dependencies
         run: |
-          pip install -U pytest pytest-cov pytest-bdd pytest-reraise pyyaml trollsift posttroll inotify pyinotify paramiko scp watchdog pytroll-collectors>=0.13.0 fsspec
+          pip install -U pytest pytest-cov pytest-bdd pytest-reraise pyyaml trollsift posttroll inotify pyinotify paramiko scp watchdog pytroll-collectors>=0.13.0 fsspec s3fs
 
       - name: Install trollmoves
         run: |
diff --git a/AUTHORS.md b/AUTHORS.md
new file mode 100644
index 00000000..c4f5d27d
--- /dev/null
+++ b/AUTHORS.md
@@ -0,0 +1,14 @@
+# Project Contributors
+
+The following people have made contributions to this project:
+
+<!--- Use your GitHub account or any other personal reference URL --->
+<!--- If you wish to not use your real name, please use your github username --->
+<!--- The list should be alphabetical by last name if possible, with github usernames at the bottom --->
+<!--- See https://gist.github.com/djhoese/52220272ec73b12eb8f4a29709be110d for auto-generating parts of this list --->
+
+- [Trygve Aspenes (TAlonglong)](https://github.com/TAlonglong)
+- [Adam Dybbroe (adybbroe)](https://github.com/adybbroe)
+- [Panu Lahtinen (pnuu)](https://github.com/pnuu)
+- [Lars Ørum Rasmussen (loerum)](https://github.com/loerum)
+- [Martin Raspaud (mraspaud)](https://github.com/mraspaud)
diff --git a/README.md b/README.md
index 6c9c4f64..20d2db8d 100644
--- a/README.md
+++ b/README.md
@@ -137,3 +137,15 @@ Additional required packages:
 
 Additional required packages:
 - ``s3fs``
+
+## s3downloader
+
+This module is able to download files from a s3 endpoint.
+
+The s3downloader module need to get posttroll messages from eg. s3stalker from the pytroll-collectors module which announces new available files in a configured s3 bucket. The s3downloader then downloads these files.
+
+Example config for the module is given in `examples/s3downloader-config.yaml`. If you use a `nameserver` with muliticast (nameserver default) you don't need to include the nameserver option nor the service name in the config. Opposite if you use nameserver without multicast.
+
+Additional required packages:
+- ``boto3``
+- ``botocore``
diff --git a/bin/s3downloader.py b/bin/s3downloader.py
new file mode 100644
index 00000000..c2b954a9
--- /dev/null
+++ b/bin/s3downloader.py
@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2022-2023 Pytroll Developers
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""s3 downloader to be used together with s3stalker from pytroll-collectors.
+S3downloader listens to messages from the s3stalker, and download files to
+configured destination. If the download is successful(ie. file exists on local disk
+a message is published to be used further downstream.
+
+A yaml config file is needed like this:
+---
+logging:
+  log_rotation_days: 1
+  log_rotation_backup: 30
+  logging_mode: DEBUG
+
+subscribe-topic:
+  - /yuhu
+publish-topic: /idnt
+endpoint_url: '<your object store endpoint url'
+access_key: ''
+secret_key: ''
+bucket: <name of the bucket>  # Not needed, else used from the message uri
+download_destination: './'
+
+"""
+
+import sys
+import logging
+
+from trollmoves.s3downloader import parse_args
+from trollmoves.s3downloader import S3Downloader
+
+LOGGER = logging.getLogger(__name__)
+
+
+def main():
+    cmd_args = parse_args(sys.argv[1:])
+
+    s3dl = S3Downloader(cmd_args)
+    s3dl.read_config()
+    s3dl.setup_logging()
+    s3dl.start()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/s3downloader-config.yaml-template b/examples/s3downloader-config.yaml-template
new file mode 100644
index 00000000..859ba426
--- /dev/null
+++ b/examples/s3downloader-config.yaml-template
@@ -0,0 +1,18 @@
+---
+logging:
+  log_rotation_days: 1
+  log_rotation_backup: 30
+  logging_mode: DEBUG
+
+subscribe-topic:
+  - /yuhu
+publish-topic: /idnt
+services:
+  - s3stalker
+bucket: '<name of the bucket>'
+download_destination: '<destination path for the downloaded file>'
+s3_kwargs:
+  anon: False
+  profile: metno # Optional
+  client_kwargs:
+    endpoint_url: <url to the object store>
diff --git a/trollmoves/s3downloader.py b/trollmoves/s3downloader.py
new file mode 100644
index 00000000..76986d3e
--- /dev/null
+++ b/trollmoves/s3downloader.py
@@ -0,0 +1,376 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2020-2023 Pytroll developers
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""s3 downloader to be used together with s3stalker from pytroll-collectors.
+S3downloader listens to messages from the s3stalker, and download files to
+configured destination. If the download is successful(ie. file exists on local disk
+a message is published to be used further downstream.
+
+A yaml config file is needed like this:
+---
+logging:
+  log_rotation_days: 1
+  log_rotation_backup: 30
+  logging_mode: DEBUG
+
+subscribe-topic:
+  - /yuhu
+publish-topic: /idnt
+bucket: <name of the bucket>  # Not needed, else used from the message uri
+download_destination: './'
+s3_kwargs:
+  anon: False
+  profile: <profile name> # Optional
+  client_kwargs:
+    endpoint_url: <url to the object store>
+"""
+
+import os
+import sys
+import yaml
+import queue
+import argparse
+import logging
+from logging import handlers
+from threading import Thread
+from urllib.parse import urlparse
+from posttroll.message import Message
+from posttroll.publisher import Publish
+from posttroll.subscriber import Subscribe
+
+try:
+    from s3fs import S3FileSystem
+except ImportError:
+    S3FileSystem = None
+
+LOGGER = logging.getLogger(__name__)
+# ----------------------------
+# Default settings for logging
+# ----------------------------
+_DEFAULT_TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
+_DEFAULT_LOG_FORMAT = '[%(levelname)s: %(asctime)s : %(name)s] %(message)s'
+
+
+class Listener(Thread):
+
+    def __init__(self, queue, config, subscribe_nameserver):
+        # Thread.__init__(self)
+        super(Listener, self).__init__()
+        self.loop = True
+        self.queue = queue
+        self.config = config
+        self.subscribe_nameserver = subscribe_nameserver
+
+    def stop(self):
+        """Stops the file listener."""
+        LOGGER.debug("Stopping FileListener.")
+        self.loop = False
+        self.queue.put(None)
+
+    def _check_and_put_message_to_queue(self, msg):
+        if not self.loop:
+            LOGGER.warning("Exiting FileListener.")
+            return False
+
+        # Check if it is a relevant message:
+        if self.check_message(msg):
+            self.queue.put(msg)
+        return True
+
+    def run(self):
+        """"Run main loop subscribing to new messages adding these to a message queue."""
+        LOGGER.debug("Starting FileListener.")
+        if type(self.config["subscribe-topic"]) not in (tuple, list, set):
+            self.config["subscribe-topic"] = [self.config["subscribe-topic"]]
+        try:
+            if 'services' not in self.config:
+                self.config['services'] = ''
+            subscriber_addresses = None
+            if 'subscriber_addresses' in self.config:
+                subscriber_addresses = self.config['subscriber_addresses'].split(',')
+
+            with Subscribe(self.config['services'], self.config['subscribe-topic'],
+                           True, addresses=subscriber_addresses,
+                           nameserver=self.subscribe_nameserver) as subscr:
+
+                LOGGER.debug("Entering for loop subscr.recv")
+                for msg in subscr.recv(timeout=1):
+                    if not self._check_and_put_message_to_queue(msg):
+                        break
+
+        except KeyError as ke:
+            LOGGER.error("KeyError: %s", str(ke))
+            raise
+        except KeyboardInterrupt:
+            LOGGER.info("Received keyboard interrupt. Shutting down.")
+            raise
+
+    def check_message(self, msg):
+        """Check the message is valid type for this scope."""
+        if not msg:
+            return False
+        if msg.type not in ('file', 'collection', 'dataset'):
+            LOGGER.debug("Received invalid message type: %s", str(msg.type))
+            return False
+        return True
+
+
+class FilePublisher(Thread):
+
+    """A publisher for result files listening to a publish queue.
+    Publishes the files via posttroll"""
+
+    def __init__(self, queue, nameservers):
+        Thread.__init__(self)
+        self.loop = True
+        self.queue = queue
+        self.service_name = 's3downloader'
+        self.nameservers = nameservers
+
+    def stop(self):
+        """Stops the file publisher."""
+        self.loop = False
+
+    def _publish_message(self, publisher):
+        if not self.loop:
+            return False
+        try:
+            message = self.queue.get(timeout=1)
+        except queue.Empty:
+            message = None
+        if message is not None:
+            LOGGER.info("Publishing new downloaded file(s).")
+            LOGGER.debug(str(message))
+            publisher.send(message)
+        return True
+
+    def run(self):
+        """Run the publish loop getting messages from the publish queue."""
+        try:
+            LOGGER.debug("Using service_name: {} with nameservers {}".format(self.service_name, self.nameservers))
+            with Publish(self.service_name, 0, nameservers=self.nameservers) as publisher:
+                while self.loop:
+                    if not self._publish_message(publisher):
+                        break
+        except KeyboardInterrupt:
+            LOGGER.info("Received keyboard interrupt. Shutting down")
+            raise
+
+
+class S3Downloader():
+    """Listen for posttroll messages and download files as described in the message from the configured
+    object store (s3) bucket. Publish the successfuly downloaded file in as a posttroll message."""
+
+    def __init__(self, cmd_args):
+        self.config = None
+        self.cmd_args = cmd_args
+        self.listener_queue = queue.Queue()
+        self.publisher_queue = queue.Queue()
+        self.listener = None
+        self.publisher = None
+
+    def _stop(self):
+        self.listener.stop()
+        self.publisher.stop()
+        LOGGER.info("Exiting s3downloader.")
+
+    def start(self):
+        """Start the threads for listening and publishing messages and handeling the download."""
+        LOGGER.info("Starting up.")
+        self.listener = Listener(self.listener_queue, self.config,
+                                 subscribe_nameserver=self.cmd_args.subscribe_nameserver)
+        self.listener.start()
+
+        self.publisher = FilePublisher(self.publisher_queue, self.cmd_args.nameservers)
+        self.publisher.start()
+
+        self._read_from_queue()
+
+        self._stop()
+
+    def read_config(self):
+        """Read the config file from cmd args.
+        """
+        if self.cmd_args.config_file and os.path.exists(self.cmd_args.config_file):
+            with open(self.cmd_args.config_file, 'r') as stream:
+                try:
+                    self.config = yaml.safe_load(stream)
+                except FileNotFoundError:
+                    print("Could not find you config file:", self.cmd_args.config_file)
+                    raise
+                except yaml.YAMLError as exc:
+                    print("Failed reading yaml config file: {} with: {}".format(self.cmd_args.config_file, exc))
+                    raise yaml.YAMLError
+        else:
+            raise FileNotFoundError(self.cmd_args.config_file)
+
+        return self.config
+
+    def setup_logging(self):
+        """
+        Init and setup logging.
+        """
+        # Set up logging
+        try:
+            loglevel = logging.INFO
+            if self.cmd_args.log and 'logging' in self.config:
+                ndays = 1
+                ncount = 30
+                try:
+                    ndays = int(self.config['logging']["log_rotation_days"])
+                    ncount = int(self.config['logging']["log_rotation_backup"])
+                except KeyError:
+                    pass
+
+                handler = handlers.TimedRotatingFileHandler(self.cmd_args.log,
+                                                            when='midnight',
+                                                            interval=ndays,
+                                                            backupCount=ncount,
+                                                            encoding=None,
+                                                            delay=False,
+                                                            utc=True)
+
+                handler.doRollover()
+            else:
+                handler = logging.StreamHandler(sys.stderr)
+
+            if 'logging_mode' in self.config['logging'] and self.config['logging']["logging_mode"] == "DEBUG":
+                loglevel = logging.DEBUG
+
+            handler.setLevel(loglevel)
+            logging.getLogger('').setLevel(loglevel)
+            logging.getLogger('').addHandler(handler)
+
+            formatter = logging.Formatter(fmt=_DEFAULT_LOG_FORMAT,
+                                          datefmt=_DEFAULT_TIME_FORMAT)
+            handler.setFormatter(formatter)
+            logging.getLogger('posttroll').setLevel(loglevel)
+            logging.getLogger('botocore').setLevel(loglevel)
+            logging.getLogger('s3transfer').setLevel(loglevel)
+            logging.getLogger('urllib3').setLevel(loglevel)
+            LOGGER = logging.getLogger('S3 downloader')
+        except Exception:
+            print("Logging setup failed. Check your config")
+            raise
+
+        return LOGGER, handler
+
+    def _get_basename(self, uri):
+        up = urlparse(uri)
+        bn = os.path.basename(up.path)
+        return bn
+
+    def _download_from_s3(self, bn):
+        """
+        https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html#aws-config-file
+        An example configuration could be for example placed in `~/.aws/config`
+
+        ```
+        [default]
+        aws_access_key_id=foo
+        aws_secret_access_key=bar
+
+        [profile metno]
+        aws_access_key_id=foo2
+        aws_secret_access_key=bar2
+        ```
+
+        and then in you s3downloader config file:
+        ```
+        s3_kwargs:
+            anon: False
+            profile: metno # Optional
+            client_kwargs:
+                endpoint_url: <url to our objext store>
+        ```
+        """
+        if S3FileSystem is None:
+            raise ImportError("s3downloader requires 's3fs' to be installed.")
+        s3_kwargs = self.config.get('s3_kwargs', {})
+        s3 = S3FileSystem(**s3_kwargs)
+        s3.get_file(os.path.join(self.config['bucket'], bn),
+                    os.path.join(self.config.get('download_destination', '.'), bn))
+        if not os.path.exists(os.path.join(self.config.get('download_destination', '.'), bn)):
+            return False
+        return True
+
+    def _generate_message_if_file_exists_after_download(self, bn, msg):
+        if os.path.exists(os.path.join(self.config.get('download_destination', '.'), bn)):
+            LOGGER.debug("Successfully downloaded file %s to %s", bn, self.config.get('download_destination', '.'))
+            to_send = msg.data.copy()
+            to_send.pop('dataset', None)
+            to_send.pop('collection', None)
+            to_send.pop('filename', None)
+            to_send.pop('compress', None)
+            to_send.pop('tst', None)
+            to_send.pop('uri', None)
+            to_send.pop('uid', None)
+            to_send.pop('file_list', None)
+            to_send.pop('path', None)
+            to_send['uri'] = os.path.join(self.config.get('download_destination', '.'), bn)
+
+            pubmsg = Message(self.config['publish-topic'], "file", to_send).encode()
+            return pubmsg
+        return None
+
+    def _get_one_message(self):
+        LOGGER.debug("Start reading from queue ... ")
+        try:
+            msg = self.listener_queue.get()
+        except KeyboardInterrupt:
+            return False
+        if msg is None:
+            LOGGER.debug("msg is none ... ")
+            return True
+        LOGGER.debug("Read from queue: {}".format(msg))
+        bn = self._get_basename(msg.data['uri'])
+        if self._download_from_s3(bn):
+            pubmsg = self._generate_message_if_file_exists_after_download(bn, msg)
+            LOGGER.info("Sending: %s", str(pubmsg))
+            self.publisher_queue.put(pubmsg)
+        else:
+            LOGGER.error("Could not download file %s for some reason. SKipping this.", bn)
+        return True
+
+    def _read_from_queue(self):
+        running = True
+        while running:
+            if not self._get_one_message():
+                running = False
+
+
+def parse_args(args):
+    """Parse the command line arguments."""
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument("-c", "--config-file",
+                        help="The configuration file to run on.")
+    parser.add_argument("-l", "--log",
+                        help="The file to log to. stdout otherwise.")
+    parser.add_argument("-r", "--subscribe-nameserver",
+                        type=str,
+                        dest='subscribe_nameserver',
+                        default="localhost",
+                        help="subscribe nameserver, defaults to localhost")
+    parser.add_argument("-n", "--nameservers",
+                        type=str,
+                        dest='nameservers',
+                        default=None,
+                        nargs='*',
+                        help="nameservers, defaults to localhost")
+    return parser.parse_args(args)


Test patch:
diff --git a/trollmoves/tests/conftest.py b/trollmoves/tests/conftest.py
index a687ade5..44e73ee9 100644
--- a/trollmoves/tests/conftest.py
+++ b/trollmoves/tests/conftest.py
@@ -3,7 +3,7 @@
 
 def pytest_collection_modifyitems(items):
     """Modifiy test items in place to ensure test modules run in a given order."""
-    MODULE_ORDER = ["test_logging"]
+    MODULE_ORDER = ["test_logging", "test_s3downloader"]
     module_mapping = {item: item.module.__name__ for item in items}
 
     sorted_items = items.copy()
diff --git a/trollmoves/tests/test_s3downloader.py b/trollmoves/tests/test_s3downloader.py
new file mode 100644
index 00000000..8ca095b6
--- /dev/null
+++ b/trollmoves/tests/test_s3downloader.py
@@ -0,0 +1,525 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2022-2023 Pytroll Developers
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Test the s3downloader."""
+
+from logging import StreamHandler
+from unittest.mock import PropertyMock, patch
+from tempfile import NamedTemporaryFile
+import os
+
+import pytest
+from posttroll.message import Message
+
+CONFIG_YAML = """
+logging:
+  log_rotation_days: 1
+  log_rotation_backup: 30
+  logging_mode: DEBUG
+
+subscribe-topic:
+  - /yuhu
+publish-topic: /idnt
+endpoint_url: 'https://your.url.space'
+access_key: 'your_access_key'
+secret_key: 'your_secret_key'
+bucket: atms-sdr
+download_destination: '/destination-directory'
+"""
+
+
+def _write_named_temporary_config(data):
+    with NamedTemporaryFile('w', delete=False) as fid:
+        config_fname = fid.name
+        fid.write(data)
+    return config_fname
+
+
+@pytest.fixture
+def config_yaml():
+    yield _write_named_temporary_config(CONFIG_YAML)
+
+
+def test_read_config(config_yaml):
+    """Test read yaml config."""
+    from trollmoves.s3downloader import parse_args
+    from trollmoves.s3downloader import S3Downloader
+    parse = parse_args(['--config-file=' + config_yaml])
+    s3dl = S3Downloader(parse)
+    config = s3dl.read_config()
+    expected_config = {'logging': {'log_rotation_days': 1, 'log_rotation_backup': 30, 'logging_mode': 'DEBUG'},
+                       'subscribe-topic': ['/yuhu'], 'publish-topic': '/idnt', 'endpoint_url': 'https://your.url.space',
+                       'access_key': 'your_access_key',
+                       'secret_key': 'your_secret_key',
+                       'bucket': 'atms-sdr',
+                       'download_destination': '/destination-directory'}
+    assert config == expected_config
+
+
+@patch('yaml.safe_load')
+def test_read_config_exception(patch_yaml, config_yaml):
+    """Test read yaml config."""
+    from trollmoves.s3downloader import parse_args
+    from trollmoves.s3downloader import S3Downloader
+    parse = parse_args(['--config-file=' + config_yaml])
+    s3dl = S3Downloader(parse)
+    patch_yaml.side_effect = FileNotFoundError
+    with pytest.raises(FileNotFoundError):
+        s3dl.read_config()
+
+
+@patch('yaml.safe_load')
+def test_read_config_exception2(patch_yaml, config_yaml):
+    """Test read yaml config."""
+    from trollmoves.s3downloader import parse_args
+    from trollmoves.s3downloader import S3Downloader
+    parse = parse_args(['--config-file=' + config_yaml])
+    s3dl = S3Downloader(parse)
+    import yaml
+    patch_yaml.side_effect = yaml.YAMLError
+    with pytest.raises(yaml.YAMLError):
+        s3dl.read_config()
+
+
+@pytest.fixture
+def s3dl(config_yaml):
+    from trollmoves.s3downloader import parse_args
+    from trollmoves.s3downloader import S3Downloader
+    parse = parse_args(['--config-file=' + config_yaml])
+    return S3Downloader(parse)
+
+
+@patch('os.path.exists')
+def test_read_config_exception3(patch_os_path_exists, s3dl):
+    """Test read yaml config."""
+    patch_os_path_exists.return_value = False
+    with pytest.raises(FileNotFoundError):
+        s3dl.read_config()
+
+
+def test_get_basename(s3dl):
+    uri = os.path.join("root", "anypath", "filename-basename")
+    bn = s3dl._get_basename(uri)
+    assert bn == 'filename-basename'
+
+
+@patch('os.path.exists')
+def test_generate_message_if_file_exists_after_download(patch_os_path_exists, s3dl):
+    s3dl.read_config()
+    bn = 'filename-basename'
+    to_send = {'some_key': 'with_a_value'}
+    msg = Message('/publish-topic', "file", to_send)
+    patch_os_path_exists.return_value = True
+    pubmsg = s3dl._generate_message_if_file_exists_after_download(bn, msg)
+    assert 'with_a_value' in pubmsg
+
+
+@patch('os.path.exists')
+def test_generate_message_if_file_does_not_exists_after_download(patch_os_path_exists, s3dl):
+    s3dl.read_config()
+    bn = 'filename-basename'
+    to_send = {'some_key': 'with_a_value'}
+    msg = Message('/publish-topic', "file", to_send)
+    patch_os_path_exists.return_value = False
+    pubmsg = s3dl._generate_message_if_file_exists_after_download(bn, msg)
+    assert pubmsg is None
+
+
+@patch('trollmoves.s3downloader.S3Downloader._download_from_s3')
+@patch('trollmoves.s3downloader.S3Downloader._get_basename')
+@patch('queue.Queue')
+def test_get_one_message(patch_subscribe, patch_get_basename, patch_download_from_s3, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    to_send = {'some_key': 'with_a_value', 'uri': 'now-this-is-a-uri'}
+    msg = Message('/publish-topic', "file", to_send)
+    s3dl.listener_queue = patch_subscribe
+    s3dl.listener_queue.get.return_value = msg
+    patch_get_basename.return_value = 'filename-basename'
+    patch_download_from_s3.return_value = True
+    result = s3dl._get_one_message()
+    assert result is True
+
+
+@patch('trollmoves.s3downloader.S3Downloader._download_from_s3')
+@patch('trollmoves.s3downloader.S3Downloader._get_basename')
+@patch('queue.Queue')
+def test_get_one_message_none(patch_sub_q, patch_get_basename, patch_download_from_s3, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    s3dl.listener_queue = patch_sub_q
+    s3dl.listener_queue.get.return_value = None
+    patch_get_basename.return_value = 'filename-basename'
+    patch_download_from_s3.return_value = True
+    result = s3dl._get_one_message()
+    assert result is True
+
+
+@patch('trollmoves.s3downloader.S3Downloader._download_from_s3')
+@patch('trollmoves.s3downloader.S3Downloader._get_basename')
+@patch('queue.Queue')
+def test_get_one_message_download_false(patch_sub_q, patch_get_bn, patch_dl_s3, caplog, s3dl):
+    import logging
+    s3dl.read_config()
+    s3dl.setup_logging()
+    patch_get_bn.return_value = 'filename-basename'
+    patch_dl_s3.return_value = False
+    caplog.set_level(logging.DEBUG)
+    s3dl.listener_queue = patch_sub_q
+    result = s3dl._get_one_message()
+    assert 'Could not download file filename-basename for some reason. SKipping this.' in caplog.text
+    assert result is True
+
+
+@patch('queue.Queue')
+def test_get_one_message_keyboardinterrupt(patch_subscribe, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    s3dl.listener_queue = patch_subscribe
+    s3dl.listener_queue.get.side_effect = KeyboardInterrupt
+    result = s3dl._get_one_message()
+    assert result is False
+
+
+@patch('trollmoves.s3downloader.S3Downloader._get_one_message')
+def test_read_from_queue(patch_get_one_message, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    patch_get_one_message.return_value = False
+    s3dl._read_from_queue()
+    # TODO: what does this tests?
+
+
+@patch('os.path.exists')
+@patch('s3fs.S3FileSystem')
+@patch('s3fs.S3FileSystem.get_file')
+def test_download_from_s3(patch_get_file, patch_S3FileSystem, patch_exists, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    bn = 'filename-basename'
+    result = s3dl._download_from_s3(bn)
+    assert result is True
+
+
+@patch('os.path.exists')
+@patch('s3fs.S3FileSystem')
+@patch('s3fs.S3FileSystem.get_file')
+def test_download_from_s3_false(patch_get_file, patch_S3FileSystem, patch_exists, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    bn = 'filename-basename'
+    patch_exists.return_value = False
+    result = s3dl._download_from_s3(bn)
+    assert result is False
+
+
+def test_setup_logging(s3dl):
+    import logging
+    s3dl.read_config()
+
+    LOGGER, handler = s3dl.setup_logging()
+    assert isinstance(LOGGER, logging.Logger) is True
+    assert logging.DEBUG == handler.level
+    assert isinstance(handler, StreamHandler) is True
+
+
+@patch('logging.StreamHandler')
+def test_setup_logging_exception(patch_stream_handler, s3dl):
+    s3dl.read_config()
+    patch_stream_handler.side_effect = Exception
+    with pytest.raises(Exception):
+        s3dl.setup_logging()
+
+
+def test_setup_logging_file(config_yaml):
+    import logging
+    from trollmoves.s3downloader import parse_args
+    from trollmoves.s3downloader import S3Downloader
+    with NamedTemporaryFile('w', delete=False) as fid:
+        config_fname = fid.name
+    parse = parse_args(['--config-file=' + config_yaml, '-l=' + config_fname])
+    s3dl = S3Downloader(parse)
+    s3dl.read_config()
+
+    LOGGER, handler = s3dl.setup_logging()
+    assert isinstance(LOGGER, logging.Logger) is True
+    assert logging.DEBUG == handler.level
+    assert isinstance(handler, logging.handlers.TimedRotatingFileHandler) is True
+
+    s3dl.config['logging'].pop('log_rotation_days')
+    LOGGER, handler = s3dl.setup_logging()
+    assert handler.interval == 60 * 60 * 24
+    assert handler.backupCount == 30
+
+
+@patch('posttroll.publisher.Publish')
+@patch('queue.Queue')
+def test_file_publisher_init(patch_publish_queue, patch_publish):
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    fp = FilePublisher(patch_publish_queue, nameservers)
+    assert fp.loop is True
+    assert fp.service_name == 's3downloader'
+    assert fp.nameservers == nameservers
+    assert fp.queue == patch_publish_queue
+
+
+MSG_1 = Message('/topic', 'file', data={'uid': 'file1'})
+
+
+@patch('trollmoves.s3downloader.Publish')
+@patch('queue.Queue')
+def test_file_publisher_break(patch_publish_queue, patch_publish):
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    patch_publish_queue.get = PropertyMock(side_effect=[[MSG_1.encode(), None], ])
+    fp = FilePublisher(patch_publish_queue, nameservers)
+    fp.loop = False
+    fp.run()
+    patch_publish().__enter__().send.assert_not_called()
+
+
+@patch('trollmoves.s3downloader.Publish')
+@patch('queue.Queue')
+def test_file_publisher_publish_message(patch_publish_queue, patch_publish):
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    patch_publish_queue.get = PropertyMock(side_effect=[[MSG_1.encode()]])
+    fp = FilePublisher(patch_publish_queue, nameservers)
+    fp._publish_message(patch_publish)
+    patch_publish.send.assert_called_once()
+
+
+@patch('trollmoves.s3downloader.Publish')
+@patch('queue.Queue')
+def test_file_publisher_message_is_none(patch_publish_queue, patch_publish):
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    patch_publish_queue.get = PropertyMock(side_effect=[None, ])
+    fp = FilePublisher(patch_publish_queue, nameservers)
+    fp._publish_message(patch_publish)
+    patch_publish.send.assert_not_called()
+
+
+@patch('trollmoves.s3downloader.Publish')
+def test_file_publisher_stop_loop(patch_publish):
+    import queue
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    pqueue = queue.Queue()
+    fp = FilePublisher(pqueue, nameservers)
+    fp.stop()
+    assert fp.loop is False
+
+
+@patch('trollmoves.s3downloader.Publish')
+@patch('queue.Queue')
+def test_file_publisher_queue_timeout(patch_publish_queue, patch_publish):
+    import queue
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    patch_publish_queue.get.side_effect = queue.Empty
+    fp = FilePublisher(patch_publish_queue, nameservers)
+    fp._publish_message(patch_publish)
+    patch_publish.send.assert_not_called()
+
+
+@patch('trollmoves.s3downloader.Publish')
+@patch('queue.Queue')
+def test_file_publisher_exception_1(patch_publish_queue, patch_publish):
+    from trollmoves.s3downloader import FilePublisher
+    nameservers = None
+    patch_publish_queue.get.side_effect = KeyboardInterrupt
+    fp = FilePublisher(patch_publish_queue, nameservers)
+    with pytest.raises(KeyboardInterrupt):
+        fp.run()
+
+
+posttroll_config = {'subscribe-topic': '/yuhu'}
+
+
+@patch('queue.Queue')
+def test_listener_init(patch_listener_queue):
+    from trollmoves.s3downloader import Listener
+    subscribe_nameserver = 'localhost'
+    listenr = Listener(patch_listener_queue, posttroll_config, subscribe_nameserver)
+    assert listenr.loop is True
+    assert listenr.queue == patch_listener_queue
+    assert listenr.config == posttroll_config
+    assert listenr.subscribe_nameserver == subscribe_nameserver
+
+
+@patch('posttroll.subscriber.Subscriber')
+@patch('posttroll.subscriber.get_pub_address')
+def test_listener_message(patch_get_pub_address, patch_subscriber, caplog):
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import logging
+    import queue
+    subscribe_nameserver = 'localhost'
+    caplog.set_level(logging.DEBUG)
+
+    patch_subscriber.return_value.recv = PropertyMock(side_effect=[[MSG_1, None], ])
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+    listener.run()
+
+    assert 'Starting FileListener.' in caplog.text
+    assert lqueue.qsize() == 1
+
+    message = lqueue.get()
+    assert message.type == 'file'
+
+
+@patch('posttroll.subscriber.Subscriber')
+@patch('posttroll.subscriber.get_pub_address')
+@patch('queue.Queue')
+def test_listener_message_break(patch_listener_queue, patch_get_pub_address, patch_subscriber, caplog):
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import logging
+    caplog.set_level(logging.DEBUG)
+    subscribe_nameserver = 'localhost'
+
+    patch_subscriber.return_value.recv = PropertyMock(side_effect=[[MSG_1, None], ])
+    listener = Listener(patch_listener_queue, posttroll_config, subscribe_nameserver)
+    listener.loop = False
+    listener.run()
+    patch_listener_queue().put.assert_not_called()
+
+
+MSG_ACK = Message('/topic', 'ack', data={'uid': 'file1'})
+
+
+def test_listener_message_check_message():
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import queue
+    subscribe_nameserver = 'localhost'
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+
+    assert listener.check_message(None) is False
+    assert listener.check_message(MSG_ACK) is False
+    assert listener.check_message(MSG_1) is True
+
+
+def test_listener_message_stop():
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import queue
+    subscribe_nameserver = 'localhost'
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+
+    listener.stop()
+    assert listener.loop is False
+    assert listener.queue.qsize() == 1
+    message = lqueue.get()
+    assert message is None
+
+
+@patch('posttroll.subscriber.Subscriber')
+@patch('posttroll.subscriber.get_pub_address')
+def test_listener_message_check_config(patch_get_pub_address, patch_subscriber):
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import queue
+    posttroll_config['subscribe-topic'] = 'is-a-string-topic'
+    posttroll_config['subscriber_addresses'] = 'first_address, second_address'
+    subscribe_nameserver = 'localhost'
+
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+    listener.run()
+    assert isinstance(listener.config["subscribe-topic"], list) is True
+    assert listener.config["services"] == ''
+
+
+@patch('posttroll.subscriber.Subscriber')
+@patch('posttroll.subscriber.get_pub_address')
+def test_listener_message_check_message_and_put(patch_get_pub_address, patch_subscriber):
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import queue
+    posttroll_config['subscribe-topic'] = 'is-a-string-topic'
+    posttroll_config['subscriber_addresses'] = 'first_address, second_address'
+    subscribe_nameserver = 'localhost'
+
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+    assert listener._check_and_put_message_to_queue(MSG_1) is True
+    assert listener._check_and_put_message_to_queue(None) is True
+
+    listener.loop = False
+    assert listener._check_and_put_message_to_queue(MSG_1) is False
+
+
+@patch('posttroll.subscriber.Subscriber')
+@patch('posttroll.subscriber.get_pub_address')
+def test_listener_message_exception_1(patch_get_pub_address, patch_subscriber):
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import queue
+    subscribe_nameserver = 'localhost'
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+    patch_subscriber.side_effect = KeyError
+    with pytest.raises(KeyError):
+        listener.run()
+
+
+@patch('posttroll.subscriber.Subscriber')
+@patch('posttroll.subscriber.get_pub_address')
+def test_listener_message_exception_2(patch_get_pub_address, patch_subscriber):
+    """Test listener push message."""
+    from trollmoves.s3downloader import Listener
+    import queue
+    subscribe_nameserver = 'localhost'
+    lqueue = queue.Queue()
+    listener = Listener(lqueue, posttroll_config, subscribe_nameserver)
+    patch_subscriber.side_effect = KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        listener.run()
+
+
+@patch('trollmoves.s3downloader.S3Downloader._get_one_message')
+@patch('trollmoves.s3downloader.FilePublisher')
+@patch('trollmoves.s3downloader.Listener')
+def test_start_stop(patch_listener, patch_publisher, patch_get_one_message, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    patch_get_one_message.return_value = False
+    s3dl.start()
+
+    patch_listener().start.assert_called_once()
+    patch_publisher().start.assert_called_once()
+    patch_get_one_message.assert_called_once()
+    patch_listener().stop.assert_called_once()
+    patch_publisher().stop.assert_called_once()
+
+
+@patch('trollmoves.s3downloader.FilePublisher')
+@patch('trollmoves.s3downloader.Listener')
+def test_stop(patch_listener, patch_publisher, s3dl):
+    s3dl.read_config()
+    s3dl.setup_logging()
+    s3dl.listener = patch_listener
+    s3dl.publisher = patch_publisher
+    s3dl._stop()
+    patch_listener.stop.assert_called_once()
+    patch_publisher.stop.assert_called_once()