Dunitz/czi hosted test server (#2254)

update hosted unit tests to use wsgi server instead of cellxgene packages
This commit is contained in:
Madison Dunitz
2021-06-30 15:07:54 -07:00
committed by GitHub
parent 82de4178d9
commit 5007d307a2
17 changed files with 514 additions and 624 deletions
@@ -1,13 +1,12 @@
import json
import unittest
import requests
from backend.czi_hosted.common.config.app_config import AppConfig
from backend.test.test_czi_hosted.unit import test_server
from backend.test import FIXTURES_ROOT
from backend.test.test_czi_hosted.unit import BaseTest
class AuthTest(unittest.TestCase):
class AuthTest(BaseTest):
def setUp(self):
self.dataset_dataroot = FIXTURES_ROOT
@@ -18,13 +17,13 @@ class AuthTest(unittest.TestCase):
app_config.update_default_dataset_config(user_annotations__enable=False)
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
config = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").json()
self.assertNotIn("authentication", config["config"])
self.assertIsNone(userinfo)
server= self.create_app(app_config)
server.testing = True
session = server.test_client()
config = json.loads(session.get(f"{self.TEST_URL_BASE}config").data)
userinfo = json.loads(session.get(f"{self.TEST_URL_BASE}userinfo").data)
self.assertNotIn("authentication", config["config"])
self.assertIsNone(userinfo)
def test_auth_session(self):
app_config = AppConfig()
@@ -33,14 +32,16 @@ class AuthTest(unittest.TestCase):
app_config.update_default_dataset_config(user_annotations__enable=True)
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
config = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").json()
server = self.create_app(app_config)
server.auth.is_user_authenticated = lambda: True
server.testing = True
session = server.test_client()
config = json.loads(session.get(f"{self.TEST_URL_BASE}config").data)
userinfo = json.loads(session.get(f"{self.TEST_URL_BASE}userinfo").data)
self.assertFalse(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "anonymous")
self.assertFalse(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "anonymous")
def test_auth_test(self):
app_config = AppConfig()
@@ -60,57 +61,59 @@ class AuthTest(unittest.TestCase):
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
server=self.create_app(app_config)
server.testing = True
session = server.test_client()
# auth datasets
config = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/userinfo").json()
# auth datasets
config = json.loads(session.get("/auth/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get(f"/auth/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
self.assertEqual(login_uri, "/login?dataset=auth/pbmc3k.cxg")
self.assertEqual(logout_uri, "/logout?dataset=auth/pbmc3k.cxg")
self.assertEqual(login_uri, "/login?dataset=auth/pbmc3k.cxg")
self.assertEqual(logout_uri, "/logout?dataset=auth/pbmc3k.cxg")
r = session.get(f"{server}/{login_uri}")
# check that the login redirect worked
self.assertEqual(r.history[0].status_code, 302)
self.assertEqual(r.url, f"{server}/auth/pbmc3k.cxg")
response = session.get(login_uri)
# check that the login redirect worked
config = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/userinfo").json()
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "test_account")
self.assertEqual(userinfo["userinfo"]["picture"], None)
self.assertTrue(config["config"]["parameters"]["annotations"])
self.assertEqual(response.status_code, 302)
self.assertEqual(response.headers['Location'], 'http://localhost/auth/pbmc3k.cxg')
config = json.loads(session.get("/auth/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get("/auth/pbmc3k.cxg/api/v0.2/userinfo").data)
r = session.get(f"{server}/{logout_uri}")
# check that the logout redirect worked
self.assertEqual(r.history[0].status_code, 302)
self.assertEqual(r.url, f"{server}/auth/pbmc3k.cxg")
config = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/userinfo").json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "test_account")
self.assertEqual(userinfo["userinfo"]["picture"], None)
self.assertTrue(config["config"]["parameters"]["annotations"])
# no-auth datasets
config = session.get(f"{server}/no-auth/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/no-auth/pbmc3k.cxg/api/v0.2/userinfo").json()
self.assertIsNone(userinfo)
self.assertFalse(config["config"]["parameters"]["annotations"])
response = session.get(logout_uri)
# check that the logout redirect worked
# login with a picture
session.get(f"{server}/{login_uri}&picture=myimage.png")
userinfo = session.get(f"{server}/auth/pbmc3k.cxg/api/v0.2/userinfo").json()
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["picture"], "myimage.png")
self.assertEqual(response.status_code, 302)
config = json.loads(session.get("/auth/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get("/auth/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
# no-auth datasets
config = json.loads(session.get("/no-auth/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get("/no-auth/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertIsNone(userinfo)
self.assertFalse(config["config"]["parameters"]["annotations"])
# login with a picture
session.get(f"{login_uri}&picture=myimage.png")
userinfo = json.loads(session.get("/auth/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["picture"], "myimage.png")
def test_auth_test_single(self):
app_config = AppConfig()
@@ -122,38 +125,44 @@ class AuthTest(unittest.TestCase):
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
server = self.create_app(app_config)
server.testing = True
session = server.test_client()
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
config = json.loads(session.get("/api/v0.2/config").data)
userinfo = json.loads(session.get("/api/v0.2/userinfo").data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
self.assertEqual(login_uri, "/login")
self.assertEqual(logout_uri, "/logout")
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
response = session.get(f"{server}/{login_uri}")
# check that the login redirect worked
self.assertEqual(response.history[0].status_code, 302)
self.assertEqual(response.url, f"{server}/")
self.assertEqual(login_uri, "/login")
self.assertEqual(logout_uri, "/logout")
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "test_account")
self.assertTrue(config["config"]["parameters"]["annotations"])
response = session.get(f"{server}/{logout_uri}")
# check that the logout redirect worked
self.assertEqual(response.history[0].status_code, 302)
self.assertEqual(response.url, f"{server}/")
config = session.get(f"{server}/api/v0.2/config").json()
userinfo = session.get(f"{server}/api/v0.2/userinfo").json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
# check that the login redirect worked
with server.test_client() as session:
response = session.get(login_uri)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.headers['Location'], "http://localhost/")
config = json.loads(session.get("api/v0.2/config").data)
userinfo = json.loads(session.get("/api/v0.2/userinfo").data)
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "test_account")
self.assertTrue(config["config"]["parameters"]["annotations"])
response = session.get(logout_uri)
# check that the logout redirect worked
self.assertEqual(response.status_code, 302)
self.assertEqual(response.headers['Location'], "http://localhost/")
config = json.loads(session.get("/api/v0.2/config").data)
userinfo = json.loads(session.get("/api/v0.2/userinfo").data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
@@ -10,7 +10,6 @@ from multiprocessing import Process
import jose
from backend.czi_hosted.common.config.app_config import AppConfig
from backend.test.test_czi_hosted.unit import test_server
from backend.test import FIXTURES_ROOT
# This tests the oauth authentication type.
@@ -20,6 +19,8 @@ from backend.test import FIXTURES_ROOT
# oauth server.
# number of seconds that the oauth token is valid
from backend.test.test_czi_hosted.unit import BaseTest
TOKEN_EXPIRES = 2
# Create a mocked out oauth token, which servers all the endpoints needed by the oauth type.
@@ -69,7 +70,7 @@ def launch_mock_oauth(mock_port):
mock_oauth_app.run(port=mock_port)
class AuthTest(unittest.TestCase):
class AuthTest(BaseTest):
@classmethod
def setUpClass(cls):
# The port that the mock oauth server will listen on
@@ -119,90 +120,95 @@ class AuthTest(unittest.TestCase):
app_config.update_server_config(multi_dataset__dataroot=self.dataset_dataroot)
app_config.complete_config()
with test_server(app_config=app_config) as server:
session = requests.Session()
server= self.create_app(app_config)
server.testing = True
session = server.test_client()
# auth datasets
config = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").json()
# auth datasets
config = json.loads(session.get("/d/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get("/d/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["authentication"]["requires_client_login"])
self.assertTrue(config["config"]["parameters"]["annotations"])
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
login_uri = config["config"]["authentication"]["login"]
logout_uri = config["config"]["authentication"]["logout"]
self.assertEqual(login_uri, f"{server}/login?dataset=d/pbmc3k.cxg/")
self.assertEqual(logout_uri, f"{server}/logout?dataset=d/pbmc3k.cxg/")
self.assertEqual(login_uri, "http://localhost:5005/login?dataset=d/pbmc3k.cxg/")
self.assertEqual(logout_uri, "http://localhost:5005/logout?dataset=d/pbmc3k.cxg/")
r = session.get(login_uri)
# check that the login redirect worked
self.assertEqual(r.history[0].status_code, 302)
self.assertEqual(r.url, f"{server}/d/pbmc3k.cxg/")
config = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").json()
response = session.get(login_uri)
# check that the login redirect worked
self.assertEqual(response.status_code, 302)
config = json.loads(session.get("/d/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get("/d/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "fake_user")
self.assertEqual(userinfo["userinfo"]["email"], "fake_user@email.com")
self.assertTrue(config["config"]["parameters"]["annotations"])
if cookie_key:
cookie = session.cookies.get(cookie_key)
token = json.loads(base64.b64decode(cookie))
access_token_before = token.get("access_token")
id_token_before = token.get("id_token")
# let the token expire
time.sleep(TOKEN_EXPIRES + 1)
# check that refresh works
session.get(login_uri)
userinfo = json.loads(session.get(f"/d/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "fake_user")
self.assertEqual(userinfo["userinfo"]["email"], "fake_user@email.com")
self.assertTrue(config["config"]["parameters"]["annotations"])
if cookie_key:
cookie = session.cookies.get(cookie_key)
token = json.loads(base64.b64decode(cookie))
access_token_before = token.get("access_token")
id_token_before = token.get("id_token")
cookie = session.cookies.get(cookie_key)
token = json.loads(base64.b64decode(cookie))
access_token_after = token.get("access_token")
id_token_after = token.get("id_token")
# let the token expire
time.sleep(TOKEN_EXPIRES + 1)
self.assertNotEqual(access_token_before, access_token_after)
self.assertNotEqual(id_token_before, id_token_after)
# check that refresh works
session.get(login_uri)
userinfo = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").json()
self.assertTrue(userinfo["userinfo"]["is_authenticated"])
self.assertEqual(userinfo["userinfo"]["username"], "fake_user")
cookie = session.cookies.get(cookie_key)
token = json.loads(base64.b64decode(cookie))
access_token_after = token.get("access_token")
id_token_after = token.get("id_token")
self.assertNotEqual(access_token_before, access_token_after)
self.assertNotEqual(id_token_before, id_token_after)
# invalid cookie is rejected
session.cookies.set(cookie_key, "TEST_" + cookie)
self.assertTrue(cookie_key in session.cookies)
response = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo")
# this is not an error, the invalid cookie is just ignored.
self.assertEqual(response.status_code, 200)
userinfo = response.json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
# invalid id_token is rejected
test_token = token
test_token["id_token"] = "TEST_" + id_token_after
encoded_cookie = base64.b64encode(json.dumps(test_token).encode()).decode()
session.cookies.set(cookie_key, encoded_cookie)
response = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo")
# this is not an error, the invalid id_token is just ignored.
self.assertEqual(response.status_code, 200)
userinfo = response.json()
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
r = session.get(logout_uri)
# check that the logout redirect worked
self.assertEqual(r.history[0].status_code, 302)
self.assertEqual(r.url, f"{server}/d/pbmc3k.cxg/")
config = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/config").json()
userinfo = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").json()
# invalid cookie is rejected
session.cookies.set(cookie_key, "TEST_" + cookie)
self.assertTrue(cookie_key in session.cookies)
response = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo")
# this is not an error, the invalid cookie is just ignored.
self.assertEqual(response.status_code, 200)
userinfo = json.loads(response.data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
# invalid id_token is rejected
test_token = token
test_token["id_token"] = "TEST_" + id_token_after
encoded_cookie = base64.b64encode(json.dumps(test_token).encode()).decode()
session.cookies.set(cookie_key, encoded_cookie)
response = session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo")
# this is not an error, the invalid id_token is just ignored.
self.assertEqual(response.status_code, 200)
userinfo = json.loads(response.data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
r = session.get(logout_uri)
# check that the logout redirect worked
self.assertEqual(r.history[0].status_code, 302)
self.assertEqual(r.url, f"{server}/d/pbmc3k.cxg/")
config = json.loads(session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/config").data)
userinfo = json.loads(session.get(f"{server}/d/pbmc3k.cxg/api/v0.2/userinfo").data)
self.assertFalse(userinfo["userinfo"]["is_authenticated"])
self.assertIsNone(userinfo["userinfo"]["username"])
self.assertTrue(config["config"]["parameters"]["annotations"])
@unittest.skip("turn on when we utilizing auth in the explorer")
def test_auth_oauth_session(self):
# test with session cookies
app_config = AppConfig()
@@ -210,6 +216,7 @@ class AuthTest(unittest.TestCase):
app_config.update_server_config(authentication__params_oauth__session_cookie=True,)
self.auth_flow(app_config)
@unittest.skip("turn on when we utilizing auth in the explorer")
def test_auth_oauth_cookie(self):
# test with specified cookie
app_config = AppConfig()