fixtures: remove s3_fake_creds_file, allow for finer boto3 client con…

…figuration

README.rst

@@ -115,7 +115,7 @@ The `tmp_upath` fixture can be used for parametrizing paths with pytest's indire
 
 In order to use real remotes instead of mocked ones, use `tmp_upath_factory` with the following methods
 
-- ``tmp_upath_factory.s3(region_name, endpoint_url)``
+- ``tmp_upath_factory.s3(region_name, client_kwargs)`` where client_kwargs are passed to the underlying S3FileSystem/boto client
 - ``tmp_upath_factory.gcs(endpoint_url)``
 - ``tmp_upath_factory.azure(connection_string)``
 

src/pytest_servers/factory.py

@@ -31,14 +31,14 @@ class TempUPathFactory:
         ),
         "s3": MockRemote(
             "s3_server",
-            "_s3_endpoint_url",
+            "_s3_client_kwargs",
             requires_docker=False,
         ),
     }
 
     def __init__(
         self,
-        s3_endpoint_url: str | None = None,
+        s3_client_kwargs: dict[str, str] | None = None,
         azure_connection_string: str | None = None,
         gcs_endpoint_url: str | None = None,
     ) -> None:
@@ -47,7 +47,7 @@ def __init__(
         self._local_path_factory: pytest.TempPathFactory | None = None
         self._azure_connection_string = azure_connection_string
         self._gcs_endpoint_url = gcs_endpoint_url
-        self._s3_endpoint_url = s3_endpoint_url
+        self._s3_client_kwargs = s3_client_kwargs
 
     @classmethod
     def from_request(
@@ -70,16 +70,13 @@ def from_request(
 
     def _mock_remote_setup(self, fs: str) -> None:
         try:
-            (
-                mock_remote_fixture,
-                remote_config_name,
-                needs_docker,
-            ) = self.mock_remotes[fs]
+            fixture, config_attr, needs_docker = self.mock_remotes[fs]
         except KeyError:
             msg = f"No mock remote available for fs: {fs}"
             raise RemoteUnavailable(msg) from None
 
-        if getattr(self, remote_config_name):  # remote is already configured
+        if getattr(self, config_attr):
+            # remote is already configured
             return
 
         if needs_docker and os.environ.get("CI") and sys.platform == "win32":
@@ -90,15 +87,15 @@ def _mock_remote_setup(self, fs: str) -> None:
 
         assert self._request
         try:
-            remote_config = self._request.getfixturevalue(mock_remote_fixture)
+            remote_config = self._request.getfixturevalue(fixture)
         except Exception as exc:  # noqa: BLE001
-            msg = f'{fs}: Failed to setup "{mock_remote_fixture}": {exc}'
+            msg = f'{fs}: Failed to setup "{fixture}": {exc}'
             if self._request.config.option.verbose >= 1:
                 raise RemoteUnavailable(msg) from exc
 
             raise RemoteUnavailable(msg) from None
 
-        setattr(self, remote_config_name, remote_config)
+        setattr(self, config_attr, remote_config)
 
     def mktemp(  # noqa: C901 # complex-structure
         self,
@@ -140,7 +137,7 @@ def mktemp(  # noqa: C901 # complex-structure
 
         if fs == "s3":
             return self.s3(
-                endpoint_url=self._s3_endpoint_url,
+                client_kwargs=self._s3_client_kwargs,
                 version_aware=version_aware,
                 **kwargs,
             )
@@ -172,22 +169,19 @@ def local(self) -> LocalPath:
 
     def s3(
         self,
-        endpoint_url: str | None = None,
+        client_kwargs: dict[str, Any] | None = None,
         *,
         version_aware: bool = False,
         **kwargs,
     ) -> UPath:
-        """Create a new S3 bucket and returns an UPath instance  .
+        """Create a new S3 bucket and returns an UPath instance.
 
-        `endpoint_url` can be used to use custom servers (e.g. moto s3).
+        `client_kwargs` can be used to configure the underlying boto client
         """
-        client_kwargs: dict[str, Any] = {}
-        if endpoint_url:
-            client_kwargs["endpoint_url"] = endpoint_url
-
         bucket_name = f"pytest-servers-{random_string()}"
         path = UPath(
             f"s3://{bucket_name}",
+            endpoint_url=client_kwargs.get("endpoint_url") if client_kwargs else None,
             client_kwargs=client_kwargs,
             version_aware=version_aware,
             **kwargs,
@@ -196,8 +190,16 @@ def s3(
             from botocore.session import Session
 
             session = Session()
-            client = session.create_client("s3", endpoint_url=endpoint_url)
-            client.create_bucket(Bucket=bucket_name, ACL="public-read")
+            client = session.create_client("s3", **client_kwargs)
+            client.create_bucket(
+                Bucket=bucket_name,
+                ACL="public-read",
+                CreateBucketConfiguration={
+                    "LocationConstraint": client_kwargs.get("region_name"),
+                }
+                if client_kwargs
+                else None,
+            )
 
             client.put_bucket_versioning(
                 Bucket=bucket_name,

src/pytest_servers/fixtures.py

@@ -8,7 +8,6 @@
 from .gcs import fake_gcs_server  # noqa: F401
 from .s3 import (  # noqa: F401
     MockedS3Server,
-    s3_fake_creds_file,
     s3_server,
     s3_server_config,
 )

src/pytest_servers/s3.py

@@ -37,49 +37,29 @@ def __exit__(self, *exc_args):
         self._server.stop()
 
 
-@pytest.fixture(scope="session")
-def s3_fake_creds_file(monkeypatch_session: pytest.MonkeyPatch) -> None:  # type: ignore[misc]
-    # https://github.com/spulec/moto#other-caveats
-    import pathlib
-
-    aws_dir = pathlib.Path("~").expanduser() / ".aws"
-    aws_dir.mkdir(exist_ok=True)
-
-    aws_creds = aws_dir / "credentials"
-    initially_exists = aws_creds.exists()
-
-    if not initially_exists:
-        aws_creds.touch()
-
-    try:
-        with monkeypatch_session.context() as m:
-            try:
-                m.delenv("AWS_PROFILE")
-            except KeyError:
-                pass
-            m.setenv("AWS_ACCESS_KEY_ID", "pytest-servers")
-            m.setenv("AWS_SECRET_ACCESS_KEY", "pytest-servers")
-            m.setenv("AWS_SECURITY_TOKEN", "pytest-servers")
-            m.setenv("AWS_SESSION_TOKEN", "pytest-servers")
-            m.setenv("AWS_DEFAULT_REGION", "us-east-1")
-            yield
-    finally:
-        if aws_creds.exists() and not initially_exists:
-            aws_creds.unlink()
-
-
 @pytest.fixture(scope="session")
 def s3_server_config() -> dict:
-    """Override to change default config of the server."""
+    """Override to change default config of the moto server."""
     return {}
 
 
 @pytest.fixture(scope="session")
 def s3_server(  # type: ignore[misc]
+    monkeypatch_session: pytest.MonkeyPatch,
     s3_server_config: dict,
-    s3_fake_creds_file: None,  # noqa: ARG001
-) -> str:
-    """Spins up a moto s3 server. Returns the endpoint URL."""
+) -> dict[str, str | None]:
+    """Spins up a moto s3 server.
+
+    Returns a client_kwargs dict that can be used with a boto client.
+    """
     assert isinstance(s3_server_config, dict)
+    monkeypatch_session.setenv("MOTO_ALLOW_NONEXISTENT_REGION", "true")
+
     with MockedS3Server(**s3_server_config) as server:
-        yield server.endpoint_url
+        yield {
+            "endpoint_url": server.endpoint_url,
+            "aws_access_key_id": "pytest-servers",
+            "aws_secret_access_key": "pytest-servers",
+            "aws_session_token": "pytest-servers",
+            "region_name": "pytest-servers-region",
+        }

tests/test_fixtures.py

@@ -12,3 +12,11 @@ def test_s3_versioning(tmp_s3_path, versioning):
 
 def test_s3_versioning_disabled(tmp_s3_path):
     assert not tmp_s3_path.fs.version_aware
+
+
+def test_s3_server_default_config(s3_server):
+    assert "endpoint_url" in s3_server
+    assert s3_server["aws_access_key_id"] == "pytest-servers"
+    assert s3_server["aws_secret_access_key"] == "pytest-servers"
+    assert s3_server["aws_session_token"] == "pytest-servers"
+    assert s3_server["region_name"] == "pytest-servers-region"

tests/test_utils.py

@@ -1,19 +1,6 @@
-import os
-from pathlib import Path
-
 from pytest_servers.fixtures import _version_aware
 
 
-def test_s3_fake_creds_file(s3_fake_creds_file):
-    assert os.getenv("AWS_PROFILE") is None
-    assert os.getenv("AWS_ACCESS_KEY_ID") == "pytest-servers"
-    assert os.getenv("AWS_SECRET_ACCESS_KEY") == "pytest-servers"
-    assert os.getenv("AWS_SECURITY_TOKEN") == "pytest-servers"
-    assert os.getenv("AWS_SESSION_TOKEN") == "pytest-servers"
-    assert os.getenv("AWS_DEFAULT_REGION") == "us-east-1"
-    assert (Path("~").expanduser() / ".aws").exists()
-
-
 def test_version_aware(request):
     assert not _version_aware(request)
     request.getfixturevalue("versioning")