Move self-signed cert fixture to utils.py (#814)

Signed-off-by: Marcel Coetzee <[email protected]>

tests/load/filesystem/test_filesystem_common.py

@@ -13,6 +13,7 @@
 from tests.common.storages.utils import assert_sample_files
 from tests.load.utils import ALL_FILESYSTEM_DRIVERS, AWS_BUCKET
 from tests.utils import preserve_environ, autouse_test_storage
+from .utils import self_signed_cert
 from tests.common.configuration.utils import environment
 
 

tests/load/filesystem/utils.py

@@ -1,14 +1,25 @@
+import os
 import posixpath
-from typing import Iterator, List, Sequence, Tuple
+import tempfile
 from contextlib import contextmanager
+from typing import Iterator
+from typing import List, Sequence, Tuple
+
+import pytest
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
 
-from dlt.load import Load
 from dlt.common.configuration.container import Container
 from dlt.common.configuration.specs.config_section_context import ConfigSectionContext
-from dlt.common.destination.reference import Destination, LoadJob, TDestination
+from dlt.common.destination.reference import LoadJob, TDestination
+from dlt.common.pendulum import timedelta, __utcnow
 from dlt.destinations import filesystem
 from dlt.destinations.impl.filesystem.filesystem import FilesystemClient
 from dlt.destinations.job_impl import EmptyLoadJob
+from dlt.load import Load
 from tests.load.utils import prepare_load_package
 
 
@@ -28,7 +39,7 @@ def perform_load(
     load_id, schema = prepare_load_package(load.load_storage, cases, write_disposition)
     client: FilesystemClient = load.get_destination_client(schema)  # type: ignore[assignment]
 
-    # for the replace disposition in the loader we truncate the tables, so do this here
+    # for the replace disposition in the loader, we truncate the tables, so do this here
     truncate_tables = []
     if write_disposition == "replace":
         for item in cases:
@@ -55,3 +66,40 @@ def perform_load(
             client.drop_storage()
         except Exception:
             print(f"Failed to delete FILESYSTEM dataset: {client.dataset_path}")
+
+
+@pytest.fixture(scope="function", autouse=False)
+def self_signed_cert() -> Iterator[str]:
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
+
+    subject = issuer = x509.Name(
+        [
+            x509.NameAttribute(NameOID.COUNTRY_NAME, "DE"),
+            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Berlin"),
+            x509.NameAttribute(NameOID.LOCALITY_NAME, "Prenzlauer Berg"),
+            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "ScaleVector GmbH"),
+            x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
+        ]
+    )
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(__utcnow())
+        .not_valid_after(__utcnow() + timedelta(days=1))
+        .add_extension(
+            x509.SubjectAlternativeName([x509.DNSName("localhost")]),
+            critical=False,
+        )
+        .sign(key, hashes.SHA256(), default_backend())
+    )
+
+    with tempfile.NamedTemporaryFile(suffix=".crt", delete=False) as cert_file:
+        cert_file.write(cert.public_bytes(serialization.Encoding.PEM))
+        cert_path = cert_file.name
+
+    yield cert_path
+
+    os.remove(cert_path)