test(filesystem): expand remote emulator integration coverage

- Add xdist-safe, controller-managed S3, Azure, and GCS emulators with pinned images.
- Cover remote formats, compression, globs, failures, native destination writes, cross-provider round trips, and S3 incremental loading.
- Refresh S3 listings between repeated loads and support Azure destination connection strings.
- Avoid emulator startup for unrelated targeted tests while keeping the fast lane Docker-free.
- Document Azure connection-string authentication and repeated S3 load behavior.

Author: hampsterx

docs/changelog.md

@@ -2,6 +2,12 @@
 
 ## in progress
 
+- Filesystem: Refreshed S3 directory listings between repeated loads so
+  file-level incremental runs detect newly uploaded objects.
+- Connectors: Added Azure Storage destination support for connection-string
+  authentication and custom endpoints.
+- Connectors: Reject mixed Azure Storage credential modes instead of silently
+  ignoring extra fields in source and destination URIs.
 - Filesystem: Fail ingestion when a concrete source path matches no file while
   keeping unmatched glob selections valid. Thanks, @hampsterx.
 - Filesystem: Added rsync source connector. Thanks, @oferchen.

docs/supported-sources/azure-storage.md

@@ -34,7 +34,8 @@ abfss://?account_name=<your_account_name>&account_key=<your_account_key>
 ## URI parameters
 
 :account_name:
-  Your Azure storage account name (required).
+  Your Azure storage account name (required for account-key, SAS-token, and
+  service-principal authentication). Do not supply it with `connection_string`.
 
 :account_key:
   Your storage account access key (account-key auth).
@@ -47,14 +48,23 @@ abfss://?account_name=<your_account_name>&account_key=<your_account_key>
 
 :account_host:
   Custom storage endpoint host (optional, for sovereign clouds or Azurite).
+  Do not supply it with `connection_string`, which carries its own endpoints.
+
+:connection_string:
+  A complete Azure Storage connection string, as an alternative to the
+  account-key, SAS-token, or service-principal fields. URL-encode the whole
+  value before placing it in the URI. Connection strings can include custom
+  service endpoints for local emulators such as Azurite.
 
 :layout:
   Layout template (optional, destination only).
 
-Supply **one** authentication mode: an `account_key`, a `sas_token`, or the
-full service-principal triplet (`tenant_id` + `client_id` + `client_secret`).
-Supplying both an account key/SAS and service-principal fields is rejected as
+Supply **one** authentication mode: a `connection_string`, an `account_key`, a
+`sas_token`, or the full service-principal triplet (`tenant_id` + `client_id` +
+`client_secret`). Supplying fields from multiple modes is rejected as
 ambiguous, and a partial service-principal triplet reports the missing field.
+When using `connection_string`, omit `account_name`, `account_host`, and all
+other credential fields.
 
 :::{warning}
 Account keys are base64 (containing `+`, `/`, `=`) and SAS tokens embed their
@@ -63,6 +73,17 @@ own `&` and `=` characters. You must URL-encode credential values in the URI
 Unencoded values are mangled when the query string is parsed.
 :::
 
+For example, write to an Azurite container by URL-encoding its connection
+string and passing it as the destination URI.
+
+```sh
+omniload ingest \
+    --source-uri 'csv:///local/users.csv' \
+    --source-table 'users' \
+    --dest-uri 'az://?connection_string=<url_encoded_connection_string>' \
+    --dest-table 'my-container/users'
+```
+
 :::{note}
 A Gen2 account differs from a plain Blob account only by having its
 hierarchical namespace enabled, so no separate configuration is required.
@@ -74,7 +95,7 @@ To integrate `omniload` with Azure Blob or Data Lake Storage, you need a
 storage account and one of the supported credentials. For guidance on
 obtaining an account key or SAS token, refer to the Microsoft documentation
 on [managing storage account access keys] and [shared access signatures].
-Service-principal credentials come from an [Azure AD app registration]. 
+Service-principal credentials come from an [Azure AD app registration].
 
 ## Authenticate with SAS token
 

src/dlt_filesystem/source/impl/remote.py

@@ -151,6 +151,9 @@ def dlt_source(self, uri: str, table: str, **kwargs):
         fs_kwargs: dict = {
             "key": access_key_id[0],
             "secret": secret_access_key[0],
+            # S3FileSystem caches directory listings by default. Disable the cache so
+            # long-lived processes see objects created between incremental runs.
+            "use_listings_cache": False,
         }
         if endpoint_url:
             fs_kwargs["endpoint_url"] = endpoint_url[0]
@@ -293,7 +296,6 @@ class SFTPSource(FilesystemSource):
     """Access files on SFTP servers."""
 
     def dlt_source(self, uri: str, table: str, **kwargs):
-
         if kwargs.get("incremental_key"):
             raise ValueError(
                 "SFTP takes care of incrementality on its own, you should not provide incremental_key"

src/dlt_filesystem/target/remote.py

@@ -59,9 +59,16 @@ def dlt_dest(self, uri: str, **kwargs):
         layout = params.get("layout", [None])[0]
         if layout is not None:
             opts["layout"] = layout
+        filesystem_kwargs = self.filesystem_kwargs(params)
+        if filesystem_kwargs:
+            opts["kwargs"] = filesystem_kwargs
 
         return BlobFS(**opts)  # type: ignore
 
+    def filesystem_kwargs(self, params: dict) -> dict:
+        """Return extra fsspec constructor arguments for the destination."""
+        return {}
+
     def validate_table(self, table: str):
         table = table.strip("/ ")
         if len(table.split("/")) < 2:
@@ -163,6 +170,13 @@ def credentials(self, params: dict) -> FileSystemCredentials:
         """
         auth = parse_azure_blob_auth(params)
 
+        if auth.connection_string is not None:
+            # dlt does not model Azure connection strings in its credential union.
+            # The actual credential is forwarded through filesystem_kwargs; a
+            # resolved empty object prevents dlt from probing for unrelated account
+            # fields before it constructs the adlfs filesystem.
+            return AzureCredentials().resolve()
+
         if auth.is_service_principal:
             # parse_azure_blob_auth guarantees the full triplet here; the
             # per-field `is not None` guards mirror the account-key branch and
@@ -191,6 +205,15 @@ def credentials(self, params: dict) -> FileSystemCredentials:
             key_credentials.azure_account_host = auth.account_host
         return key_credentials
 
+    def filesystem_kwargs(self, params: dict) -> dict:
+        """Forward connection-string credentials that dlt does not model."""
+        auth = parse_azure_blob_auth(params)
+        if auth.connection_string is None:
+            return {}
+        # adlfs does not forward api_version when constructing a client from a
+        # connection string, so do not pass that parsed option here.
+        return {"connection_string": auth.connection_string}
+
 
 class BlobFSClient(dlt.destinations.impl.filesystem.filesystem.FilesystemClient):
     @property

src/dlt_filesystem/util/auth.py

@@ -51,8 +51,9 @@ def parse_azure_blob_auth(params: dict) -> AzureBlobAuth:
     which ``parse_qs`` would otherwise mangle (``+`` becomes a space, an
     unencoded SAS token shatters into junk params).
 
-    Two auth modes are supported:
+    Three auth modes are supported:
 
+    * connection string: ``connection_string``
     * account-key / SAS: ``account_key`` or ``sas_token``
     * service principal: the full ``tenant_id`` + ``client_id`` +
       ``client_secret`` triplet
@@ -61,29 +62,43 @@ def parse_azure_blob_auth(params: dict) -> AzureBlobAuth:
         MissingValueError: if ``account_name`` is absent, if no auth material is
             supplied, or if the service-principal triplet is only partially
             supplied (naming the missing field(s)).
-        ValueError: if mutually exclusive credentials are supplied together
-            (``account_key`` with ``sas_token``, or account-key/SAS material
-            with service-principal material), rather than silently picking one.
+        ValueError: if mutually exclusive credentials are supplied together,
+            rather than silently picking one.
     """
 
     def one(key: str) -> Optional[str]:
         return params.get(key, [None])[0]
 
     connection_string = one("connection_string")
     api_version = one("api_version")
+    account_name = one("account_name")
+    account_key = one("account_key")
+    sas_token = one("sas_token")
+    sp_values = {field: one(field) for field in AZURE_SERVICE_PRINCIPAL_FIELDS}
+    account_host = one("account_host")
 
     if connection_string is not None:
+        conflicting_fields = [
+            field
+            for field, value in {
+                "account_name": account_name,
+                "account_key": account_key,
+                "sas_token": sas_token,
+                "account_host": account_host,
+                **sp_values,
+            }.items()
+            if value is not None
+        ]
+        if conflicting_fields:
+            raise ValueError(
+                "Conflicting Azure credentials: connection_string cannot be "
+                f"combined with {', '.join(conflicting_fields)}."
+            )
         return AzureBlobAuth(
             connection_string=connection_string,
             api_version=api_version,
         )
 
-    account_name = one("account_name")
-    account_key = one("account_key")
-    sas_token = one("sas_token")
-    sp_values = {field: one(field) for field in AZURE_SERVICE_PRINCIPAL_FIELDS}
-    account_host = one("account_host")
-
     if account_name is None:
         raise MissingConnectorOption("account_name", "Azure")
 

tests/conftest.py

@@ -12,11 +12,72 @@
 
 from omniload.target.clickhouse import ClickhouseDestination
 from tests.util.common import get_testdata_path
+from tests.warehouse.manager import get_remote_filesystem_services
 from tests.warehouse.settings import DESTINATIONS, SOURCES
 
 logger = logging.getLogger(__name__)
 
 
+def _explicit_test_targets(config) -> list[Path]:
+    """Return explicit repo test paths from pytest's original invocation."""
+    targets = []
+    for arg in config.invocation_params.args:
+        raw_path = arg.split("::", 1)[0]
+        if raw_path.startswith("-"):
+            continue
+        path = Path(raw_path)
+        if not path.is_absolute():
+            normalized = raw_path.removeprefix("./")
+            if normalized.split("/", 1)[0] not in {"tests", "examples", "omniload"}:
+                continue
+            path = Path(config.rootpath) / normalized
+        targets.append(path.resolve())
+    return targets
+
+
+def _only_main_filesystem_tests(config) -> bool:
+    """Return whether every explicit pytest target is in main/filesystem."""
+    filesystem_root = (Path(config.rootpath) / "tests/main/filesystem").resolve()
+    targets = _explicit_test_targets(config)
+    return bool(targets) and all(
+        target == filesystem_root or filesystem_root in target.parents
+        for target in targets
+    )
+
+
+def _remote_filesystem_tests_selected(config) -> bool:
+    """Return whether the invocation can collect the remote emulator test module."""
+    if hasattr(config, "workerinput"):
+        return config.workerinput["remote_filesystem_tests_selected"]
+    if _integration_deselected(config):
+        return False
+    remote_tests = (
+        Path(config.rootpath) / "tests/main/filesystem/test_remote_integration.py"
+    ).resolve()
+    targets = _explicit_test_targets(config)
+    if not targets:
+        return True
+    return any(
+        target == remote_tests or target in remote_tests.parents for target in targets
+    )
+
+
+def _all_managed_containers(config):
+    containers = set(SOURCES.values()) | set(DESTINATIONS.values())
+    if _remote_filesystem_tests_selected(config):
+        containers |= set(get_remote_filesystem_services().values())
+    return containers
+
+
+def _managed_containers(config):
+    containers = set()
+    if not _only_main_filesystem_tests(config):
+        containers |= set(SOURCES.values()) | set(DESTINATIONS.values())
+    if _remote_filesystem_tests_selected(config):
+        containers |= set(get_remote_filesystem_services().values())
+    return containers
+
+
 def pytest_configure(config):
     logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
     logging.getLogger("testcontainers.core.waiting_utils").setLevel(logging.WARNING)
@@ -28,6 +89,9 @@ def pytest_configure(config):
 def pytest_configure_node(node):
     """xdist hook"""
     node.workerinput["shared_directory"] = node.config.shared_directory
+    node.workerinput["remote_filesystem_tests_selected"] = (
+        _remote_filesystem_tests_selected(node.config)
+    )
 
 
 def pytest_sessionstart(session):
@@ -68,6 +132,18 @@ def _integration_deselected(config):
     return (config.option.markexpr or "").strip() == "not integration"
 
 
+def pytest_ignore_collect(collection_path: Path, config) -> bool | None:
+    """Avoid importing emulator-only tests in the Docker-free fast lane."""
+    if not _integration_deselected(config):
+        return None
+    remote_tests = (
+        Path(config.rootpath) / "tests/main/filesystem/test_remote_integration.py"
+    ).resolve()
+    if collection_path.resolve() == remote_tests:
+        return True
+    return None
+
+
 def _skip_containers(config):
     return _integration_deselected(config) or not _docker_available()
 
@@ -100,8 +176,10 @@ def testdata_path() -> Path:
 
 @pytest.fixture(scope="session", autouse=True)
 def manage_containers(request, shared_directory):
-    unique_containers = set(SOURCES.values()) | set(DESTINATIONS.values())
-    for container in unique_containers:
+    # Set every lock directory on both controller and workers. Container boot is
+    # still filtered below, but fixture polling must not depend on both processes
+    # deriving exactly the same target set from their invocation arguments.
+    for container in _all_managed_containers(request.config):
         container.lock_dir = shared_directory  # ty: ignore[invalid-assignment, unresolved-attribute, unused-ignore-comment]
 
 
@@ -111,8 +189,7 @@ def start_containers(config):
     if is_worker(config):
         return
 
-    unique_containers = set(SOURCES.values()) | set(DESTINATIONS.values())
-    unique_containers = [x for x in unique_containers if x is not None]
+    unique_containers = [x for x in _managed_containers(config) if x is not None]
 
     for container in unique_containers:
         container.lock_dir = config.shared_directory  # ty: ignore[invalid-assignment, unresolved-attribute, unused-ignore-comment]
@@ -164,8 +241,7 @@ def stop_containers(config):
     if not should_manage_containers:
         return
 
-    unique_containers = set(SOURCES.values()) | set(DESTINATIONS.values())
-    unique_containers = [x for x in unique_containers if x is not None]
+    unique_containers = [x for x in _managed_containers(config) if x is not None]
 
     for container in unique_containers:
         try:

tests/dlt_filesystem/gcs.py

@@ -31,9 +31,10 @@ class GCSFakeServerContainer(DockerContainer):
 
             >>> import google.cloud.storage
             >>> from google.auth.credentials import AnonymousCredentials
-            >>> from testcontainers.gcs import GCSFakeServerContainer
+            >>> from tests.dlt_filesystem.gcs import GCSFakeServerContainer
 
-            >>> with GCSFakeServerContainer() as container:
+            >>> image = "docker.io/fsouza/fake-gcs-server:1.55.1"
+            >>> with GCSFakeServerContainer(image) as container:
             ...   endpoint_url = container.get_endpoint_url()
             ...   client = google.cloud.storage.Client(
             ...     credentials=AnonymousCredentials(),
@@ -48,7 +49,7 @@ class GCSFakeServerContainer(DockerContainer):
 
     def __init__(
         self,
-        image: str = "docker.io/fsouza/fake-gcs-server:latest",
+        image: str,
         *,
         service_port: int = 4443,
         **kwargs,