Speed up HDFS, S3 and WebDAV tests using pytest-xdist

.github/workflows/codeql-analysis.yml

@@ -43,7 +43,7 @@ jobs:
 
       - name: Install dependencies
         run: |
-          uv sync --extra files --extra spark --group dev
+          uv sync --extra files --group test-spark-3.5 --group dev
           uv pip install -U flake8-commas
 
           # Set the `CODEQL-PYTHON` environment variable to the Python executable

.github/workflows/test-hdfs.yml

@@ -80,7 +80,7 @@ jobs:
           mkdir reports/ || echo "Directory exists"
           source .env.local
           echo "127.0.0.1 hdfs" | sudo tee -a /etc/hosts
-          ./pytest_runner.sh -m hdfs
+          ./pytest_runner.sh -n logical -m hdfs
 
       - name: Dump HDFS logs on failure
         if: failure()

.github/workflows/test-hive-and-iceberg.yml

@@ -71,8 +71,7 @@ jobs:
         run: |
           mkdir reports/ || echo "Directory exists"
           source .env.local
-          ./pytest_runner.sh -m hive
-          ./pytest_runner.sh -m "iceberg and not s3 and not hdfs"
+          ./pytest_runner.sh -m "hive or (iceberg and local_fs)"
 
       - name: Upload coverage results
         uses: actions/upload-artifact@v6

.github/workflows/test-local-fs.yml

@@ -71,7 +71,7 @@ jobs:
         run: |
           mkdir reports/ || echo "Directory exists"
           source .env.local
-          ./pytest_runner.sh -m local_fs
+          ./pytest_runner.sh -m "local_fs and not iceberg"
 
       - name: Upload coverage results
         uses: actions/upload-artifact@v6

.github/workflows/test-postgres.yml

@@ -92,7 +92,7 @@ jobs:
           source .env.local
           export ONETL_PG_PORT=6432
           # Run only some basic tests
-          ./pytest_runner.sh -m postgres  -k "tests_core_integration or tests_db_connection_integration"
+          ./pytest_runner.sh -m postgres -k "tests_core_integration or tests_db_connection_integration"
 
       - name: Dump Postgres logs on failure
         if: failure()

.github/workflows/test-s3.yml

@@ -83,7 +83,7 @@ jobs:
         run: |
           mkdir reports/ || echo "Directory exists"
           source .env.local
-          ./pytest_runner.sh -m s3
+          ./pytest_runner.sh -n logical -m s3
 
       - name: Dump S3 logs on failure
         if: failure()

.github/workflows/test-webdav.yml

@@ -67,7 +67,7 @@ jobs:
         run: |
           mkdir reports/ || echo "Directory exists"
           source .env.local
-          ./pytest_runner.sh -m webdav
+          ./pytest_runner.sh -n 2 -m webdav
 
       - name: Dump WebDAV logs on failure
         if: failure()

.github/workflows/tests.yml

@@ -14,7 +14,7 @@ concurrency:
 
 env:
   DEFAULT_PYTHON: '3.13'
-  MIN_COVERAGE: 96
+  MIN_COVERAGE: 95
 
 permissions:
   contents: write

pyproject.toml

@@ -175,8 +175,10 @@ test = [
     "pytest-lazy-fixture",
     "pytest-mock",
     "pytest-rerunfailures",
+    "pytest-xdist[psutil]",
     "requests",
     "responses",
+    "filelock",
 ]
 dev = [
     "black",

tests/fixtures/connections/file_connections.py

@@ -1,4 +1,3 @@
-import secrets
 import shutil
 
 import pytest
@@ -51,10 +50,11 @@ def file_connection_with_path_and_files(request):
 
 
 @pytest.fixture()
-def file_connection_resource_path(resource_path, tmp_path_factory):
-    temp_dir = tmp_path_factory.mktemp("test_files") / secrets.token_hex(5)
+def file_connection_resource_path(resource_path, tmp_path_factory, worker_id):
+    temp_dir = tmp_path_factory.mktemp("test_files") / worker_id
     shutil.copytree(resource_path / "file_connection", temp_dir)
-    return temp_dir
+    yield temp_dir
+    shutil.rmtree(temp_dir, ignore_errors=True)
 
 
 @pytest.fixture()

tests/fixtures/connections/ftp.py

@@ -37,19 +37,16 @@ def ftp_file_connection(ftp_server):
 
 
 @pytest.fixture()
-def ftp_file_connection_with_path(request, ftp_file_connection):
+def ftp_file_connection_with_path(ftp_file_connection, worker_id):
     connection = ftp_file_connection
-    root = PurePosixPath("/data")
-
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
-
-    request.addfinalizer(finalizer)
+    root = PurePosixPath("/data", worker_id)
 
     connection.remove_dir(root, recursive=True)
     connection.create_dir(root)
 
-    return connection, root
+    yield connection, root
+
+    connection.remove_dir(root, recursive=True)
 
 
 @pytest.fixture()

tests/fixtures/connections/ftps.py

@@ -37,19 +37,16 @@ def ftps_file_connection(ftps_server):
 
 
 @pytest.fixture()
-def ftps_file_connection_with_path(request, ftps_file_connection):
+def ftps_file_connection_with_path(ftps_file_connection, worker_id):
     connection = ftps_file_connection
-    root = PurePosixPath("/data")
-
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
-
-    request.addfinalizer(finalizer)
+    root = PurePosixPath("/data", worker_id)
 
     connection.remove_dir(root, recursive=True)
     connection.create_dir(root)
 
-    return connection, root
+    yield connection, root
+
+    connection.remove_dir(root, recursive=True)
 
 
 @pytest.fixture()

tests/fixtures/connections/hdfs.py

@@ -35,19 +35,16 @@ def hdfs_file_connection(hdfs_server):
 
 
 @pytest.fixture()
-def hdfs_file_connection_with_path(request, hdfs_file_connection):
+def hdfs_file_connection_with_path(hdfs_file_connection, worker_id):
     connection = hdfs_file_connection
-    root = PurePosixPath("/data")
-
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
-
-    request.addfinalizer(finalizer)
+    root = PurePosixPath("/data", worker_id)
 
     connection.remove_dir(root, recursive=True)
     connection.create_dir(root)
 
-    return connection, root
+    yield connection, root
+
+    connection.remove_dir(root, recursive=True)
 
 
 @pytest.fixture()

tests/fixtures/connections/local_fs.py

@@ -1,5 +1,4 @@
 import os
-import secrets
 import shutil
 from pathlib import Path
 
@@ -21,26 +20,33 @@ def local_fs_file_df_connection(spark):
 
 
 @pytest.fixture()
-def local_fs_file_df_connection_with_path(local_fs_file_df_connection, tmp_path_factory):
+def local_fs_file_df_connection_with_path(local_fs_file_df_connection, tmp_path_factory, worker_id):
     connection = local_fs_file_df_connection
-    root = tmp_path_factory.mktemp("local_fs")
-    return connection, root
+    root = tmp_path_factory.mktemp("local-fs") / worker_id
+    # Iceberg warehouse dir should be created beforehand
+    root.mkdir(exist_ok=True)
+    yield connection, root
+    shutil.rmtree(root, ignore_errors=True)
 
 
 @pytest.fixture()
 def local_fs_file_df_connection_with_path_and_files(
     local_fs_file_df_connection,
     tmp_path_factory,
+    worker_id,
     resource_path,
 ):
     connection = local_fs_file_df_connection
-    root = tmp_path_factory.mktemp("local_fs") / secrets.token_hex(5)
+    root = tmp_path_factory.mktemp("local-fs") / worker_id
     copy_from = resource_path / "file_df_connection"
 
+    # there is no dirs_exist_ok in python 3.7, so we don't create root dir before copy
     shutil.copytree(copy_from, root)
 
     files = []
     for directory, _, content in os.walk(root):
         for file in content:
             files.append(Path(directory) / file)
-    return connection, root, files
+
+    yield connection, root, files
+    shutil.rmtree(root, ignore_errors=True)

tests/fixtures/connections/s3.py

@@ -1,5 +1,6 @@
 import os
 from collections import namedtuple
+from contextlib import suppress
 from pathlib import PurePosixPath
 
 import pytest
@@ -34,6 +35,8 @@ def s3_server():
     ],
 )
 def s3_file_connection(s3_server):
+    from minio.error import S3Error
+
     from onetl.connection import S3
 
     s3 = S3(
@@ -45,26 +48,23 @@ def s3_file_connection(s3_server):
         protocol=s3_server.protocol,
     )
 
-    if not s3.client.bucket_exists(bucket_name=s3_server.bucket):
+    with suppress(S3Error):
         s3.client.make_bucket(bucket_name=s3_server.bucket)
 
     return s3
 
 
 @pytest.fixture()
-def s3_file_connection_with_path(request, s3_file_connection):
+def s3_file_connection_with_path(s3_file_connection, worker_id):
     connection = s3_file_connection
-    root = PurePosixPath("/data")
+    root = PurePosixPath("/data", worker_id)
 
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
+    connection.remove_dir(root, recursive=True)
 
-    request.addfinalizer(finalizer)
+    yield connection, root
 
     connection.remove_dir(root, recursive=True)
 
-    return connection, root
-
 
 @pytest.fixture()
 def s3_file_connection_with_path_and_files(resource_path, s3_file_connection_with_path):

tests/fixtures/connections/samba.py

@@ -41,19 +41,16 @@ def samba_file_connection(samba_server):
 
 
 @pytest.fixture()
-def samba_file_connection_with_path(request, samba_file_connection):
+def samba_file_connection_with_path(samba_file_connection, worker_id):
     connection = samba_file_connection
-    root = PurePosixPath("/data")
+    root = PurePosixPath("/data", worker_id)
 
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
+    connection.remove_dir(root, recursive=True)
 
-    request.addfinalizer(finalizer)
+    yield connection, root
 
     connection.remove_dir(root, recursive=True)
 
-    return connection, root
-
 
 @pytest.fixture()
 def samba_file_connection_with_path_and_files(resource_path, samba_file_connection_with_path):

tests/fixtures/connections/sftp.py

@@ -37,19 +37,16 @@ def sftp_file_connection(sftp_server):
 
 
 @pytest.fixture()
-def sftp_file_connection_with_path(request, sftp_file_connection):
+def sftp_file_connection_with_path(sftp_file_connection, worker_id):
     connection = sftp_file_connection
-    root = PurePosixPath("/app/data")
-
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
-
-    request.addfinalizer(finalizer)
+    root = PurePosixPath("/app/data", worker_id)
 
     connection.remove_dir(root, recursive=True)
     connection.create_dir(root)
 
-    return connection, root
+    yield connection, root
+
+    connection.remove_dir(root, recursive=True)
 
 
 @pytest.fixture()

tests/fixtures/connections/webdav.py

@@ -44,19 +44,16 @@ def webdav_file_connection(webdav_server):
 
 
 @pytest.fixture()
-def webdav_file_connection_with_path(request, webdav_file_connection):
+def webdav_file_connection_with_path(webdav_file_connection, worker_id):
     connection = webdav_file_connection
-    root = PurePosixPath("/data")
-
-    def finalizer():
-        connection.remove_dir(root, recursive=True)
-
-    request.addfinalizer(finalizer)
+    root = PurePosixPath("/data", worker_id)
 
     connection.remove_dir(root, recursive=True)
     connection.create_dir(root)
 
-    return connection, root
+    yield connection, root
+
+    connection.remove_dir(root, recursive=True)
 
 
 @pytest.fixture()

tests/fixtures/processing/fixtures/kafka.py

@@ -4,15 +4,13 @@
 
 
 @pytest.fixture
-def kafka_topic(processing, request):
-    topic = secrets.token_hex(6)
+def kafka_topic(processing, worker_id):
+    topic = f"{worker_id}_topic_{secrets.token_hex(5)}"
     processing.create_topic(topic, num_partitions=1)
 
-    def delete_topic():
-        processing.delete_topic([topic])
+    yield topic
 
-    request.addfinalizer(delete_topic)
-    return topic
+    processing.delete_topic([topic])
 
 
 @pytest.fixture

tests/fixtures/processing/fixtures/processing.py

@@ -42,11 +42,11 @@ def processing(request, spark):
 
 
 @pytest.fixture
-def get_schema_table(processing):
+def get_schema_table(processing, worker_id):
     schema = processing.schema
     processing.create_schema(schema=schema)
 
-    table = f"test_{secrets.token_hex(5)}"
+    table = f"test_{worker_id}_{secrets.token_hex(3)}"
     full_name = f"{schema}.{table}"
 
     yield PreparedDbInfo(full_name=full_name, schema=schema, table=table)