Always present MI token in production environment

Author: steventux

src/services/dicom/dicom_uploader.py

@@ -9,10 +9,11 @@
 import os
 from typing import Optional
 
-
 import requests
 from azure.identity import ManagedIdentityCredential
 
+from environment import Environment
+
 logger = logging.getLogger(__name__)
 
 
@@ -65,9 +66,9 @@ def headers(self) -> dict:
         }
 
     @property
-    def access_token(self) -> str:
-        resource = os.getenv("CLOUD_API_RESOURCE")
-        if resource:
+    def access_token(self) -> str | None:
+        resource = os.getenv("CLOUD_API_RESOURCE", "")
+        if resource or Environment().production:
             return ManagedIdentityCredential().get_token(resource).token
         else:
             return os.getenv("CLOUD_API_TOKEN", "")

tests/services/dicom/test_dicom_uploader.py

@@ -99,6 +99,17 @@ def test_upload_headers_without_managed_identity_resource(self, _, monkeypatch):
 
         assert DICOMUploader().headers == {"Authorization": "Bearer env_access_token"}
 
+    def test_upload_headers_in_production_with_no_cloud_api_resource(self, _, monkeypatch):
+        """Test that headers include access token from ManagedIdentityCredential in production even if CLOUD_API_RESOURCE is not set."""
+        monkeypatch.setenv("ENVIRONMENT", "prod")
+        with patch("services.dicom.dicom_uploader.ManagedIdentityCredential") as mock_credential:
+            mock_credential_instance = Mock()
+            mock_credential_instance.get_token.return_value.token = "prod_access_token"
+            mock_credential.return_value = mock_credential_instance
+
+            assert DICOMUploader().headers == {"Authorization": "Bearer prod_access_token"}
+            assert mock_credential_instance.get_token.call_args[0][0] is None
+
     def test_upload_headers_without_any_token(self, _):
         """Test that headers include empty token if neither CLOUD_API_RESOURCE nor CLOUD_API_TOKEN is set."""
         assert DICOMUploader().headers == {"Authorization": "Bearer "}