test(cli): fix environment variable isolation in CLI utils tests

Add clear=True to mock.patch.dict calls to properly isolate environment
variables in tests. Without this, tests fail when DATAHUB_* environment
variables are set in the user's shell.

Author: kyungsoo-datahub

metadata-ingestion/tests/unit/cli/test_cli_utils.py

@@ -15,13 +15,15 @@ def test_first_non_null():
     assert cli_utils.first_non_null([" ", "1", "2"]) == "1"
 
 
-@mock.patch.dict(os.environ, {"DATAHUB_GMS_HOST": "http://localhost:9092"})
+@mock.patch.dict(os.environ, {"DATAHUB_GMS_HOST": "http://localhost:9092"}, clear=True)
 def test_correct_url_when_gms_host_in_old_format():
     assert _get_config_from_env() == ("http://localhost:9092", None)
 
 
 @mock.patch.dict(
-    os.environ, {"DATAHUB_GMS_HOST": "localhost", "DATAHUB_GMS_PORT": "8080"}
+    os.environ,
+    {"DATAHUB_GMS_HOST": "localhost", "DATAHUB_GMS_PORT": "8080"},
+    clear=True,
 )
 def test_correct_url_when_gms_host_and_port_set():
     assert _get_config_from_env() == ("http://localhost:8080", None)
@@ -34,6 +36,7 @@ def test_correct_url_when_gms_host_and_port_set():
         "DATAHUB_GMS_HOST": "localhost",
         "DATAHUB_GMS_PORT": "8080",
     },
+    clear=True,
 )
 def test_correct_url_when_gms_host_port_url_set():
     assert _get_config_from_env() == ("http://localhost:8080", None)
@@ -47,6 +50,7 @@ def test_correct_url_when_gms_host_port_url_set():
         "DATAHUB_GMS_PORT": "8080",
         "DATAHUB_GMS_PROTOCOL": "https",
     },
+    clear=True,
 )
 def test_correct_url_when_gms_host_port_url_protocol_set():
     assert _get_config_from_env() == ("https://localhost:8080", None)
@@ -57,6 +61,7 @@ def test_correct_url_when_gms_host_port_url_protocol_set():
     {
         "DATAHUB_GMS_URL": "https://example.com",
     },
+    clear=True,
 )
 def test_correct_url_when_url_set():
     assert _get_config_from_env() == ("https://example.com", None)

metadata-ingestion/tests/unit/config/test_nested_secrets.py

@@ -251,5 +251,141 @@ class UnityCatalogConnectionConfig(ConfigModel):
         )
 
 
+class TestConfigLoggingMasking:
+    """Test that SecretStr fields are masked when configs are logged."""
+
+    def setup_method(self):
+        from datahub.masking.bootstrap import (
+            initialize_secret_masking,
+            shutdown_secret_masking,
+        )
+
+        shutdown_secret_masking()
+        SecretRegistry.reset_instance()
+        initialize_secret_masking(force=True)
+
+    def teardown_method(self):
+        from datahub.masking.bootstrap import shutdown_secret_masking
+
+        shutdown_secret_masking()
+        SecretRegistry.reset_instance()
+
+    def test_config_secrets_masked_in_logs(self):
+        """Test that SecretStr fields are masked when config is logged."""
+        import logging
+
+        from datahub.masking.masking_filter import SecretMaskingFilter
+
+        class TestSourceConfig(ConfigModel):
+            password: SecretStr = Field(description="Database password")
+            api_key: SecretStr = Field(description="API key")
+            host: str = Field(description="Database host")
+
+        # Create config with secrets
+        config = TestSourceConfig(
+            password="my-secret-password", api_key="my-api-key", host="my-host"
+        )
+
+        # Verify secrets are registered
+        registry = SecretRegistry.get_instance()
+        assert registry.has_secret("password")
+        assert registry.has_secret("api_key")
+
+        # Get the masking filter
+        masking_filter = SecretMaskingFilter(registry)
+
+        # Create log records with secrets and verify they're masked
+        test_cases = [
+            (f"Password is: {config.password.get_secret_value()}", "password"),
+            (f"API key is: {config.api_key.get_secret_value()}", "api_key"),
+            (f"Host is: {config.host}", None),  # Should not be masked
+        ]
+
+        for message, secret_name in test_cases:
+            record = logging.LogRecord(
+                name="test",
+                level=logging.INFO,
+                pathname="",
+                lineno=0,
+                msg=message,
+                args=(),
+                exc_info=None,
+            )
+
+            # Apply filter
+            masking_filter.filter(record)
+
+            # Check result
+            masked_message = record.getMessage()
+
+            if secret_name:
+                # Secret should be masked
+                assert "my-secret-password" not in masked_message, (
+                    f"Password not masked: {masked_message}"
+                )
+                assert "my-api-key" not in masked_message, (
+                    f"API key not masked: {masked_message}"
+                )
+                assert f"***REDACTED:{secret_name}***" in masked_message, (
+                    f"Redaction marker not found: {masked_message}"
+                )
+            else:
+                # Host should not be masked
+                assert "my-host" in masked_message
+
+    def test_config_secrets_masked_at_all_log_levels(self):
+        """Test that SecretStr fields are masked at all log levels (DEBUG, INFO, WARNING, ERROR)."""
+        import logging
+
+        from datahub.masking.masking_filter import SecretMaskingFilter
+
+        class TestSourceConfig(ConfigModel):
+            password: SecretStr = Field(description="Database password")
+
+        config = TestSourceConfig(password="debug-secret-password")
+
+        # Verify secret is registered
+        registry = SecretRegistry.get_instance()
+        assert registry.has_secret("password")
+
+        # Get the masking filter
+        masking_filter = SecretMaskingFilter(registry)
+
+        # Test all log levels
+        log_levels = [
+            (logging.DEBUG, "DEBUG"),
+            (logging.INFO, "INFO"),
+            (logging.WARNING, "WARNING"),
+            (logging.ERROR, "ERROR"),
+            (logging.CRITICAL, "CRITICAL"),
+        ]
+
+        for level, level_name in log_levels:
+            record = logging.LogRecord(
+                name="test",
+                level=level,
+                pathname="",
+                lineno=0,
+                msg=f"[{level_name}] Password: {config.password.get_secret_value()}",
+                args=(),
+                exc_info=None,
+            )
+
+            # Apply filter
+            masking_filter.filter(record)
+
+            # Check result
+            masked_message = record.getMessage()
+
+            # Secret should be masked at ALL log levels
+            assert "debug-secret-password" not in masked_message, (
+                f"Password not masked at {level_name} level: {masked_message}"
+            )
+            assert "***REDACTED:password***" in masked_message, (
+                f"Redaction marker not found at {level_name} level: {masked_message}"
+            )
+            assert level_name in masked_message  # Level name should still be there
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])