Automated tests for examples in docs

.gitignore

@@ -21,3 +21,4 @@ dist/
 t.py
 
 t2.py
+tests/examples/tmp_*py

docs/examples/native_spa_pkce_auth0.py

@@ -1,8 +1,8 @@
 
-client_id = 'your_client_id'
+client_id = 'OAUTH_CLIENT_ID'
 
-authorization_base_url = "https://dev-foobar.eu.auth0.com/authorize"
-token_url = "https://dev-foobar.eu.auth0.com/oauth/token"
+authorization_base_url = "https://OAUTH_IDP_DOMAIN/authorize"
+token_url = "https://OAUTH_IDP_DOMAIN/oauth/token"
 scope = ["openid"]
 
 from requests_oauthlib import OAuth2Session

requirements-test.txt

@@ -2,3 +2,4 @@
 coveralls==3.3.1
 mock==4.0.3
 requests-mock==1.11.0
+selenium==4.18.1

tests/examples/base.py

@@ -0,0 +1,106 @@
+import os.path
+import os
+import subprocess
+import shlex
+import shutil
+from selenium import webdriver
+from selenium.webdriver.common.by import By
+from selenium.webdriver.common.keys import Keys
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.support.wait import WebDriverWait
+
+
+cwd = os.path.dirname(os.path.realpath(__file__))
+
+
+class Sample():
+    def setUp(self):
+        super().setUp()
+        self.proc = None
+        self.outputs = []
+
+    def tearDown(self):
+        super().tearDown()
+        if self.proc is not None:
+            self.proc.stdin.close()
+            self.proc.stdout.close()
+            self.proc.kill()
+
+    def replaceVariables(self, filein ,fileout, vars):
+        with open(filein, "rt") as fin:
+            with open(fileout, "wt") as fout:
+                for line in fin:
+                    for k, v in vars.items():
+                        line = line.replace(k, v)
+                    fout.write(line)
+
+    def run_sample(self, filepath, variables):
+        inpath = os.path.join(cwd, "..", "..", "docs", "examples", filepath)
+        outpath = os.path.join(cwd, "tmp_{}".format(filepath))
+        self.replaceVariables(inpath, outpath, variables)
+
+        self.proc = subprocess.Popen(
+            [shutil.which("python"),
+             outpath],
+            text=True, bufsize=1,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE
+        )
+
+    def write(self, string):
+        self.proc.stdin.write(string)
+        self.proc.stdin.flush()
+
+    def wait_for_pattern(self, pattern):
+        try:
+            while True:
+                line = self.proc.stdout.readline()
+                self.outputs.append(line)
+                if pattern in line:
+                    return line
+        except subprocess.TimeoutExpired:
+            self.assertTrue(False, "timeout when looking for output")
+
+    def wait_for_end(self):
+        try:
+            outs, err = self.proc.communicate(timeout=10)
+            self.outputs += filter(lambda x: x != '', outs.split('\n'))
+        except subprocess.TimeoutExpired:
+            self.assertTrue(False, "timeout when looking for output")
+        return self.outputs[-1]
+
+
+
+class Browser():
+    def setUp(self):
+        super().setUp()
+        options = webdriver.ChromeOptions()
+        options.add_argument("--headless=new")
+        self.driver = webdriver.Chrome(options=options)
+        self.user_username = os.environ.get("AUTH0_USERNAME")
+        self.user_password = os.environ.get("AUTH0_PASSWORD")
+
+        if not self.user_username or not self.user_password:
+            self.skipTest("auth0 is not configured properly")
+
+    def tearDown(self):
+        super().tearDown()
+        self.driver.quit()
+
+    def authorize_auth0(self, authorize_url, expected_redirect_uri):
+        self.driver.get(authorize_url)
+        username = self.driver.find_element(By.ID, "username")
+        password = self.driver.find_element(By.ID, "password")
+
+        wait = WebDriverWait(self.driver, timeout=2)
+        wait.until(lambda d : username.is_displayed())
+        wait.until(lambda d : password.is_displayed())
+
+        username.clear()
+        username.send_keys(self.user_username)
+        password.send_keys(self.user_password)
+        username.send_keys(Keys.RETURN)
+
+        wait.until(EC.url_contains(expected_redirect_uri))
+        return self.driver.current_url
+

tests/examples/test_native_spa_pkce_auth0.py

@@ -0,0 +1,39 @@
+import os
+import unittest
+
+from . import base
+
+class TestNativeAuth0Test(base.Sample, base.Browser, unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.client_id = os.environ.get("AUTH0_PKCE_CLIENT_ID")
+        self.idp_domain = os.environ.get("AUTH0_DOMAIN")
+
+        if not self.client_id or not self.idp_domain:
+            self.skipTest("native auth0 is not configured properly")
+
+    def test_login(self):
+        # redirect_uri is http://
+        os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = "1"
+
+        self.run_sample(
+            "native_spa_pkce_auth0.py", {
+                "OAUTH_CLIENT_ID": self.client_id,
+                "OAUTH_IDP_DOMAIN": self.idp_domain,
+            }
+        )
+        authorize_url = self.wait_for_pattern("https://")
+        redirect_uri = self.authorize_auth0(authorize_url, "http://")
+        self.write(redirect_uri)
+        last_line = self.wait_for_end()
+
+        import ast
+        response = ast.literal_eval(last_line)
+        self.assertIn("access_token", response)
+        self.assertIn("id_token", response)
+        self.assertIn("scope", response)
+        self.assertIn("openid", response["scope"])
+        self.assertIn("expires_in", response)
+        self.assertIn("expires_at", response)
+        self.assertIn("token_type", response)
+        self.assertEqual("Bearer", response["token_type"])

tox.ini

@@ -6,6 +6,8 @@ description=run test on {basepython}
 deps=
     -r{toxinidir}/requirements-test.txt
 commands=coverage run --source=requests_oauthlib -m unittest discover
+pass_env=OAUTH_*
+        AUTH0_*
 
 # tox -e docs to mimic readthedocs build.
 # should be similar to .readthedocs.yaml pipeline