Reset AWS settings (env var) to mock values for each amazon unit test

tests/providers/amazon/aws/hooks/test_base_aws.py

@@ -27,6 +27,7 @@
 import pytest
 from botocore.config import Config
 from botocore.credentials import ReadOnlyCredentials
+from botocore.exceptions import NoCredentialsError
 from moto import mock_dynamodb, mock_emr, mock_iam, mock_sts
 from moto.core import DEFAULT_ACCOUNT_ID
 
@@ -871,3 +872,19 @@ def test_raise_exception_when_no_retry_args(self):
         custom_fn = ThrowErrorUntilCount(count=2, quota_retry=None)
         with pytest.raises(Exception):
             _retryable_test(custom_fn)
+
+
+def test_raise_no_creds_default_credentials_strategy(tmp_path_factory, monkeypatch):
+    """Test raise an error if no credentials provided and default boto3 strategy unable to get creds."""
+    for env_key in ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "AWS_SECURITY_TOKEN"):
+        # Delete aws credentials environment variables
+        monkeypatch.delenv(env_key, raising=False)
+
+    hook = AwsBaseHook(aws_conn_id=None, client_type="sts")
+    with pytest.raises(NoCredentialsError):
+        # Call AWS STS API method GetCallerIdentity
+        # which should return result in case of valid credentials
+        result = hook.conn.get_caller_identity()
+        # In normal circumstances lines below should not execute.
+        # We want to show additional information why this test not passed
+        assert not result, f"Credentials Method: {hook.get_session().get_credentials().method}"

tests/providers/amazon/aws/hooks/test_s3.py

@@ -26,7 +26,7 @@
 
 import boto3
 import pytest
-from botocore.exceptions import ClientError, NoCredentialsError
+from botocore.exceptions import ClientError
 from moto import mock_s3
 
 from airflow.exceptions import AirflowException
@@ -48,21 +48,6 @@ def s3_bucket(mocked_s3_res):
     return bucket
 
 
-# This class needs to be separated out because if there are earlier mocks in the same class
-# the tests will fail on teardown.
-class TestAwsS3HookNoMock:
-    def test_check_for_bucket_raises_error_with_invalid_conn_id(self, monkeypatch):
-        monkeypatch.delenv("AWS_PROFILE", raising=False)
-        monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False)
-        monkeypatch.delenv("AWS_SECRET_ACCESS_KEY", raising=False)
-        hook = S3Hook(aws_conn_id="does_not_exist")
-        # We're mocking all actual AWS calls and don't need a connection. This
-        # avoids an Airflow warning about connection cannot be found.
-        hook.get_connection = lambda _: None
-        with pytest.raises(NoCredentialsError):
-            hook.check_for_bucket("test-non-existing-bucket")
-
-
 class TestAwsS3Hook:
     @mock_s3
     def test_get_conn(self):

tests/providers/amazon/conftest.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import os
 import warnings
 
 try:
@@ -59,3 +60,46 @@ def filter_botocore_warnings(botocore_version):
                 message="The .* client is currently using a deprecated endpoint.*",
             )
         yield
+
+
+@pytest.fixture(scope="package")
+def aws_testing_env_vars(tmp_path_factory):
+    """Session scoped fixture, return mock AWS specific environment variables for unit tests."""
+    tmp_dir = tmp_path_factory.mktemp("aws-configs-")
+
+    def empty_config(name: str) -> str:
+        config = tmp_dir / name
+        config.touch()
+        return str(config)
+
+    return {
+        # Mock values for access_key, secret_key and token
+        "AWS_ACCESS_KEY_ID": "airflow-testing",
+        "AWS_SECRET_ACCESS_KEY": "airflow-testing",
+        "AWS_SESSION_TOKEN": "airflow-testing",
+        "AWS_SECURITY_TOKEN": "airflow-testing",
+        # Set default region as N.Virginia (eu-west-1).
+        # Otherwise some unit tests might fail if this sets to other region.
+        "AWS_DEFAULT_REGION": "us-east-1",
+        "AWS_REGION": "us-east-1",
+        # Create empty configuration file
+        "AWS_SHARED_CREDENTIALS_FILE": empty_config("aws_shared_credentials_file"),
+        "AWS_CONFIG_FILE": empty_config("aws_config_file"),
+        "BOTO_CONFIG": empty_config("legacy_boto2_config.cfg"),
+    }
+
+
+@pytest.fixture(autouse=True)
+def set_default_aws_settings(aws_testing_env_vars, monkeypatch):
+    """
+    Change AWS configurations (env vars) before start each test.
+    1. Remove all existed variables which prefixed by AWS.
+        It might be some credentials, botocore configurations, etc.
+    2. Use pre-defined variables for unit testing.
+    """
+    for env_name in os.environ:
+        if env_name.startswith("AWS"):
+            monkeypatch.delenv(env_name, raising=False)
+    for env_name, value in aws_testing_env_vars.items():
+        monkeypatch.setenv(env_name, value)
+    yield