Regional endpoint test cases do not rely on env var REGION_NAME

Author: rayluo

tests/test_e2e.py

@@ -137,10 +137,15 @@ def assertCacheWorksForApp(self, result_from_wire, scope):
     def _test_username_password(self,
             authority=None, client_id=None, username=None, password=None, scope=None,
             client_secret=None,  # Since MSAL 1.11, confidential client has ROPC too
+            azure_region=None,
+            http_client=None,
             **ignored):
         assert authority and client_id and username and password and scope
         self.app = msal.ClientApplication(
-            client_id, authority=authority, http_client=MinimalHttpClient(),
+            client_id, authority=authority,
+            http_client=http_client or MinimalHttpClient(),
+            azure_region=azure_region,  # Regional endpoint does not support ROPC.
+                # Here we just use it to test a regional app won't break ROPC.
             client_credential=client_secret)
         result = self.app.acquire_token_by_username_password(
             username, password, scopes=scope)
@@ -541,11 +546,16 @@ def _test_acquire_token_by_auth_code_flow(
                 error_description=result.get("error_description")))
         self.assertCacheWorksForUser(result, scope, username=None)
 
-    def _test_acquire_token_obo(self, config_pca, config_cca):
+    def _test_acquire_token_obo(self, config_pca, config_cca,
+            azure_region=None,  # Regional endpoint does not really support OBO.
+                # Here we just test regional apps won't adversely break OBO
+            http_client=None,
+            ):
         # 1. An app obtains a token representing a user, for our mid-tier service
         pca = msal.PublicClientApplication(
             config_pca["client_id"], authority=config_pca["authority"],
-            http_client=MinimalHttpClient())
+            azure_region=azure_region,
+            http_client=http_client or MinimalHttpClient())
         pca_result = pca.acquire_token_by_username_password(
             config_pca["username"],
             config_pca["password"],
@@ -560,7 +570,8 @@ def _test_acquire_token_obo(self, config_pca, config_cca):
             config_cca["client_id"],
             client_credential=config_cca["client_secret"],
             authority=config_cca["authority"],
-            http_client=MinimalHttpClient(),
+            azure_region=azure_region,
+            http_client=http_client or MinimalHttpClient(),
             # token_cache= ...,  # Default token cache is all-tokens-store-in-memory.
                 # That's fine if OBO app uses short-lived msal instance per session.
                 # Otherwise, the OBO app need to implement a one-cache-per-user setup.
@@ -778,6 +789,7 @@ def test_b2c_acquire_token_by_ropc(self):
 
 class WorldWideRegionalEndpointTestCase(LabBasedTestCase):
     region = "westus"
+    timeout = 2  # Short timeout makes this test case responsive on non-VM
 
     def test_acquire_token_for_client_should_hit_regional_endpoint(self):
         """This is the only grant supported by regional endpoint, for now"""
@@ -808,15 +820,6 @@ def test_acquire_token_for_client_should_hit_regional_endpoint(self):
         self.assertIn('access_token', result)
         self.assertCacheWorksForApp(result, scopes)
 
-
-class RegionalEndpointViaEnvVarTestCase(WorldWideRegionalEndpointTestCase):
-
-    def setUp(self):
-        os.environ["REGION_NAME"] = "eastus"
-
-    def tearDown(self):
-        del os.environ["REGION_NAME"]
-
     @unittest.skipUnless(
         os.getenv("LAB_OBO_CLIENT_SECRET"),
         "Need LAB_OBO_CLIENT_SECRET from https://aka.ms/GetLabSecret?Secret=TodoListServiceV2-OBO")
@@ -842,7 +845,11 @@ def test_cca_obo_should_bypass_regional_endpoint_therefore_still_work(self):
         config_pca["password"] = self.get_lab_user_secret(config_pca["lab_name"])
         config_pca["scope"] = ["api://%s/read" % config_cca["client_id"]]
 
-        self._test_acquire_token_obo(config_pca, config_cca)
+        self._test_acquire_token_obo(
+            config_pca, config_cca,
+            azure_region=self.region,
+            http_client=MinimalHttpClient(timeout=self.timeout),
+            )
 
     @unittest.skipUnless(
         os.getenv("LAB_OBO_CLIENT_SECRET"),
@@ -859,7 +866,10 @@ def test_cca_ropc_should_bypass_regional_endpoint_therefore_still_work(self):
         config["client_id"] = os.getenv("LAB_OBO_CONFIDENTIAL_CLIENT_ID")
         config["scope"] = ["https://graph.microsoft.com/.default"]
         config["client_secret"] = os.getenv("LAB_OBO_CLIENT_SECRET")
-        self._test_username_password(**config)
+        self._test_username_password(
+            azure_region=self.region,
+            http_client=MinimalHttpClient(timeout=self.timeout),
+            **config)
 
 
 class ArlingtonCloudTestCase(LabBasedTestCase):