Move some tests from test_application to test_e2e

Author: rayluo

tests/test_application.py

@@ -1,3 +1,5 @@
+# Note: Since Aug 2019 we move all e2e tests into test_e2e.py,
+# so this test_application file contains only unit tests without dependency.
 import os
 import json
 import logging
@@ -13,91 +15,11 @@
 from tests.test_token_cache import TokenCacheTestCase
 
 
-THIS_FOLDER = os.path.dirname(__file__)
-CONFIG_FILE = os.path.join(THIS_FOLDER, 'config.json')
-CONFIG = {}
-if os.path.exists(CONFIG_FILE):
-    with open(CONFIG_FILE) as conf:
-        CONFIG = json.load(conf)
-
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.DEBUG)
 
 
-class Oauth2TestCase(unittest.TestCase):
-
-    def assertLoosely(self, response, assertion=None,
-            skippable_errors=("invalid_grant", "interaction_required")):
-        if response.get("error") in skippable_errors:
-            logger.debug("Response = %s", response)
-            # Some of these errors are configuration issues, not library issues
-            raise unittest.SkipTest(response.get("error_description"))
-        else:
-            if assertion is None:
-                assertion = lambda: self.assertIn(
-                    "access_token", response,
-                    "{error}: {error_description}".format(
-                        # Do explicit response.get(...) rather than **response
-                        error=response.get("error"),
-                        error_description=response.get("error_description")))
-            assertion()
-
-    def assertCacheWorks(self, result_from_wire):
-        result = result_from_wire
-        # You can filter by predefined username, or let end user to choose one
-        accounts = self.app.get_accounts(username=CONFIG.get("username"))
-        self.assertNotEqual(0, len(accounts))
-        account = accounts[0]
-        # Going to test acquire_token_silent(...) to locate an AT from cache
-        result_from_cache = self.app.acquire_token_silent(
-                CONFIG["scope"], account=account)
-        self.assertIsNotNone(result_from_cache)
-        self.assertEqual(result['access_token'], result_from_cache['access_token'],
-                "We should get a cached AT")
-
-        # Going to test acquire_token_silent(...) to obtain an AT by a RT from cache
-        self.app.token_cache._cache["AccessToken"] = {}  # A hacky way to clear ATs
-        result_from_cache = self.app.acquire_token_silent(
-                CONFIG["scope"], account=account)
-        self.assertIsNotNone(result_from_cache,
-                "We should get a result from acquire_token_silent(...) call")
-        self.assertNotEqual(result['access_token'], result_from_cache['access_token'],
-                "We should get a fresh AT (via RT)")
-
-
-@unittest.skipUnless("client_id" in CONFIG, "client_id missing")
-class TestConfidentialClientApplication(unittest.TestCase):
-
-    def assertCacheWorks(self, result_from_wire, result_from_cache):
-        self.assertIsNotNone(result_from_cache)
-        self.assertEqual(
-            result_from_wire['access_token'], result_from_cache['access_token'])
-
-    @unittest.skipUnless("client_secret" in CONFIG, "Missing client secret")
-    def test_client_secret(self):
-        app = ConfidentialClientApplication(
-            CONFIG["client_id"], client_credential=CONFIG.get("client_secret"),
-            authority=CONFIG.get("authority"))
-        scope = CONFIG.get("scope", [])
-        result = app.acquire_token_for_client(scope)
-        self.assertIn('access_token', result)
-        self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None))
-
-    @unittest.skipUnless("client_certificate" in CONFIG, "Missing client cert")
-    def test_client_certificate(self):
-        client_certificate = CONFIG["client_certificate"]
-        assert ("private_key_path" in client_certificate
-                and "thumbprint" in client_certificate)
-        key_path = os.path.join(THIS_FOLDER, client_certificate['private_key_path'])
-        with open(key_path) as f:
-            pem = f.read()
-        app = ConfidentialClientApplication(
-            CONFIG['client_id'],
-            {"private_key": pem, "thumbprint": client_certificate["thumbprint"]})
-        scope = CONFIG.get("scope", [])
-        result = app.acquire_token_for_client(scope)
-        self.assertIn('access_token', result)
-        self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None))
+class TestHelperExtractCerts(unittest.TestCase):  # It is used by SNI scenario
 
     def test_extract_a_tag_less_public_cert(self):
         pem = "my_cert"
@@ -116,92 +38,13 @@ def test_extract_multiple_tag_enclosed_certs(self):
         -----BEGIN CERTIFICATE-----
         my_cert1
         -----END CERTIFICATE-----
-        
+
         -----BEGIN CERTIFICATE-----
         my_cert2
         -----END CERTIFICATE-----
         """
         self.assertEqual(["my_cert1", "my_cert2"], extract_certs(pem))
 
-    @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert")
-    def test_subject_name_issuer_authentication(self):
-        assert ("private_key_file" in CONFIG
-                and "thumbprint" in CONFIG and "public_certificate" in CONFIG)
-        with open(os.path.join(THIS_FOLDER, CONFIG['private_key_file'])) as f:
-            pem = f.read()
-        with open(os.path.join(THIS_FOLDER, CONFIG['public_certificate'])) as f:
-            public_certificate = f.read()
-        app = ConfidentialClientApplication(
-            CONFIG['client_id'], authority=CONFIG["authority"],
-            client_credential={"private_key": pem, "thumbprint": CONFIG["thumbprint"],
-                               "public_certificate": public_certificate})
-        scope = CONFIG.get("scope", [])
-        result = app.acquire_token_for_client(scope)
-        self.assertIn('access_token', result)
-        self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None))
-
-@unittest.skipUnless("client_id" in CONFIG, "client_id missing")
-class TestPublicClientApplication(Oauth2TestCase):
-
-    @unittest.skipUnless("username" in CONFIG and "password" in CONFIG, "Missing U/P")
-    def test_username_password(self):
-        self.app = PublicClientApplication(
-                CONFIG["client_id"], authority=CONFIG["authority"])
-        result = self.app.acquire_token_by_username_password(
-                CONFIG["username"], CONFIG["password"], scopes=CONFIG.get("scope"))
-        self.assertLoosely(result)
-        self.assertCacheWorks(result)
-
-    def test_device_flow(self):
-        self.app = PublicClientApplication(
-            CONFIG["client_id"], authority=CONFIG["authority"])
-        flow = self.app.initiate_device_flow(scopes=CONFIG.get("scope"))
-        assert "user_code" in flow, str(flow)  # Provision or policy might block DF
-        logging.warning(flow["message"])
-
-        duration = 30
-        logging.warning("We will wait up to %d seconds for you to sign in" % duration)
-        flow["expires_at"] = time.time() + duration  # Shorten the time for quick test
-        result = self.app.acquire_token_by_device_flow(flow)
-        self.assertLoosely(
-                result,
-                assertion=lambda: self.assertIn('access_token', result),
-                skippable_errors=self.app.client.DEVICE_FLOW_RETRIABLE_ERRORS)
-
-        if "access_token" in result:
-            self.assertCacheWorks(result)
-
-
-@unittest.skipUnless("client_id" in CONFIG, "client_id missing")
-class TestClientApplication(Oauth2TestCase):
-
-    @classmethod
-    def setUpClass(cls):
-        cls.app = ClientApplication(
-            CONFIG["client_id"], client_credential=CONFIG.get("client_secret"),
-            authority=CONFIG.get("authority"))
-
-    @unittest.skipUnless("scope" in CONFIG, "Missing scope")
-    def test_auth_code(self):
-        from msal.oauth2cli.authcode import obtain_auth_code
-        port = CONFIG.get("listen_port", 44331)
-        redirect_uri = "http://localhost:%s" % port
-        auth_request_uri = self.app.get_authorization_request_url(
-            CONFIG["scope"], redirect_uri=redirect_uri)
-        ac = obtain_auth_code(port, auth_uri=auth_request_uri)
-        self.assertNotEqual(ac, None)
-
-        result = self.app.acquire_token_by_authorization_code(
-            ac, CONFIG["scope"], redirect_uri=redirect_uri)
-        logging.debug("cache = %s", json.dumps(self.app.token_cache._cache, indent=4))
-        self.assertIn(
-            "access_token", result,
-            "{error}: {error_description}".format(
-                # Note: No interpolation here, cause error won't always present
-                error=result.get("error"),
-                error_description=result.get("error_description")))
-        self.assertCacheWorks(result)
-
 
 class TestClientApplicationAcquireTokenSilentFociBehaviors(unittest.TestCase):
 

tests/test_client.py

@@ -78,6 +78,7 @@ def load_conf(filename):
 
 # Since the OAuth2 specs uses snake_case, this test config also uses snake_case
 @unittest.skipUnless("client_id" in CONFIG, "client_id missing")
+@unittest.skipUnless(CONFIG.get("openid_configuration"), "openid_configuration missing")
 class TestClient(Oauth2TestCase):
 
     @classmethod

tests/test_e2e.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import json
+import time
 
 import requests
 
@@ -31,25 +32,34 @@ def assertLoosely(self, response, assertion=None,
                         error_description=response.get("error_description")))
             assertion()
 
-    def assertCacheWorks(self, result_from_wire, username, scope):
-        result = result_from_wire
+    def assertCacheWorksForUser(self, result_from_wire, scope, username=None):
         # You can filter by predefined username, or let end user to choose one
         accounts = self.app.get_accounts(username=username)
         self.assertNotEqual(0, len(accounts))
         account = accounts[0]
         # Going to test acquire_token_silent(...) to locate an AT from cache
         result_from_cache = self.app.acquire_token_silent(scope, account=account)
         self.assertIsNotNone(result_from_cache)
-        self.assertEqual(result['access_token'], result_from_cache['access_token'],
-                "We should get a cached AT")
+        self.assertEqual(
+            result_from_wire['access_token'], result_from_cache['access_token'],
+            "We should get a cached AT")
 
         # Going to test acquire_token_silent(...) to obtain an AT by a RT from cache
         self.app.token_cache._cache["AccessToken"] = {}  # A hacky way to clear ATs
         result_from_cache = self.app.acquire_token_silent(scope, account=account)
         self.assertIsNotNone(result_from_cache,
                 "We should get a result from acquire_token_silent(...) call")
-        self.assertNotEqual(result['access_token'], result_from_cache['access_token'],
-                "We should get a fresh AT (via RT)")
+        self.assertNotEqual(
+            result_from_wire['access_token'], result_from_cache['access_token'],
+            "We should get a fresh AT (via RT)")
+
+    def assertCacheWorksForApp(self, result_from_wire, scope):
+        # Going to test acquire_token_silent(...) to locate an AT from cache
+        result_from_cache = self.app.acquire_token_silent(scope, account=None)
+        self.assertIsNotNone(result_from_cache)
+        self.assertEqual(
+            result_from_wire['access_token'], result_from_cache['access_token'],
+            "We should get a cached AT")
 
     def _test_username_password(self,
             authority=None, client_id=None, username=None, password=None, scope=None,
@@ -60,19 +70,137 @@ def _test_username_password(self,
             username, password, scopes=scope)
         self.assertLoosely(result)
         # self.assertEqual(None, result.get("error"), str(result))
-        self.assertCacheWorks(result, username, scope)
+        self.assertCacheWorksForUser(result, scope, username=username)
 
 
-CONFIG = os.path.join(os.path.dirname(__file__), "config.json")
-@unittest.skipIf(not os.path.exists(CONFIG), "Optional %s not found" % CONFIG)
+THIS_FOLDER = os.path.dirname(__file__)
+CONFIG = os.path.join(THIS_FOLDER, "config.json")
+@unittest.skipUnless(os.path.exists(CONFIG), "Optional %s not found" % CONFIG)
 class FileBasedTestCase(E2eTestCase):
-    def setUp(self):
+    # This covers scenarios that are not currently available for test automation.
+    # So they mean to be run on maintainer's machine for semi-automated tests.
+
+    @classmethod
+    def setUpClass(cls):
         with open(CONFIG) as f:
-            self.config = json.load(f)
+            cls.config = json.load(f)
+
+    def skipUnlessWithConfig(self, fields):
+        for field in fields:
+            if field not in self.config:
+                self.skipTest('Skipping due to lack of configuration "%s"' % field)
 
     def test_username_password(self):
+        self.skipUnlessWithConfig(["client_id", "username", "password", "scope"])
         self._test_username_password(**self.config)
 
+    def test_auth_code(self):
+        self.skipUnlessWithConfig(["client_id", "scope"])
+        from msal.oauth2cli.authcode import obtain_auth_code
+        self.app = msal.ClientApplication(
+            self.config["client_id"],
+            client_credential=self.config.get("client_secret"),
+            authority=self.config.get("authority"))
+        port = self.config.get("listen_port", 44331)
+        redirect_uri = "http://localhost:%s" % port
+        auth_request_uri = self.app.get_authorization_request_url(
+            self.config["scope"], redirect_uri=redirect_uri)
+        ac = obtain_auth_code(port, auth_uri=auth_request_uri)
+        self.assertNotEqual(ac, None)
+
+        result = self.app.acquire_token_by_authorization_code(
+            ac, self.config["scope"], redirect_uri=redirect_uri)
+        logger.debug("%s.cache = %s",
+            self.id(), json.dumps(self.app.token_cache._cache, indent=4))
+        self.assertIn(
+            "access_token", result,
+            "{error}: {error_description}".format(
+                # Note: No interpolation here, cause error won't always present
+                error=result.get("error"),
+                error_description=result.get("error_description")))
+        self.assertCacheWorksForUser(result, self.config["scope"], username=None)
+
+    def test_client_secret(self):
+        self.skipUnlessWithConfig(["client_id", "client_secret"])
+        self.app = msal.ConfidentialClientApplication(
+            self.config["client_id"],
+            client_credential=self.config.get("client_secret"),
+            authority=self.config.get("authority"))
+        scope = self.config.get("scope", [])
+        result = self.app.acquire_token_for_client(scope)
+        self.assertIn('access_token', result)
+        self.assertCacheWorksForApp(result, scope)
+
+    def test_client_certificate(self):
+        self.skipUnlessWithConfig(["client_id", "client_certificate"])
+        client_cert = self.config["client_certificate"]
+        assert "private_key_path" in client_cert and "thumbprint" in client_cert
+        with open(os.path.join(THIS_FOLDER, client_cert['private_key_path'])) as f:
+            private_key = f.read()  # Should be in PEM format
+        self.app = msal.ConfidentialClientApplication(
+            self.config['client_id'],
+            {"private_key": private_key, "thumbprint": client_cert["thumbprint"]})
+        scope = self.config.get("scope", [])
+        result = self.app.acquire_token_for_client(scope)
+        self.assertIn('access_token', result)
+        self.assertCacheWorksForApp(result, scope)
+
+    def test_subject_name_issuer_authentication(self):
+        self.skipUnlessWithConfig(["client_id", "client_certificate"])
+        client_cert = self.config["client_certificate"]
+        assert "private_key_path" in client_cert and "thumbprint" in client_cert
+        if not "public_certificate" in client_cert:
+            self.skipTest("Skipping SNI test due to lack of public_certificate")
+        with open(os.path.join(THIS_FOLDER, client_cert['private_key_path'])) as f:
+            private_key = f.read()  # Should be in PEM format
+        with open(os.path.join(THIS_FOLDER, client_cert['public_certificate'])) as f:
+            public_certificate = f.read()
+        self.app = msal.ConfidentialClientApplication(
+            self.config['client_id'], authority=self.config["authority"],
+            client_credential={
+                "private_key": private_key,
+                "thumbprint": self.config["thumbprint"],
+                "public_certificate": public_certificate,
+                })
+        scope = self.config.get("scope", [])
+        result = self.app.acquire_token_for_client(scope)
+        self.assertIn('access_token', result)
+        self.assertCacheWorksForApp(result, scope)
+
+
+@unittest.skipUnless(os.path.exists(CONFIG), "Optional %s not found" % CONFIG)
+class DeviceFlowTestCase(E2eTestCase):  # A leaf class so it will be run only once
+    @classmethod
+    def setUpClass(cls):
+        with open(CONFIG) as f:
+            cls.config = json.load(f)
+
+    def test_device_flow(self):
+        scopes = self.config["scope"]
+        self.app = msal.PublicClientApplication(
+            self.config['client_id'], authority=self.config["authority"])
+        flow = self.app.initiate_device_flow(scopes=scopes)
+        assert "user_code" in flow, "DF does not seem to be provisioned: %s".format(
+            json.dumps(flow, indent=4))
+        logger.info(flow["message"])
+
+        duration = 60
+        logger.info("We will wait up to %d seconds for you to sign in" % duration)
+        flow["expires_at"] = min(  # Shorten the time for quick test
+            flow["expires_at"], time.time() + duration)
+        result = self.app.acquire_token_by_device_flow(flow)
+        self.assertLoosely(  # It will skip this test if there is no user interaction
+            result,
+            assertion=lambda: self.assertIn('access_token', result),
+            skippable_errors=self.app.client.DEVICE_FLOW_RETRIABLE_ERRORS)
+        if "access_token" not in result:
+            self.skip("End user did not complete Device Flow in time")
+        self.assertCacheWorksForUser(result, scopes, username=None)
+        result["access_token"] = result["refresh_token"] = "************"
+        logger.info(
+            "%s obtained tokens: %s", self.id(), json.dumps(result, indent=4))
+
+
 def get_lab_user(mam=False, mfa=False, isFederated=False, federationProvider=None):
     # Based on https://microsoft.sharepoint-df.com/teams/MSIDLABSExtended/SitePages/LAB.aspx
     user = requests.get("https://api.msidlab.com/api/user", params=dict(  # Publicly available