Repository: RaRe-Technologies/smart_open (python)

Issue:
Probably avoidable fatal S3 race condition
#### Problem description

When using `s3.iter_bucket` from my Lambda, I observed the following problem.

A user on my system was deleting an object (valid use case). While that user was deleting it, unrelated to it, `smart_open.s3.iter_bucket` kicked off. Inside that function, the key iterator is created, and then a download is started for each key. The key (of the object about to be deleted) showed up in the result. Once it was time to download the object, it was no longer there, and a 404 was thrown, which proved fatal and an exception was thrown.

While this may seem like a fluke,

a) it was on a system with very little user activity.
b) S3's list operation is (iirc) not strongly consistent and may for a certain duration return keys that have already been deleted.

So it may be more common than it seems. I think that it would be reasonable for `iter_bucket` to skip objects that return a 404 during their download (ie: catch the error and suppress it), and carry on iterating. I think `iter_bucket` would still fulfill its duty this way.

#### Steps/code to reproduce the problem

The code:

```python
def get_all_metadata() -> list[dict]:
    bucket = get_s3_bucket()
    prefix = f"{get_s3_path()}/"
    accept_key = lambda key: key.endswith(get_metadata_ext())

    # Not explicitly setting this property to False will cause an error in the Lambda runtime when
    # running iter_bucket. This is a known issue, but it doesn't look like it will be resolved:
    # https://github.com/RaRe-Technologies/smart_open/issues/605

    smart_open.concurrency._MULTIPROCESSING = False

    return [
        json.loads(content)
        for key, content in smart_open.s3.iter_bucket(
            bucket, prefix=prefix, accept_key=accept_key
        )
    ]
```

The error:

```
[ERROR] ClientError: An error occurred (404) when calling the HeadObject operation: Not Found
Traceback (most recent call last):
  File "/var/task/lambdas/filemanager/list.py", line 20, in handler
    metadata_list = filemanager.get_all_metadata()
  File "/var/task/lambdas/lib/filemanager.py", line 78, in get_all_metadata
    return [
  File "/var/task/lambdas/lib/filemanager.py", line 78, in <listcomp>
    return [
  File "/var/task/smart_open/s3.py", line 1192, in iter_bucket
    for key_no, (key, content) in enumerate(result_iterator):
  File "/var/task/smart_open/concurrency.py", line 58, in imap_unordered
    yield future.result()
  File "/var/lang/lib/python3.9/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/var/lang/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/var/lang/lib/python3.9/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/var/task/smart_open/s3.py", line 1254, in _download_key
    content_bytes = _download_fileobj(bucket, key_name)
  File "/var/task/smart_open/s3.py", line 1271, in _download_fileobj
    bucket.download_fileobj(key_name, buf)
  File "/var/runtime/boto3/s3/inject.py", line 719, in bucket_download_fileobj
    return self.meta.client.download_fileobj(
  File "/var/runtime/boto3/s3/inject.py", line 679, in download_fileobj
    return future.result()
  File "/var/runtime/s3transfer/futures.py", line 103, in result
    return self._coordinator.result()
  File "/var/runtime/s3transfer/futures.py", line 266, in result
    raise self._exception
  File "/var/runtime/s3transfer/tasks.py", line 269, in _main
    self._submit(transfer_future=transfer_future, **kwargs)
  File "/var/runtime/s3transfer/download.py", line 354, in _submit
    response = client.head_object(
  File "/var/runtime/botocore/client.py", line 391, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/var/runtime/botocore/client.py", line 719, in _make_api_call
    raise error_class(parsed_response, operation_name)
```

#### Versions

This is running on:

* AWS Lambda
* Intel
* Python 3.9
* smart_open 5.2.1

#### Checklist

Before you create the issue, please make sure you have:

- [x] Described the problem clearly
- [x] Provided a minimal reproducible example, including any required data
- [x] Provided the version numbers of the relevant software

Pull request:
Fix avoidable S3 race condition (#693)
#### Motivation

Fixes #693 

As proposed by @ronkorving, it may be common for the s3 `iter_bucket` function to encounter `404 not found` errors due to the s3 `list bucket` command sometimes returning recently deleted keys. So a workaround of this problem is to surpress 404 errors for `iter_bucket`, and instead log them as warnings for the user to assess.

#### Tests

All existing tests related to `iter_bucket` within `s3.py` pass. I also added two new tests: `test_iter_bucket_404` and `test_iter_bucket_non_404`. The first test simulates the scenario with `iter_bucket` for a key throws a 404 error, and expects that error to be suppressed and for computation to carry on for the rest of the keys. The latter test simulates the scenario where a non 404 error occurs, in which case it should not be suppressed. 


Patch:
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 23cfa7d..c2cb5b3 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -138,7 +138,9 @@ jobs:
       - run: bash ci_helpers/helpers.sh enable_moto_server
         if: ${{ matrix.moto_server }}
       
-      - run: bash ci_helpers/helpers.sh create_ftp_server
+      - run: |
+          sudo apt-get install vsftpd
+          sudo bash ci_helpers/helpers.sh create_ftp_ftps_servers
 
       - name: Run integration tests
         run: python ci_helpers/run_integration_tests.py
@@ -149,7 +151,7 @@ jobs:
       - run: bash ci_helpers/helpers.sh disable_moto_server
         if: ${{ matrix.moto_server }}
       
-      - run: bash ci_helpers/helpers.sh delete_ftp_server
+      - run: sudo bash ci_helpers/helpers.sh delete_ftp_ftps_servers
 
   benchmarks:
     needs: [linters,unit_tests]
diff --git a/.gitignore b/.gitignore
index 485c692..3aa10d6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -65,3 +65,4 @@ target/
 
 # env files
 .env
+.venv
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..c55d71e
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,28 @@
+# Quickstart
+
+Clone the repo and use a python installation to create a venv:
+
+```sh
+git clone git@github.com:RaRe-Technologies/smart_open.git
+cd smart_open
+python -m venv .venv
+```
+
+Activate the venv to start working and install test deps:
+
+```sh
+.venv/bin/activate
+pip install -e ".[test]"
+```
+
+Tests should pass:
+
+```sh
+pytest
+```
+
+Thats it! When you're done, deactivate the venv:
+
+```sh
+deactivate
+```
diff --git a/ci_helpers/helpers.sh b/ci_helpers/helpers.sh
old mode 100755
new mode 100644
index c24567a..4533b33
--- a/ci_helpers/helpers.sh
+++ b/ci_helpers/helpers.sh
@@ -7,17 +7,52 @@ enable_moto_server(){
   moto_server -p5000 2>/dev/null&
 }
 
-create_ftp_server(){
-  docker run -d -p 21:21 -p 21000-21010:21000-21010 -e USERS="user|123|/home/user/dir" -e ADDRESS=localhost --name my-ftp-server delfer/alpine-ftp-server 
+create_ftp_ftps_servers(){
+  #
+  # Must be run as root
+  #
+  home_dir=/home/user
+  user=user
+  pass=123
+  ftp_port=21
+  ftps_port=90
+
+  mkdir $home_dir
+  useradd -p $(echo $pass | openssl passwd -1 -stdin) -d $home_dir $user
+  chown $user:$user $home_dir
+
+  server_setup='''
+listen=YES
+listen_ipv6=NO
+write_enable=YES
+pasv_enable=YES
+pasv_min_port=40000
+pasv_max_port=40009
+chroot_local_user=YES
+allow_writeable_chroot=YES'''
+
+  additional_ssl_setup='''
+ssl_enable=YES
+allow_anon_ssl=NO
+force_local_data_ssl=NO
+force_local_logins_ssl=NO
+require_ssl_reuse=NO
+'''
+
+  cp /etc/vsftpd.conf /etc/vsftpd-ssl.conf
+  echo -e "$server_setup\nlisten_port=${ftp_port}" >> /etc/vsftpd.conf
+  echo -e "$server_setup\nlisten_port=${ftps_port}\n$additional_ssl_setup" >> /etc/vsftpd-ssl.conf
+
+  service vsftpd restart
+  vsftpd /etc/vsftpd-ssl.conf &
 }
 
 disable_moto_server(){
   lsof -i tcp:5000 | tail -n1 | cut -f2 -d" " | xargs kill -9
 }
 
-delete_ftp_server(){
-  docker kill my-ftp-server
-  docker rm my-ftp-server
+delete_ftp_ftps_servers(){
+  service vsftpd stop
 }
 
 "$@"
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..6cab0ef
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,2 @@
+[tool.pytest.ini_options]
+testpaths = ["smart_open"]
diff --git a/setup.py b/setup.py
index 6c08ade..5bd4621 100644
--- a/setup.py
+++ b/setup.py
@@ -40,13 +40,13 @@ aws_deps = ['boto3']
 gcs_deps = ['google-cloud-storage>=1.31.0']
 azure_deps = ['azure-storage-blob', 'azure-common', 'azure-core']
 http_deps = ['requests']
+ssh_deps = ['paramiko']
 
-all_deps = aws_deps + gcs_deps + azure_deps + http_deps
+all_deps = aws_deps + gcs_deps + azure_deps + http_deps + ssh_deps
 tests_require = all_deps + [
     'moto[server]',
     'responses',
     'boto3',
-    'paramiko',
     'pytest',
     'pytest-rerunfailures'
 ]
@@ -79,6 +79,7 @@ setup(
         'all': all_deps,
         'http': http_deps,
         'webhdfs': http_deps,
+        'ssh': ssh_deps,
     },
     python_requires=">=3.6,<4.0",
 
diff --git a/smart_open/azure.py b/smart_open/azure.py
index 15168a9..96f944a 100644
--- a/smart_open/azure.py
+++ b/smart_open/azure.py
@@ -11,7 +11,6 @@
 import base64
 import io
 import logging
-from typing import Union
 
 import smart_open.bytebuffer
 import smart_open.constants
diff --git a/smart_open/compression.py b/smart_open/compression.py
index fe02246..08469e7 100644
--- a/smart_open/compression.py
+++ b/smart_open/compression.py
@@ -62,6 +62,7 @@ def register_compressor(ext, callback):
     """
     if not (ext and ext[0] == '.'):
         raise ValueError('ext must be a string starting with ., not %r' % ext)
+    ext = ext.lower()
     if ext in _COMPRESSOR_REGISTRY:
         logger.warning('overriding existing compression handler for %r', ext)
     _COMPRESSOR_REGISTRY[ext] = callback
@@ -103,24 +104,23 @@ def _handle_gzip(file_obj, mode):
     return result
 
 
-def compression_wrapper(file_obj, mode, compression):
+def compression_wrapper(file_obj, mode, compression=INFER_FROM_EXTENSION, filename=None):
     """
-    This function will wrap the file_obj with an appropriate
-    [de]compression mechanism based on the specified extension.
+    Wrap `file_obj` with an appropriate [de]compression mechanism based on its file extension.
 
-    file_obj must either be a filehandle object, or a class which behaves
-    like one. It must have a .name attribute.
+    If the filename extension isn't recognized, simply return the original `file_obj` unchanged.
 
-    If the filename extension isn't recognized, will simply return the original
-    file_obj.
+    `file_obj` must either be a filehandle object, or a class which behaves like one.
+
+    If `filename` is specified, it will be used to extract the extension.
+    If not, the `file_obj.name` attribute is used as the filename.
 
     """
     if compression == NO_COMPRESSION:
         return file_obj
     elif compression == INFER_FROM_EXTENSION:
         try:
-            filename = file_obj.name
-            filename.upper()  # make sure this thing is a string
+            filename = (filename or file_obj.name).lower()
         except (AttributeError, TypeError):
             logger.warning(
                 'unable to transparently decompress %r because it '
diff --git a/smart_open/ftp.py b/smart_open/ftp.py
index 36d86bb..3dbe26f 100644
--- a/smart_open/ftp.py
+++ b/smart_open/ftp.py
@@ -12,11 +12,11 @@
 import logging
 import urllib.parse
 import smart_open.utils
-from ftplib import FTP, error_reply
+from ftplib import FTP, FTP_TLS, error_reply
 import types
 logger = logging.getLogger(__name__)
 
-SCHEME = "ftp"
+SCHEMES = ("ftp", "ftps")
 
 """Supported URL schemes."""
 
@@ -26,6 +26,9 @@ URI_EXAMPLES = (
     "ftp://username@host/path/file",
     "ftp://username:password@host/path/file",
     "ftp://username:password@host:port/path/file",
+    "ftps://username@host/path/file",
+    "ftps://username:password@host/path/file",
+    "ftps://username:password@host:port/path/file",
 )
 
 
@@ -35,7 +38,7 @@ def _unquote(text):
 
 def parse_uri(uri_as_string):
     split_uri = urllib.parse.urlsplit(uri_as_string)
-    assert split_uri.scheme == SCHEME, 'unexpected scheme: %r' % split_uri.scheme
+    assert split_uri.scheme in SCHEMES
     return dict(
         scheme=split_uri.scheme,
         uri_path=_unquote(split_uri.path),
@@ -50,8 +53,10 @@ def open_uri(uri, mode, transport_params):
     smart_open.utils.check_kwargs(open, transport_params)
     parsed_uri = parse_uri(uri)
     uri_path = parsed_uri.pop("uri_path")
-    parsed_uri.pop("scheme")
-    return open(uri_path, mode, transport_params=transport_params, **parsed_uri)
+    scheme = parsed_uri.pop("scheme")
+    secure_conn = True if scheme == "ftps" else False
+    return open(uri_path, mode, secure_connection=secure_conn,
+            transport_params=transport_params, **parsed_uri)
 
 
 def convert_transport_params_to_args(transport_params):
@@ -71,9 +76,12 @@ def convert_transport_params_to_args(transport_params):
     return kwargs
 
 
-def _connect(hostname, username, port, password, transport_params):
+def _connect(hostname, username, port, password, secure_connection, transport_params):
     kwargs = convert_transport_params_to_args(transport_params)
-    ftp = FTP(**kwargs)
+    if secure_connection:
+        ftp = FTP_TLS(**kwargs)
+    else:
+        ftp = FTP(**kwargs)
     try:
         ftp.connect(hostname, port)
     except Exception as e:
@@ -84,10 +92,11 @@ def _connect(hostname, username, port, password, transport_params):
     except error_reply as e:
         logger.error("Unable to login to FTP server: try checking the username and password!")
         raise e
+    if secure_connection:
+        ftp.prot_p()
     return ftp
 
 
-# transport paramaters can include any extra parameters that you want to be passed into FTP_TLS
 def open(
     path,
     mode="r",
@@ -95,22 +104,42 @@ def open(
     user=None,
     password=None,
     port=DEFAULT_PORT,
+    secure_connection=False,
     transport_params=None,
 ):
+    """Open a file for reading or writing via FTP/FTPS.
+
+    Parameters
+    ----------
+    path: str
+        The path on the remote server
+    mode: str
+        Must be "rb" or "wb"
+    host: str
+        The host to connect to
+    user: str
+        The username to use for the connection
+    password: str
+        The password for the specified username
+    port: int
+        The port to connect to
+    secure_connection: bool
+        True for FTPS, False for FTP
+    transport_params: dict
+        Additional parameters for the FTP connection.
+        Currently supported parameters: timeout, source_address, encoding.
+    """
     if not host:
         raise ValueError("you must specify the host to connect to")
     if not user:
         raise ValueError("you must specify the user")
     if not transport_params:
         transport_params = {}
-    conn = _connect(host, user, port, password, transport_params)
+    conn = _connect(host, user, port, password, secure_connection, transport_params)
     mode_to_ftp_cmds = {
-        "r": ("RETR", "r"),
         "rb": ("RETR", "rb"),
-        "w": ("STOR", "w"),
         "wb": ("STOR", "wb"),
-        "a": ("APPE", "w"),
-        "ab": ("APPE", "wb")
+        "ab": ("APPE", "wb"),
     }
     try:
         ftp_mode, file_obj_mode = mode_to_ftp_cmds[mode]
diff --git a/smart_open/s3.py b/smart_open/s3.py
index c5959bd..b211163 100644
--- a/smart_open/s3.py
+++ b/smart_open/s3.py
@@ -1189,18 +1189,31 @@ def iter_bucket(
 
     with smart_open.concurrency.create_pool(processes=workers) as pool:
         result_iterator = pool.imap_unordered(download_key, key_iterator)
-        for key_no, (key, content) in enumerate(result_iterator):
-            if True or key_no % 1000 == 0:
-                logger.info(
-                    "yielding key #%i: %s, size %i (total %.1fMB)",
-                    key_no, key, len(content), total_size / 1024.0 ** 2
-                )
-            yield key, content
-            total_size += len(content)
-
-            if key_limit is not None and key_no + 1 >= key_limit:
-                # we were asked to output only a limited number of keys => we're done
+        key_no = 0
+        while True:
+            try:
+                (key, content) = result_iterator.__next__()
+                if key_no % 1000 == 0:
+                    logger.info(
+                        "yielding key #%i: %s, size %i (total %.1fMB)",
+                        key_no, key, len(content), total_size / 1024.0 ** 2
+                    )
+                yield key, content
+                total_size += len(content)
+                if key_limit is not None and key_no + 1 >= key_limit:
+                    # we were asked to output only a limited number of keys => we're done
+                    break
+            except botocore.exceptions.ClientError as err:
+                #
+                # ignore 404 not found errors: they mean the object was deleted
+                # after we listed the contents of the bucket, but before we
+                # downloaded the object.
+                #
+                if not ('Error' in err.response and err.response['Error'].get('Code') == '404'):
+                    raise err
+            except StopIteration:
                 break
+            key_no += 1
     logger.info("processed %i keys, total size %i" % (key_no + 1, total_size))
 
 
diff --git a/smart_open/ssh.py b/smart_open/ssh.py
index fa762eb..4b31893 100644
--- a/smart_open/ssh.py
+++ b/smart_open/ssh.py
@@ -25,7 +25,11 @@ Similarly, from a command line::
 import getpass
 import logging
 import urllib.parse
-import warnings
+
+try:
+    import paramiko
+except ImportError:
+    MISSING_DEPS = True
 
 import smart_open.utils
 
@@ -74,29 +78,15 @@ def open_uri(uri, mode, transport_params):
     return open(uri_path, mode, transport_params=transport_params, **parsed_uri)
 
 
-def _connect(hostname, username, port, password, transport_params):
-    try:
-        import paramiko
-    except ImportError:
-        warnings.warn(
-            'paramiko missing, opening SSH/SCP/SFTP paths will be disabled. '
-            '`pip install paramiko` to suppress'
-        )
-        raise
-
-    key = (hostname, username)
-    ssh = _SSH.get(key)
-    if ssh is None:
-        ssh = _SSH[key] = paramiko.client.SSHClient()
-        ssh.load_system_host_keys()
-        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-        kwargs = transport_params.get('connect_kwargs', {}).copy()
-        # if 'key_filename' is present in transport_params, then I do not
-        #   overwrite the credentials.
-        if 'key_filename' not in kwargs:
-            kwargs.setdefault('password', password)
-        kwargs.setdefault('username', username)
-        ssh.connect(hostname, port, **kwargs)
+def _connect_ssh(hostname, username, port, password, transport_params):
+    ssh = paramiko.SSHClient()
+    ssh.load_system_host_keys()
+    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    kwargs = transport_params.get('connect_kwargs', {}).copy()
+    if 'key_filename' not in kwargs:
+        kwargs.setdefault('password', password)
+    kwargs.setdefault('username', username)
+    ssh.connect(hostname, port, **kwargs)
     return ssh
 
 
@@ -142,8 +132,30 @@ def open(path, mode='r', host=None, user=None, password=None, port=DEFAULT_PORT,
     if not transport_params:
         transport_params = {}
 
-    conn = _connect(host, user, port, password, transport_params)
-    sftp_client = conn.get_transport().open_sftp_client()
+    key = (host, user)
+
+    attempts = 2
+    for attempt in range(attempts):
+        try:
+            ssh = _SSH[key]
+        except KeyError:
+            ssh = _SSH[key] = _connect_ssh(host, user, port, password, transport_params)
+
+        try:
+            transport = ssh.get_transport()
+            sftp_client = transport.open_sftp_client()
+            break
+        except paramiko.SSHException as ex:
+            connection_timed_out = ex.args and ex.args[0] == 'SSH session not active'
+            if attempt == attempts - 1 or not connection_timed_out:
+                raise
+
+            #
+            # Try again.  Delete the connection from the cache to force a
+            # reconnect in the next attempt.
+            #
+            del _SSH[key]
+
     fobj = sftp_client.open(path, mode)
     fobj.name = path
     return fobj


Test patch:
diff --git a/integration-tests/test_ftp.py b/integration-tests/test_ftp.py
index 79129dd..000faae 100644
--- a/integration-tests/test_ftp.py
+++ b/integration-tests/test_ftp.py
@@ -1,71 +1,84 @@
 from __future__ import unicode_literals
-
+import pytest
 from smart_open import open
 
-def test_nonbinary():
+
+@pytest.fixture(params=[("ftp", 21), ("ftps", 90)])
+def server_info(request):
+    return request.param
+
+def test_nonbinary(server_info):
+    server_type = server_info[0]
+    port_num = server_info[1]
     file_contents = "Test Test \n new test \n another tests"
     appended_content1 = "Added \n to end"
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file", "w") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file", "w") as f:
         f.write(file_contents)
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file", "r") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file", "r") as f:
         read_contents = f.read()
         assert read_contents == file_contents
     
-    with open("ftp://user:123@localhost:21/home/user/dir/file", "a") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file", "a") as f:
         f.write(appended_content1)
     
-    with open("ftp://user:123@localhost:21/home/user/dir/file", "r") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file", "r") as f:
         read_contents = f.read()
         assert read_contents == file_contents + appended_content1
 
-def test_binary():
+def test_binary(server_info):
+    server_type = server_info[0]
+    port_num = server_info[1]
     file_contents = b"Test Test \n new test \n another tests"
     appended_content1 = b"Added \n to end"
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file2", "wb") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file2", "wb") as f:
         f.write(file_contents)
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file2", "rb") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file2", "rb") as f:
         read_contents = f.read()
         assert read_contents == file_contents
     
-    with open("ftp://user:123@localhost:21/home/user/dir/file2", "ab") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file2", "ab") as f:
         f.write(appended_content1)
     
-    with open("ftp://user:123@localhost:21/home/user/dir/file2", "rb") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file2", "rb") as f:
         read_contents = f.read()
         assert read_contents == file_contents + appended_content1
 
-def test_line_endings_non_binary():
+def test_line_endings_non_binary(server_info):
+    server_type = server_info[0]
+    port_num = server_info[1]
     B_CLRF = b'\r\n'
     CLRF = '\r\n'
     file_contents = f"Test Test {CLRF} new test {CLRF} another tests{CLRF}"
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file3", "w") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file3", "w") as f:
         f.write(file_contents)
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file3", "r") as f:    
+    with open(f"{server_type}://user:123@localhost:{port_num}/file3", "r") as f:    
         for line in f:
             assert not CLRF in line
     
-    with open("ftp://user:123@localhost:21/home/user/dir/file3", "rb") as f:    
+    with open(f"{server_type}://user:123@localhost:{port_num}/file3", "rb") as f:    
         for line in f:
             assert B_CLRF in line
 
-def test_line_endings_binary():
+def test_line_endings_binary(server_info):
+    server_type = server_info[0]
+    port_num = server_info[1]
     B_CLRF = b'\r\n'
     CLRF = '\r\n'
     file_contents = f"Test Test {CLRF} new test {CLRF} another tests{CLRF}".encode('utf-8')
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file4", "wb") as f:
+    with open(f"{server_type}://user:123@localhost:{port_num}/file4", "wb") as f:
         f.write(file_contents)
 
-    with open("ftp://user:123@localhost:21/home/user/dir/file4", "r") as f:    
+    with open(f"{server_type}://user:123@localhost:{port_num}/file4", "r") as f:    
         for line in f:
             assert not CLRF in line
     
-    with open("ftp://user:123@localhost:21/home/user/dir/file4", "rb") as f:    
+    with open(f"{server_type}://user:123@localhost:{port_num}/file4", "rb") as f:    
         for line in f:
             assert B_CLRF in line
\ No newline at end of file
diff --git a/integration-tests/test_ssh.py b/integration-tests/test_ssh.py
new file mode 100644
index 0000000..252a6fb
--- /dev/null
+++ b/integration-tests/test_ssh.py
@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 Radim Rehurek <me@radimrehurek.com>
+#
+# This code is distributed under the terms and conditions
+# from the MIT License (MIT).
+#
+
+import os
+import tempfile
+import pytest
+
+import smart_open
+import smart_open.ssh
+
+
+def explode(*args, **kwargs):
+    raise RuntimeError("this function should never have been called")
+
+
+@pytest.mark.skipif("SMART_OPEN_SSH" not in os.environ, reason="this test only works on the dev machine")
+def test():
+    with smart_open.open("ssh://misha@localhost/Users/misha/git/smart_open/README.rst") as fin:
+        readme = fin.read()
+
+    assert 'smart_open — utils for streaming large files in Python' in readme
+
+    #
+    # Ensure the cache is being used
+    #
+    assert ('localhost', 'misha') in smart_open.ssh._SSH
+
+    try:
+        connect_ssh = smart_open.ssh._connect_ssh
+        smart_open.ssh._connect_ssh = explode
+
+        with smart_open.open("ssh://misha@localhost/Users/misha/git/smart_open/howto.md") as fin:
+            howto = fin.read()
+
+        assert 'How-to Guides' in howto
+    finally:
+        smart_open.ssh._connect_ssh = connect_ssh
diff --git a/smart_open/tests/test_compression.py b/smart_open/tests/test_compression.py
new file mode 100644
index 0000000..4b93f63
--- /dev/null
+++ b/smart_open/tests/test_compression.py
@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2020 Radim Rehurek <me@radimrehurek.com>
+#
+# This code is distributed under the terms and conditions
+# from the MIT License (MIT).
+#
+import io
+import gzip
+import pytest
+
+import smart_open.compression
+
+
+plain = 'доброе утро планета!'.encode()
+
+
+def label(thing, name):
+    setattr(thing, 'name', name)
+    return thing
+
+
+@pytest.mark.parametrize(
+    'fileobj,compression,filename',
+    [
+        (io.BytesIO(plain), 'disable', None),
+        (io.BytesIO(plain), 'disable', ''),
+        (io.BytesIO(plain), 'infer_from_extension', 'file.txt'),
+        (io.BytesIO(plain), 'infer_from_extension', 'file.TXT'),
+        (io.BytesIO(plain), '.unknown', ''),
+        (io.BytesIO(gzip.compress(plain)), 'infer_from_extension', 'file.gz'),
+        (io.BytesIO(gzip.compress(plain)), 'infer_from_extension', 'file.GZ'),
+        (label(io.BytesIO(gzip.compress(plain)), 'file.gz'), 'infer_from_extension', ''),
+        (io.BytesIO(gzip.compress(plain)), '.gz', 'file.gz'),
+    ]
+)
+def test_compression_wrapper_read(fileobj, compression, filename):
+    wrapped = smart_open.compression.compression_wrapper(fileobj, 'rb', compression, filename)
+    assert wrapped.read() == plain
diff --git a/smart_open/tests/test_s3.py b/smart_open/tests/test_s3.py
index 4e5ca5b..a91a731 100644
--- a/smart_open/tests/test_s3.py
+++ b/smart_open/tests/test_s3.py
@@ -681,6 +681,38 @@ class IterBucketTest(unittest.TestCase):
         results = list(smart_open.s3.iter_bucket(BUCKET_NAME))
         self.assertEqual(len(results), 10)
 
+    @pytest.mark.skipif(condition=sys.platform == 'win32', reason="does not run on windows")
+    @pytest.mark.xfail(
+        condition=sys.platform == 'darwin',
+        reason="MacOS uses spawn rather than fork for multiprocessing",
+    )
+    def test_iter_bucket_404(self):
+        populate_bucket()
+
+        def throw_404_error_for_key_4(*args):
+            if args[1] == "key_4":
+                raise botocore.exceptions.ClientError(
+                    error_response={"Error": {"Code": "404", "Message": "Not Found"}},
+                    operation_name="HeadObject",
+                )
+            else:
+                return [0]
+
+        with mock.patch("smart_open.s3._download_fileobj", side_effect=throw_404_error_for_key_4):
+            results = list(smart_open.s3.iter_bucket(BUCKET_NAME))
+            self.assertEqual(len(results), 9)
+
+    @pytest.mark.skipif(condition=sys.platform == 'win32', reason="does not run on windows")
+    @pytest.mark.xfail(
+        condition=sys.platform == 'darwin',
+        reason="MacOS uses spawn rather than fork for multiprocessing",
+    )
+    def test_iter_bucket_non_404(self):
+        populate_bucket()
+        with mock.patch("smart_open.s3._download_fileobj", side_effect=ARBITRARY_CLIENT_ERROR):
+            with pytest.raises(botocore.exceptions.ClientError):
+                list(smart_open.s3.iter_bucket(BUCKET_NAME))
+
     def test_deprecated_top_level_s3_iter_bucket(self):
         populate_bucket()
         with self.assertLogs(smart_open.logger.name, level='WARN') as cm:
diff --git a/smart_open/tests/test_ssh.py b/smart_open/tests/test_ssh.py
index 18a8f16..a4b91eb 100644
--- a/smart_open/tests/test_ssh.py
+++ b/smart_open/tests/test_ssh.py
@@ -4,6 +4,8 @@ import logging
 import unittest
 from unittest import mock
 
+from paramiko import SSHException
+
 import smart_open.ssh
 
 
@@ -12,8 +14,8 @@ def mock_ssh(func):
         smart_open.ssh._SSH.clear()
         return func(*args, **kwargs)
 
-    return mock.patch("paramiko.client.SSHClient.get_transport")(
-        mock.patch("paramiko.client.SSHClient.connect")(wrapper)
+    return mock.patch("paramiko.SSHClient.get_transport")(
+        mock.patch("paramiko.SSHClient.connect")(wrapper)
     )
 
 
@@ -49,6 +51,23 @@ class SSHOpen(unittest.TestCase):
         )
         mock_connect.assert_called_with("some-host", 22, username="user", key_filename="key")
 
+    @mock_ssh
+    def test_reconnect_after_session_timeout(self, mock_connect, get_transp_mock):
+        mock_sftp = get_transp_mock().open_sftp_client()
+        get_transp_mock().open_sftp_client.reset_mock()
+
+        def mocked_open_sftp():
+            if len(mock_connect.call_args_list) < 2:  # simulate timeout until second connect()
+                yield SSHException('SSH session not active')
+            while True:
+                yield mock_sftp
+
+        get_transp_mock().open_sftp_client.side_effect = mocked_open_sftp()
+
+        smart_open.open("ssh://user:pass@some-host/")
+        mock_connect.assert_called_with("some-host", 22, username="user", password="pass")
+        mock_sftp.open.assert_called_once()
+
 
 if __name__ == "__main__":
     logging.basicConfig(format="%(asctime)s : %(levelname)s : %(message)s", level=logging.DEBUG)