Repository: barseghyanartur/faker-file (language: python)

Issue:
Dev



Discussion / hints:

Fix patch:
diff --git a/.secrets.baseline b/.secrets.baseline
index ce2c60b4..ad31e62a 100644
--- a/.secrets.baseline
+++ b/.secrets.baseline
@@ -148,6 +148,15 @@
         "line_number": 34
       }
     ],
+    "src/faker_file/tests/test_sftp_server.py": [
+      {
+        "type": "Secret Keyword",
+        "filename": "src/faker_file/tests/test_sftp_server.py",
+        "hashed_secret": "e8662cfb96bd9c7fe84c31d76819ec3a92c80e63",
+        "is_verified": true,
+        "line_number": 77
+      }
+    ],
     "src/faker_file/tests/test_storages.py": [
       {
         "type": "Secret Keyword",
@@ -158,5 +167,5 @@
       }
     ]
   },
-  "generated_at": "2023-06-13T23:24:27Z"
+  "generated_at": "2023-06-14T22:21:03Z"
 }
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index ba4055b2..8a8b2013 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -15,6 +15,12 @@ are used for versioning (schema follows below):
   0.3.4 to 0.4).
 - All backwards incompatible changes are mentioned in this document.
 
+0.15.4
+------
+2023-06-15
+
+- Improved ``SFTPStorage`` tests.
+
 0.15.3
 ------
 2023-06-14
diff --git a/setup.py b/setup.py
index d725aad6..58d636b1 100644
--- a/setup.py
+++ b/setup.py
@@ -2,7 +2,7 @@
 
 from setuptools import find_packages, setup
 
-version = "0.15.3"
+version = "0.15.4"
 
 try:
     readme = open(os.path.join(os.path.dirname(__file__), "README.rst")).read()
diff --git a/src/faker_file/__init__.py b/src/faker_file/__init__.py
index 56276fde..af4dff45 100644
--- a/src/faker_file/__init__.py
+++ b/src/faker_file/__init__.py
@@ -1,5 +1,5 @@
 __title__ = "faker_file"
-__version__ = "0.15.3"
+__version__ = "0.15.4"
 __author__ = "Artur Barseghyan <artur.barseghyan@gmail.com>"
 __copyright__ = "2022-2023 Artur Barseghyan"
 __license__ = "MIT"
diff --git a/tox.ini b/tox.ini
index 57881a91..ba9ea8bb 100644
--- a/tox.ini
+++ b/tox.ini
@@ -20,7 +20,7 @@ commands =
     mkdir -p examples/django_example/project/media/
     mkdir -p examples/django_example/project/static/
     pip install -e .[all]
-    {envpython} -m pytest -vrx -m "not optional" --ignore src/faker_file/tests/test_sqlalchemy_integration.py --ignore src/faker_file/tests/test_augment_file_from_dir_provider.py --ignore src/faker_file/tests/test_sftp_storage.py
+    {envpython} -m pytest -vrx -m "not optional" --ignore src/faker_file/tests/test_sqlalchemy_integration.py --ignore src/faker_file/tests/test_augment_file_from_dir_provider.py
 
 [gh-actions]
 python =


Test patch:
diff --git a/src/faker_file/tests/sftp_server.py b/src/faker_file/tests/sftp_server.py
new file mode 100644
index 00000000..83e49201
--- /dev/null
+++ b/src/faker_file/tests/sftp_server.py
@@ -0,0 +1,155 @@
+import asyncio
+import logging
+import os
+import sys
+import tempfile
+import threading
+from asyncio import Semaphore
+
+import asyncssh
+
+DIR_PATH = os.environ.get("DIR_PATH", tempfile.gettempdir())
+SFTP_USER = os.environ.get("SFTP_USER", "foo")
+SFTP_PASS = os.environ.get("SFTP_PASS", "pass")
+SFTP_HOST = os.environ.get("SFTP_HOST", "0.0.0.0")
+SFTP_PORT = int(os.environ.get("SFTP_PORT", 2222))
+NUM_CONCURRENT_CONNECTIONS = int(
+    os.environ.get("NUM_CONCURRENT_CONNECTIONS", 50)
+)
+LOGGER = logging.getLogger(__name__)
+
+__all__ = (
+    "SFTPServer",
+    "SFTPServerManager",
+    "SSHServer",
+    "start_server",
+    "start_server_async",
+)
+
+
+class SFTPServer(asyncssh.SFTPServer):
+    def __init__(self: "SFTPServer", conn: asyncssh.SSHServerChannel) -> None:
+        root = DIR_PATH
+        super().__init__(conn, chroot=root)
+
+
+class SSHServer(asyncssh.SSHServer):
+    def __init__(self: "SSHServer", connection_semaphore: Semaphore) -> None:
+        self._connection_semaphore = connection_semaphore
+
+    def password_auth_supported(self: "SSHServer") -> bool:
+        return True
+
+    def validate_password(
+        self: "SSHServer", username: str, password: str
+    ) -> bool:
+        user_passwords = {SFTP_USER: SFTP_PASS}
+        return user_passwords.get(username) == password
+
+    def session_requested(self: "SSHServer") -> bool:
+        return True
+
+    def sftp_requested(self: "SSHServer") -> type[SFTPServer]:
+        return SFTPServer
+
+    async def begin_auth(self: "SSHServer", username: str) -> bool:
+        await self._connection_semaphore.acquire()
+        return True
+
+    def auth_completed(self: "SSHServer") -> None:
+        self._connection_semaphore.release()
+
+
+async def start_server_async(
+    host: str = SFTP_HOST, port: int = SFTP_PORT
+) -> None:
+    # Generate an SSH keypair or use an existing one
+    server_key = asyncssh.generate_private_key("ssh-rsa")
+
+    # Create a connection semaphore with the desired maximum number of
+    # connections.
+    connection_semaphore = Semaphore(50)
+
+    LOGGER.info(f"Starting SFTP server at {host}:{port}")
+    print(f"start_server_async: Starting SFTP server at {host}:{port}")
+
+    server = await asyncssh.listen(
+        host,
+        port,
+        server_host_keys=[server_key],
+        server_factory=lambda: SSHServer(connection_semaphore),
+        sftp_factory=SFTPServer,
+    )
+
+    async with server:
+        try:
+            await server.wait_closed()
+        except asyncio.CancelledError:
+            pass
+
+
+def start_server(host: str = SFTP_HOST, port: int = SFTP_PORT) -> None:
+    print(f"start_server: Starting SFTP server at {host}:{port}")
+
+    # This function will be run in a new thread
+    def run_loop_in_thread(loop):
+        asyncio.set_event_loop(loop)
+        loop.run_forever()
+
+    # Get the current event loop, create if it doesn't exist
+    loop = asyncio.new_event_loop()
+
+    # Schedule the coroutine to be executed
+    loop.create_task(start_server_async(host=host, port=port))
+
+    # Start a new thread running the loop
+    server_thread = threading.Thread(target=run_loop_in_thread, args=(loop,))
+    server_thread.daemon = True
+    server_thread.start()
+
+
+class SFTPServerManager:
+    def __init__(self):
+        self.loop = asyncio.get_event_loop()
+        self.stop_event = asyncio.Event()
+
+    async def start_server(self, host: str = SFTP_HOST, port: int = SFTP_PORT):
+        # Generate an SSH keypair or use an existing one
+        server_key = asyncssh.generate_private_key("ssh-rsa")
+
+        # Create a connection semaphore with the desired maximum number of
+        # connections.
+        connection_semaphore = Semaphore(50)
+
+        server = await asyncssh.listen(
+            host,
+            port,
+            server_host_keys=[server_key],
+            server_factory=lambda: SSHServer(connection_semaphore),
+            sftp_factory=SFTPServer,
+        )
+        # Just replace stop_event with self.stop_event
+
+        async with server:
+            try:
+                await self.stop_event.wait()
+            except asyncio.CancelledError:
+                pass
+            finally:
+                server.close()
+                await server.wait_closed()
+
+    def start(self):
+        self.loop.run_until_complete(self.start_server())
+
+    def stop(self):
+        self.loop.call_soon_threadsafe(self.stop_event.set)
+
+
+if __name__ == "__main__":
+    try:
+        print(f"Starting SFTP server on port {SFTP_PORT}")
+        asyncssh.logging.set_debug_level(3)  # Set verbosity to max
+        asyncio.run(start_server_async())
+    except (OSError, asyncssh.Error) as exc:
+        sys.exit("Error starting SFTP server: " + str(exc))
diff --git a/src/faker_file/tests/test_sftp_server.py b/src/faker_file/tests/test_sftp_server.py
new file mode 100644
index 00000000..818f6e3b
--- /dev/null
+++ b/src/faker_file/tests/test_sftp_server.py
@@ -0,0 +1,276 @@
+import asyncio
+import os
+import socket
+import threading
+import time
+from typing import Any, Callable
+from unittest import IsolatedAsyncioTestCase
+
+import asyncssh
+from faker import Faker
+
+from faker_file.providers.txt_file import TxtFileProvider
+
+from .sftp_server import SFTPServerManager, start_server, start_server_async
+
+__author__ = "Artur Barseghyan <artur.barseghyan@gmail.com>"
+__copyright__ = "2022-2023 Artur Barseghyan"
+__license__ = "MIT"
+__all__ = (
+    "TestSFTPServerWithManager",
+    "TestSFTPServerWithStartServer",
+    "TestSFTPServerWithStartServerAsync",
+)
+
+SFTP_USER = os.environ.get("SFTP_USER", "foo")
+SFTP_PASS = os.environ.get("SFTP_PASS", "pass")
+SFTP_HOST = os.environ.get("SFTP_HOST", "0.0.0.0")
+SFTP_PORT = int(os.environ.get("SFTP_PORT", 2222))
+SFTP_ROOT_PATH = os.environ.get("SFTP_ROOT_PATH", "/upload")
+
+FAKER = Faker()
+FAKER.add_provider(TxtFileProvider)
+
+
+class __TestSFTPServerMixin:
+    """Test SFTP server mix-in."""
+
+    assertIsInstance: Callable
+    assertRaises: Callable
+    assertEqual: Callable
+    assertTrue: Callable
+    assertFalse: Callable
+    sftp_host: str
+    sftp_port: int
+    sftp_user: str
+    sftp_pass: str
+
+    @staticmethod
+    def is_port_in_use(port: Any) -> bool:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            return s.connect_ex(("0.0.0.0", port)) == 0
+
+    @classmethod
+    def free_port(cls: "__TestSFTPServerMixin") -> None:
+        # Check if the port is in use and wait until it is free
+        while cls.is_port_in_use(cls.sftp_port):
+            print(f"Port {cls.sftp_port} in use, waiting...")
+            time.sleep(1)
+
+    async def test_successful_connection(self: "__TestSFTPServerMixin") -> None:
+        async with asyncssh.connect(
+            self.sftp_host,
+            port=self.sftp_port,
+            username=self.sftp_user,
+            password=self.sftp_pass,
+            known_hosts=None,
+        ) as conn:
+            async with conn.start_sftp_client() as sftp:
+                self.assertIsInstance(sftp, asyncssh.SFTPClient)
+
+    async def test_failed_connection(self: "__TestSFTPServerMixin") -> None:
+        with self.assertRaises(asyncssh.PermissionDenied):
+            async with asyncssh.connect(
+                self.sftp_host,
+                port=self.sftp_port,
+                username=self.sftp_user,
+                password="wrong_password",
+                known_hosts=None,
+            ):
+                pass
+
+    async def test_file_upload(self: "__TestSFTPServerMixin") -> None:
+        async with asyncssh.connect(
+            self.sftp_host,
+            port=self.sftp_port,
+            username=self.sftp_user,
+            password=self.sftp_pass,
+            known_hosts=None,
+        ) as conn:
+            async with conn.start_sftp_client() as sftp:
+                test_file = FAKER.txt_file()
+                await sftp.put(
+                    test_file.data["filename"], "/testfile_upload.txt"
+                )
+
+                # Read back the file and check its contents
+                async with sftp.open(
+                    "/testfile_upload.txt", "r"
+                ) as uploaded_file:
+                    uploaded_contents = await uploaded_file.read()
+
+                self.assertEqual(test_file.data["content"], uploaded_contents)
+
+    async def test_file_delete(self: "__TestSFTPServerMixin") -> None:
+        async with asyncssh.connect(
+            self.sftp_host,
+            port=self.sftp_port,
+            username=self.sftp_user,
+            password=self.sftp_pass,
+            known_hosts=None,
+        ) as conn:
+            async with conn.start_sftp_client() as sftp:
+                test_file = FAKER.txt_file()
+                await sftp.put(
+                    test_file.data["filename"], "/testfile_delete.txt"
+                )
+
+                # Ensure the file exists
+                self.assertTrue(await sftp.exists("/testfile_delete.txt"))
+
+                # Delete the file and ensure it's gone
+                await sftp.remove("/testfile_delete.txt")
+                self.assertFalse(await sftp.exists("/testfile_delete.txt"))
+
+
+class TestSFTPServerWithStartServerAsync(
+    IsolatedAsyncioTestCase,
+    __TestSFTPServerMixin,
+):
+    sftp_host = SFTP_HOST
+    sftp_port = 2223
+    sftp_user = SFTP_USER
+    sftp_pass = SFTP_PASS
+
+    @classmethod
+    def setUpClass(cls):
+        # Free port
+        cls.free_port()
+
+        # This function will be run in a new thread
+        def run_loop_in_thread(_loop):
+            asyncio.set_event_loop(_loop)
+            _loop.run_forever()
+
+        # Get the current event loop, create if it doesn't exist
+        loop = asyncio.new_event_loop()
+
+        # Schedule the coroutine to be executed
+        loop.create_task(
+            start_server_async(host=cls.sftp_host, port=cls.sftp_port)
+        )
+
+        # Start a new thread running the loop
+        server_thread = threading.Thread(
+            target=run_loop_in_thread, args=(loop,)
+        )
+        server_thread.daemon = True
+        server_thread.start()
+
+        # Allow some time for the server to start
+        time.sleep(2)
+
+    @classmethod
+    def tearDownClass(cls):
+        # Since the server thread is a daemon, it will be stopped when the
+        # main thread exits.
+        pass
+
+
+class TestSFTPServerWithStartServer(
+    IsolatedAsyncioTestCase,
+    __TestSFTPServerMixin,
+):
+    sftp_host = SFTP_HOST
+    sftp_port = 2224
+    sftp_user = SFTP_USER
+    sftp_pass = SFTP_PASS
+
+    @classmethod
+    def setUpClass(cls):
+        # Free port
+        cls.free_port()
+
+        # Note: start_server is not async, and it creates its own thread
+        start_server(host=cls.sftp_host, port=cls.sftp_port)
+
+        # Give server some time to start
+        loop = asyncio.new_event_loop()
+        loop.run_until_complete(asyncio.sleep(2))
+
+    @classmethod
+    def tearDownClass(cls):
+        # Since the server is running in a daemonized thread,
+        # it will be terminated when the main process finishes.
+        # No explicit tear down is required for the server in this test case.
+        pass
+
+
+class TestSFTPServerWithManager(
+    IsolatedAsyncioTestCase,
+    __TestSFTPServerMixin,
+):
+    # manager: SFTPServerManager
+    # manager_thread: threading.Thread
+    sftp_host = SFTP_HOST
+    sftp_port = SFTP_PORT
+    sftp_user = SFTP_USER
+    sftp_pass = SFTP_PASS
+
+    @classmethod
+    def setUpClass(cls):
+        # Create and set an event loop
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        # Free port
+        cls.free_port()
+
+        cls.manager = SFTPServerManager()
+        # Starting the manager in a separate thread since it uses
+        # `run_until_complete`.
+        cls.manager_thread = threading.Thread(target=cls.manager.start)
+        cls.manager_thread.daemon = True
+        cls.manager_thread.start()
+
+        # Allow some time for the server to start and check if it's ready
+        max_retries = 10
+        retries = 0
+        while retries < max_retries:
+            try:
+                # Try to establish a connection to the server
+                with socket.create_connection(
+                    (cls.sftp_host, cls.sftp_port), timeout=5
+                ):
+                    print(f"Server started on port {cls.sftp_port}")
+                    break
+            except (ConnectionRefusedError, socket.timeout):
+                print("Waiting for server to start...")
+                retries += 1
+                time.sleep(1)
+        else:
+            raise RuntimeError("Server did not start")
+
+    @classmethod
+    def tearDownClass(cls):
+        # Stop the server
+        cls.manager.stop()
+        cls.manager_thread.join()
+
+
+# **************************************
+
+# class TestSFTPServerWithManager(
+#     IsolatedAsyncioTestCase,
+#     __TestSFTPServerMixin,
+# ):
+#     manager: SFTPServerManager
+#     manager_thread: threading.Thread
+#     sftp_host = SFTP_HOST
+#     sftp_port = SFTP_PORT
+#     sftp_user = SFTP_USER
+#     sftp_pass = SFTP_PASS
+#
+#     async def asyncSetUpClass(self):
+#         self.manager = SFTPServerManager()
+#         # Starting the manager in a separate thread since it uses
+#         # `run_until_complete`
+#         self.manager_thread = threading.Thread(target=self.manager.start)
+#         self.manager_thread.daemon = True
+#         self.manager_thread.start()
+#         time.sleep(2)  # Allow some time for the server to start
+#
+#     async def asyncTearDown(self):
+#         # Stop the server
+#         self.manager.stop()
+#         self.manager_thread.join()
diff --git a/src/faker_file/tests/test_sftp_storage.py b/src/faker_file/tests/test_sftp_storage.py
index f78a5a2c..912edcd9 100644
--- a/src/faker_file/tests/test_sftp_storage.py
+++ b/src/faker_file/tests/test_sftp_storage.py
@@ -1,5 +1,10 @@
+import logging
 import os
+import tempfile
+import threading
+import time
 import unittest
+from functools import partial
 from typing import Any, Dict, Type, Union
 
 from faker import Faker
@@ -7,30 +12,74 @@
 
 from ..providers.txt_file import TxtFileProvider
 from ..storages.sftp_storage import SFTPStorage
+from .sftp_server import SFTPServerManager, start_server
 
 __author__ = "Artur Barseghyan <artur.barseghyan@gmail.com>"
 __copyright__ = "2022-2023 Artur Barseghyan"
 __license__ = "MIT"
 __all__ = ("TestSFTPStorageTestCase",)
 
-FAKER = Faker()
-FAKER.add_provider(TxtFileProvider)
 SFTP_USER = os.environ.get("SFTP_USER", "foo")
 SFTP_PASS = os.environ.get("SFTP_PASS", "pass")
-SFTP_HOST = "0.0.0.0"
-SFTP_PORT = 2222
-SFTP_ROOT_PATH = "/upload"
+SFTP_HOST = os.environ.get("SFTP_HOST", "0.0.0.0")
+SFTP_PORT = int(os.environ.get("SFTP_PORT", 2222))
+SFTP_ROOT_PATH = os.environ.get("SFTP_ROOT_PATH", "/upload")
 
+LOGGER = logging.getLogger(__name__)
 
-class TestSFTPStorageTestCase(unittest.TestCase):
-    """Test SFTP storage.
+FAKER = Faker()
+FAKER.add_provider(TxtFileProvider)
 
-    It's assumed that the server is running on port 2222.
 
-    Example:
+class TestSFTPStorageTestCase(unittest.TestCase):
+    """Test SFTP storage."""
+
+    server_manager: SFTPServerManager
+    server_thread: threading.Thread
+    sftp_host = SFTP_HOST
+    sftp_port = 2229
+    sftp_user = SFTP_USER
+    sftp_pass = SFTP_PASS
+    sftp_root_path = SFTP_ROOT_PATH
+
+    # @classmethod
+    # def setUpClass(cls):
+    #     os.makedirs(
+    #         os.path.join(tempfile.gettempdir(), "upload", "sub"),
+    #         exist_ok=True,
+    #     )
+    #     # Start the server in a separate thread
+    #     cls.server_manager = SFTPServerManager()
+    #     cls.server_thread = threading.Thread(target=cls.server_manager.start)
+    #     # Daemonize the thread, so it exits when the main thread exits
+    #     cls.server_thread.daemon = True
+    #     cls.server_thread.start()
+    #     time.sleep(2)
+    #
+    # @classmethod
+    # def tearDownClass(cls):
+    #     # Stop the server when tests are done
+    #     cls.server_manager.stop()
+    #     cls.server_thread.join()  # Wait for the server thread to finish
+
+    @classmethod
+    def setUpClass(cls):
+        os.makedirs(
+            os.path.join(tempfile.gettempdir(), "upload", "sub"),
+            exist_ok=True,
+        )
 
-        docker run -p 2222:22 -d atmoz/sftp foo:pass:::upload
-    """
+        # Create a partial function with custom arguments
+        start_server_partial = partial(
+            start_server, host=cls.sftp_host, port=cls.sftp_port
+        )
+
+        # Start the server in a separate thread
+        cls.server_thread = threading.Thread(target=start_server_partial)
+        # Daemonize the thread, so it exits when the main thread exits
+        cls.server_thread.daemon = True
+        cls.server_thread.start()
+        time.sleep(2)
 
     @parametrize(
         "storage_cls, kwargs, prefix, basename, extension",
@@ -39,11 +88,11 @@ class TestSFTPStorageTestCase(unittest.TestCase):
             (
                 SFTPStorage,
                 {
-                    "host": SFTP_HOST,
-                    "port": SFTP_PORT,
-                    "username": SFTP_USER,
-                    "password": SFTP_PASS,
-                    "root_path": SFTP_ROOT_PATH,
+                    "host": sftp_host,
+                    "port": sftp_port,
+                    "username": sftp_user,
+                    "password": sftp_pass,
+                    "root_path": sftp_root_path,
                 },
                 "zzz",
                 None,
@@ -52,11 +101,11 @@ class TestSFTPStorageTestCase(unittest.TestCase):
             (
                 SFTPStorage,
                 {
-                    "host": SFTP_HOST,
-                    "port": SFTP_PORT,
-                    "username": SFTP_USER,
-                    "password": SFTP_PASS,
-                    "root_path": SFTP_ROOT_PATH,
+                    "host": sftp_host,
+                    "port": sftp_port,
+                    "username": sftp_user,
+                    "password": sftp_pass,
+                    "root_path": sftp_root_path,
                 },
                 None,
                 "my_zzz_filename",
@@ -103,11 +152,11 @@ def test_storage(
             (
                 SFTPStorage,
                 {
-                    "host": SFTP_HOST,
-                    "port": SFTP_PORT,
-                    "username": SFTP_USER,
-                    "password": SFTP_PASS,
-                    "root_path": SFTP_ROOT_PATH,
+                    "host": sftp_host,
+                    "port": sftp_port,
+                    "username": sftp_user,
+                    "password": sftp_pass,
+                    "root_path": sftp_root_path,
                 },
                 "zzz",
                 "",
@@ -138,42 +187,46 @@ def test_file_system_storage_abspath(
         """Test `FileSystemStorage` `abspath`."""
 
         storage = SFTPStorage(
-            host=SFTP_HOST,
-            port=SFTP_PORT,
-            username=SFTP_USER,
-            password=SFTP_PASS,
-            root_path=SFTP_ROOT_PATH,
+            host=self.sftp_host,
+            port=self.sftp_port,
+            username=self.sftp_user,
+            password=self.sftp_pass,
+            root_path=self.sftp_root_path,
         )
         filename = storage.generate_filename(prefix="", extension="tmp")
-        self.assertTrue(storage.abspath(filename).startswith(SFTP_ROOT_PATH))
+        self.assertTrue(
+            storage.abspath(filename).startswith(self.sftp_root_path)
+        )
 
     def test_integration(self: "TestSFTPStorageTestCase") -> None:
         storage = SFTPStorage(
-            host=SFTP_HOST,
-            port=SFTP_PORT,
-            username=SFTP_USER,
-            password=SFTP_PASS,
-            root_path=SFTP_ROOT_PATH,
+            host=self.sftp_host,
+            port=self.sftp_port,
+            username=self.sftp_user,
+            password=self.sftp_pass,
+            root_path=self.sftp_root_path,
         )
         file = FAKER.txt_file(storage=storage)
         self.assertTrue(storage.exists(file.data["filename"]))
         self.assertTrue(
-            storage.abspath(file.data["filename"]).startswith(SFTP_ROOT_PATH)
+            storage.abspath(file.data["filename"]).startswith(
+                self.sftp_root_path
+            )
         )
 
     def test_integration_sub_dir(self: "TestSFTPStorageTestCase") -> None:
         storage = SFTPStorage(
-            host=SFTP_HOST,
-            port=SFTP_PORT,
-            username=SFTP_USER,
-            password=SFTP_PASS,
-            root_path=SFTP_ROOT_PATH,
+            host=self.sftp_host,
+            port=self.sftp_port,
+            username=self.sftp_user,
+            password=self.sftp_pass,
+            root_path=self.sftp_root_path,
             rel_path="sub",
         )
         file = FAKER.txt_file(storage=storage)
         self.assertTrue(storage.exists(file.data["filename"]))
         self.assertTrue(
             storage.abspath(file.data["filename"]).startswith(
-                os.path.join(SFTP_ROOT_PATH, "sub")
+                os.path.join(self.sftp_root_path, "sub")
             )
         )