From 953d9b848827cfbcf1874728ac85aa5f6f4a97dc Mon Sep 17 00:00:00 2001 From: Revant Nandgaonkar Date: Mon, 21 Oct 2019 13:21:20 +0530 Subject: [PATCH] fix(tests): OAuth 2.0 tests --- .../doctype/oauth_client/test_records.json | 2 +- frappe/tests/test_api.py | 10 +- frappe/tests/{ui => }/test_oauth20.py | 147 +++++++++++------- 3 files changed, 98 insertions(+), 61 deletions(-) rename frappe/tests/{ui => }/test_oauth20.py (56%) diff --git a/frappe/integrations/doctype/oauth_client/test_records.json b/frappe/integrations/doctype/oauth_client/test_records.json index 904d959625..cff06457c5 100644 --- a/frappe/integrations/doctype/oauth_client/test_records.json +++ b/frappe/integrations/doctype/oauth_client/test_records.json @@ -11,6 +11,6 @@ "redirect_uris": "http://localhost", "response_type": "Code", "scopes": "all openid", - "skip_authorization": 0 + "skip_authorization": 1 } ] diff --git a/frappe/tests/test_api.py b/frappe/tests/test_api.py index 20dc25e013..232b2be4a8 100644 --- a/frappe/tests/test_api.py +++ b/frappe/tests/test_api.py @@ -12,7 +12,7 @@ import base64 class TestAPI(unittest.TestCase): def test_insert_many(self): server = FrappeClient(frappe.get_site_config().host_name, "Administrator", "admin", verify=False) - frappe.db.sql('delete from `tabNote` where title in ("Sing","a","song","of","sixpence")') + frappe.db.sql("delete from `tabNote` where title in ('Sing','a','song','of','sixpence')") frappe.db.commit() server.insert_many([ @@ -31,7 +31,7 @@ class TestAPI(unittest.TestCase): def test_create_doc(self): server = FrappeClient(frappe.get_site_config().host_name, "Administrator", "admin", verify=False) - frappe.db.sql('delete from `tabNote` where title = "test_create"') + frappe.db.sql("delete from `tabNote` where title = 'test_create'") frappe.db.commit() server.insert({"doctype": "Note", "public": True, "title": "test_create"}) @@ -46,7 +46,7 @@ class TestAPI(unittest.TestCase): def test_get_doc(self): server = FrappeClient(frappe.get_site_config().host_name, "Administrator", "admin", verify=False) - frappe.db.sql('delete from `tabNote` where title = "get_this"') + frappe.db.sql("delete from `tabNote` where title = 'get_this'") frappe.db.commit() server.insert_many([ @@ -57,7 +57,7 @@ class TestAPI(unittest.TestCase): def test_update_doc(self): server = FrappeClient(frappe.get_site_config().host_name, "Administrator", "admin", verify=False) - frappe.db.sql('delete from `tabNote` where title = "sing"') + frappe.db.sql("delete from `tabNote` where title in ('Sing','sing')") frappe.db.commit() server.insert({"doctype":"Note", "public": True, "title": "Sing"}) @@ -69,7 +69,7 @@ class TestAPI(unittest.TestCase): def test_delete_doc(self): server = FrappeClient(frappe.get_site_config().host_name, "Administrator", "admin", verify=False) - frappe.db.sql('delete from `tabNote` where title = "delete"') + frappe.db.sql("delete from `tabNote` where title = 'delete'") frappe.db.commit() server.insert_many([ diff --git a/frappe/tests/ui/test_oauth20.py b/frappe/tests/test_oauth20.py similarity index 56% rename from frappe/tests/ui/test_oauth20.py rename to frappe/tests/test_oauth20.py index cbd9ff5662..941533f2ae 100644 --- a/frappe/tests/ui/test_oauth20.py +++ b/frappe/tests/test_oauth20.py @@ -4,12 +4,10 @@ from __future__ import unicode_literals import unittest, frappe, requests, time from frappe.test_runner import make_test_records -from frappe.utils.selenium_testdriver import TestDriver from six.moves.urllib.parse import urlparse, parse_qs class TestOAuth20(unittest.TestCase): def setUp(self): - self.driver = TestDriver() make_test_records("OAuth Client") make_test_records("User") self.client_id = frappe.get_all("OAuth Client", fields=["*"])[0].get("client_id") @@ -23,41 +21,40 @@ class TestOAuth20(unittest.TestCase): frappe_login_key.base_url = "http://localhost:8000" frappe_login_key.enable_social_login = 0 frappe_login_key.save() + frappe.db.commit() def test_invalid_login(self): self.assertFalse(check_valid_openid_response()) def test_login_using_authorization_code(self): + client = frappe.get_doc("OAuth Client", self.client_id) + client.grant_type = "Authorization Code" + client.response_type = "Code" + client.save() + frappe.db.commit() - # Go to Authorize url - self.driver.get( - "api/method/frappe.integrations.oauth2.authorize?client_id=" + - self.client_id + - "&scope=all%20openid&response_type=code&redirect_uri=http%3A%2F%2Flocalhost" - ) - - time.sleep(2) + session = requests.Session() # Login - username = self.driver.find("#login_email")[0] - username.send_keys("test@example.com") + session.post( + frappe.get_site_config().host_name + "/api/method/login", + data={"usr":"test@example.com","pwd":"Eastern_43A1W"} + ) - password = self.driver.find("#login_password")[0] - password.send_keys("Eastern_43A1W") + redirect_destination = None - sign_in = self.driver.find(".btn-login")[0] - sign_in.submit() - - time.sleep(2) - - # Allow access to resource - allow = self.driver.find("#allow")[0] - allow.click() - - time.sleep(2) + # Go to Authorize url + try: + session.get( + frappe.get_site_config().host_name + "/api/method/frappe.integrations.oauth2.authorize?client_id=" + + self.client_id + + "&scope=all%20openid&response_type=code&redirect_uri=http%3A%2F%2Flocalhost" + ) + except requests.exceptions.ConnectionError as ex: + redirect_destination = ex.request.url # Get authorization code from redirected URL - auth_code = urlparse(self.driver.driver.current_url).query.split("=")[1] + auth_code = urlparse(redirect_destination).query.split("=")[1] payload = "grant_type=authorization_code&code=" payload += auth_code @@ -81,15 +78,66 @@ class TestOAuth20(unittest.TestCase): self.assertTrue(bearer_token.get("token_type") == "Bearer") self.assertTrue(check_valid_openid_response(bearer_token.get("access_token"))) + def test_revoke_token(self): + client = frappe.get_doc("OAuth Client", self.client_id) + client.grant_type = "Authorization Code" + client.response_type = "Code" + client.save() + frappe.db.commit() + + session = requests.Session() + + # Login + session.post( + frappe.get_site_config().host_name + "/api/method/login", + data={"usr":"test@example.com","pwd":"Eastern_43A1W"} + ) + + redirect_destination = None + + # Go to Authorize url + try: + session.get( + frappe.get_site_config().host_name + "/api/method/frappe.integrations.oauth2.authorize?client_id=" + + self.client_id + + "&scope=all%20openid&response_type=code&redirect_uri=http%3A%2F%2Flocalhost" + ) + except requests.exceptions.ConnectionError as ex: + redirect_destination = ex.request.url + + # Get authorization code from redirected URL + auth_code = urlparse(redirect_destination).query.split("=")[1] + + payload = "grant_type=authorization_code&code=" + payload += auth_code + payload += "&redirect_uri=http%3A%2F%2Flocalhost&client_id=" + payload += self.client_id + + headers = {'content-type':'application/x-www-form-urlencoded'} + + # Request for bearer token + token_response = requests.post( frappe.get_site_config().host_name + + "/api/method/frappe.integrations.oauth2.get_token", data=payload, headers=headers) + + # Parse bearer token json + bearer_token = token_response.json() + # Revoke Token revoke_token_response = requests.post(frappe.get_site_config().host_name + "/api/method/frappe.integrations.oauth2.revoke_token", - data="token=" + bearer_token.get("access_token")) + data="token=" + bearer_token.get("access_token"), headers=headers) + self.assertTrue(revoke_token_response.status_code == 200) # Check revoked token self.assertFalse(check_valid_openid_response(bearer_token.get("access_token"))) def test_resource_owner_password_credentials_grant(self): + client = frappe.get_doc("OAuth Client", self.client_id) + client.grant_type = "Authorization Code" + client.response_type = "Code" + client.save() + frappe.db.commit() + # Set payload payload = "grant_type=password" payload += "&username=test@example.com" @@ -117,44 +165,33 @@ class TestOAuth20(unittest.TestCase): oauth_client.save() frappe.db.commit() - # Go to Authorize url - self.driver.get( - "api/method/frappe.integrations.oauth2.authorize?client_id=" + - self.client_id + - "&scope=all%20openid&response_type=token&redirect_uri=http%3A%2F%2Flocalhost" - ) - - time.sleep(2) + session = requests.Session() # Login - username = self.driver.find("#login_email")[0] - username.send_keys("test@example.com") + session.post( + frappe.get_site_config().host_name + "/api/method/login", + data={"usr":"test@example.com","pwd":"Eastern_43A1W"} + ) - password = self.driver.find("#login_password")[0] - password.send_keys("Eastern_43A1W") + redirect_destination = None - sign_in = self.driver.find(".btn-login")[0] - sign_in.submit() + # Go to Authorize url + try: + session.get( + frappe.get_site_config().host_name + "/api/method/frappe.integrations.oauth2.authorize?client_id=" + + self.client_id + + "&scope=all%20openid&response_type=token&redirect_uri=http%3A%2F%2Flocalhost" + ) + except requests.exceptions.ConnectionError as ex: + redirect_destination = ex.request.url - time.sleep(2) - - # Allow access to resource - allow = self.driver.find("#allow")[0] - allow.click() - - time.sleep(2) - - # Get token from redirected URL - response_url = dict(parse_qs(urlparse(self.driver.driver.current_url).fragment)) + response_url = dict(parse_qs(urlparse(redirect_destination).fragment)) self.assertTrue(response_url.get("access_token")) self.assertTrue(response_url.get("expires_in")) self.assertTrue(response_url.get("scope")) self.assertTrue(response_url.get("token_type")) - self.assertTrue(check_valid_openid_response(response_url.get("access_token"))) - - def tearDown(self): - self.driver.close() + self.assertTrue(check_valid_openid_response(response_url.get("access_token")[0])) def check_valid_openid_response(access_token=None): # Returns True for valid response @@ -162,7 +199,7 @@ def check_valid_openid_response(access_token=None): # Use token in header headers = {} if access_token: - headers["Authorization"] = 'Bearer' + access_token + headers["Authorization"] = 'Bearer ' + access_token # check openid for email test@example.com openid_response = requests.get(frappe.get_site_config().host_name +