<commit_before># Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# MIT License. See license.txt
from __future__ import unicode_literals

import unittest, frappe, requests, time
from frappe.test_runner import make_test_records
from six.moves.urllib.parse import urlparse, parse_qs, urljoin

class TestOAuth20(unittest.TestCase):

	def setUp(self):
		make_test_records("OAuth Client")
		make_test_records("User")
		self.client_id = frappe.get_all("OAuth Client", fields=["*"])[0].get("client_id")
		self.form_header = {"content-type": "application/x-www-form-urlencoded"}
		self.scope = "all"
		self.redirect_uri = "http://localhost"

		# Set Frappe server URL reqired for id_token generation
		try:
			frappe_login_key = frappe.get_doc("Social Login Key", "frappe")
		except frappe.DoesNotExistError:
			frappe_login_key = frappe.new_doc("Social Login Key")

		frappe_login_key.get_social_login_provider("Frappe", initialize=True)
		frappe_login_key.base_url = frappe.utils.get_url()
		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()

		session = requests.Session()
		login(session)

		redirect_destination = None

		# Go to Authorize url
		try:
			session.get(
				get_full_url("/api/method/frappe.integrations.oauth2.authorize"),
				params={
					"client_id": self.client_id,
					"scope": self.scope,
					"response_type": "code",
					"redirect_uri": self.redirect_uri
				}
			)
		except requests.exceptions.ConnectionError as ex:
			redirect_destination = ex.request.url

		# Get authorization code from redirected URL
		query = parse_qs(urlparse(redirect_destination).query)
		auth_code = query.get("code")[0]

		# Request for bearer token
		token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.get_token"),
			headers=self.form_header,
			data={
				"grant_type": "authorization_code",
				"code": auth_code,
				"redirect_uri": self.redirect_uri,
				"client_id": self.client_id
			}
		)

		# Parse bearer token json
		bearer_token = token_response.json()

		self.assertTrue(bearer_token.get("access_token"))
		self.assertTrue(bearer_token.get("expires_in"))
		# self.assertTrue(bearer_token.get("id_token"))
		self.assertTrue(bearer_token.get("refresh_token"))
		self.assertTrue(bearer_token.get("scope"))
		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)

		redirect_destination = None

		# Go to Authorize url
		try:
			session.get(
				get_full_url("/api/method/frappe.integrations.oauth2.authorize"),
				params={
					"client_id": self.client_id,
					"scope": self.scope,
					"response_type": "code",
					"redirect_uri": self.redirect_uri
				}
			)
		except requests.exceptions.ConnectionError as ex:
			redirect_destination = ex.request.url

		# Get authorization code from redirected URL
		query = parse_qs(urlparse(redirect_destination).query)
		auth_code = query.get("code")[0]

		# Request for bearer token
		token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.get_token"),
			headers=self.form_header,
			data={
				"grant_type": "authorization_code",
				"code": auth_code,
				"redirect_uri": self.redirect_uri,
				"client_id": self.client_id
			}
		)

		# Parse bearer token json
		bearer_token = token_response.json()

		# Revoke Token
		revoke_token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.revoke_token"),
			headers=self.form_header,
			data={"token": bearer_token.get("access_token")}
		)

		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()

		# Request for bearer token
		token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.get_token"),
			headers=self.form_header,
			data={
				"grant_type": "password",
				"username": "<EMAIL>",
				"password": "<PASSWORD>",
				"client_id":  self.client_id,
				"scope": self.scope
			}
		)

		# Parse bearer token json
		bearer_token = token_response.json()

		# Check token for valid response
		self.assertTrue(check_valid_openid_response(bearer_token.get("access_token")))

	def test_login_using_implicit_token(self):
		oauth_client = frappe.get_doc("OAuth Client", self.client_id)
		oauth_client.grant_type = "Implicit"
		oauth_client.response_type = "Token"
		oauth_client.save()
		frappe.db.commit()

		session = requests.Session()
		login(session)

		redirect_destination = None

		# Go to Authorize url
		try:
			session.get(
				get_full_url("/api/method/frappe.integrations.oauth2.authorize"),
				params={
					"client_id": self.client_id,
					"scope": self.scope,
					"response_type": "token",
					"redirect_uri": self.redirect_uri
				}
			)
		except requests.exceptions.ConnectionError as ex:
			redirect_destination = ex.request.url

		response_dict = parse_qs(urlparse(redirect_destination).fragment)

		self.assertTrue(response_dict.get("access_token"))
		self.assertTrue(response_dict.get("expires_in"))
		self.assertTrue(response_dict.get("scope"))
		self.assertTrue(response_dict.get("token_type"))
		self.assertTrue(check_valid_openid_response(response_dict.get("access_token")[0]))


def check_valid_openid_response(access_token=None):
	"""Return True for valid response."""
	# Use token in header
	headers = {}
	if access_token:
		headers["Authorization"] = "Bearer " + access_token

	# check openid for email <EMAIL>
	openid_response = requests.get(
		get_full_url("/api/method/frappe.integrations.oauth2.openid_profile"),
		headers=headers
	)

	return True if openid_response.status_code == 200 else False


def get_full_url(endpoint):
	return urljoin(frappe.utils.get_url(), endpoint)


def login(session):
	session.post(
		frappe.utils.get_url() + "/api/method/login",
		data={
			"usr": "<EMAIL>",
			"pwd": "<PASSWORD>"
		}
	)<commit_msg>fix: login, get_full_url<commit_after># Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# MIT License. See license.txt
from __future__ import unicode_literals

import unittest, frappe, requests, time
from frappe.test_runner import make_test_records
from six.moves.urllib.parse import urlparse, parse_qs, urljoin

class TestOAuth20(unittest.TestCase):

	def setUp(self):
		make_test_records("OAuth Client")
		make_test_records("User")
		self.client_id = frappe.get_all("OAuth Client", fields=["*"])[0].get("client_id")
		self.form_header = {"content-type": "application/x-www-form-urlencoded"}
		self.scope = "all"
		self.redirect_uri = "http://localhost"

		# Set Frappe server URL reqired for id_token generation
		try:
			frappe_login_key = frappe.get_doc("Social Login Key", "frappe")
		except frappe.DoesNotExistError:
			frappe_login_key = frappe.new_doc("Social Login Key")

		frappe_login_key.get_social_login_provider("Frappe", initialize=True)
		frappe_login_key.base_url = frappe.utils.get_url()
		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()

		session = requests.Session()
		login(session)

		redirect_destination = None

		# Go to Authorize url
		try:
			session.get(
				get_full_url("/api/method/frappe.integrations.oauth2.authorize"),
				params={
					"client_id": self.client_id,
					"scope": self.scope,
					"response_type": "code",
					"redirect_uri": self.redirect_uri
				}
			)
		except requests.exceptions.ConnectionError as ex:
			redirect_destination = ex.request.url

		# Get authorization code from redirected URL
		query = parse_qs(urlparse(redirect_destination).query)
		auth_code = query.get("code")[0]

		# Request for bearer token
		token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.get_token"),
			headers=self.form_header,
			data={
				"grant_type": "authorization_code",
				"code": auth_code,
				"redirect_uri": self.redirect_uri,
				"client_id": self.client_id
			}
		)

		# Parse bearer token json
		bearer_token = token_response.json()

		self.assertTrue(bearer_token.get("access_token"))
		self.assertTrue(bearer_token.get("expires_in"))
		# self.assertTrue(bearer_token.get("id_token"))
		self.assertTrue(bearer_token.get("refresh_token"))
		self.assertTrue(bearer_token.get("scope"))
		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)

		redirect_destination = None

		# Go to Authorize url
		try:
			session.get(
				get_full_url("/api/method/frappe.integrations.oauth2.authorize"),
				params={
					"client_id": self.client_id,
					"scope": self.scope,
					"response_type": "code",
					"redirect_uri": self.redirect_uri
				}
			)
		except requests.exceptions.ConnectionError as ex:
			redirect_destination = ex.request.url

		# Get authorization code from redirected URL
		query = parse_qs(urlparse(redirect_destination).query)
		auth_code = query.get("code")[0]

		# Request for bearer token
		token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.get_token"),
			headers=self.form_header,
			data={
				"grant_type": "authorization_code",
				"code": auth_code,
				"redirect_uri": self.redirect_uri,
				"client_id": self.client_id
			}
		)

		# Parse bearer token json
		bearer_token = token_response.json()

		# Revoke Token
		revoke_token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.revoke_token"),
			headers=self.form_header,
			data={"token": bearer_token.get("access_token")}
		)

		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()

		# Request for bearer token
		token_response = requests.post(
			get_full_url("/api/method/frappe.integrations.oauth2.get_token"),
			headers=self.form_header,
			data={
				"grant_type": "password",
				"username": "<EMAIL>",
				"password": "<PASSWORD>",
				"client_id":  self.client_id,
				"scope": self.scope
			}
		)

		# Parse bearer token json
		bearer_token = token_response.json()

		# Check token for valid response
		self.assertTrue(check_valid_openid_response(bearer_token.get("access_token")))

	def test_login_using_implicit_token(self):
		oauth_client = frappe.get_doc("OAuth Client", self.client_id)
		oauth_client.grant_type = "Implicit"
		oauth_client.response_type = "Token"
		oauth_client.save()
		frappe.db.commit()

		session = requests.Session()
		login(session)

		redirect_destination = None

		# Go to Authorize url
		try:
			session.get(
				get_full_url("/api/method/frappe.integrations.oauth2.authorize"),
				params={
					"client_id": self.client_id,
					"scope": self.scope,
					"response_type": "token",
					"redirect_uri": self.redirect_uri
				}
			)
		except requests.exceptions.ConnectionError as ex:
			redirect_destination = ex.request.url

		response_dict = parse_qs(urlparse(redirect_destination).fragment)

		self.assertTrue(response_dict.get("access_token"))
		self.assertTrue(response_dict.get("expires_in"))
		self.assertTrue(response_dict.get("scope"))
		self.assertTrue(response_dict.get("token_type"))
		self.assertTrue(check_valid_openid_response(response_dict.get("access_token")[0]))


def check_valid_openid_response(access_token=None):
	"""Return True for valid response."""
	# Use token in header
	headers = {}
	if access_token:
		headers["Authorization"] = "Bearer " + access_token

	# check openid for email <EMAIL>
	openid_response = requests.get(
		get_full_url("/api/method/frappe.integrations.oauth2.openid_profile"),
		headers=headers
	)

	return True if openid_response.status_code == 200 else False


def login(session):
	session.post(
		get_full_url("/api/method/login"),
		data={
			"usr": "<EMAIL>",
			"pwd": "<PASSWORD>"
		}
	)


def get_full_url(endpoint):
	"""Turn '/endpoint' into 'http://127.0.0.1:8000/endpoint'."""
	return urljoin(frappe.utils.get_url(), endpoint)
{'fraction_non_alphanumeric': 0.12440014765596161, 'fraction_numerical': 0.004429678848283499, 'mean_word_length': 4.76916524701874, 'pattern_counts': {'":': 58, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '215037', 'n_tokens_mistral': 5221, 'n_tokens_neox': 4641, 'n_words': 781}