GraphQL Domains Resolvers

api/queries.py

@@ -15,9 +15,12 @@
 
 from schemas.user import *
 
+from scalars.url import URL
+
 from schemas.sectors import Sectors
 from schemas.groups import Groups
 from schemas.organizations import Organizations
+from schemas.domains import Domains
 
 
 from resolvers.sectors import (
@@ -33,7 +36,6 @@
 
 )
 
-
 from resolvers.users import (
 	resolve_test_user_claims,
 	resolve_generate_otp_url,
@@ -45,6 +47,11 @@
 	resolve_get_orgs_by_group
 )
 
+from resolvers.domains import (
+	resolve_get_domain_by_id,
+	resolve_get_domain_by_domain,
+	resolve_get_domain_by_organization
+)
 
 class Query(graphene.ObjectType):
 	"""The central gathering point for all of the GraphQL queries."""
@@ -107,6 +114,24 @@ class Query(graphene.ObjectType):
 		resolver=resolve_get_orgs_by_group,
 		description="Allows the selection of organizations from a given group"
 	)
+	get_domain_by_id = graphene.List(
+		of_type=Domains,
+		id=graphene.Argument(graphene.Int, required=True),
+		resolver=resolve_get_domain_by_id,
+		description="Allows the selection of a domain from a given ID"
+	)
+	get_domain_by_domain = graphene.List(
+		of_type=Domains,
+		url=graphene.Argument(URL, required=True),
+		resolver=resolve_get_domain_by_domain,
+		description="Allows the selection of a domain from a given domain"
+	)
+	get_domain_by_organization = graphene.List(
+		of_type=Domains,
+		org=graphene.Argument(OrganizationsEnum, required=True),
+		resolver=resolve_get_domain_by_organization,
+		description="Allows the selection of domains under an organization"
+	)
 
 	generate_otp_url = graphene.String(
 		email=graphene.Argument(EmailAddress, required=True),

api/resolvers/domains.py

@@ -0,0 +1,55 @@
+from graphql import GraphQLError
+from sqlalchemy.orm import load_only
+
+from schemas.domains import (
+	Domains,
+	DomainModel
+)
+
+from schemas.organizations import (
+	Organizations,
+	OrganizationsModel
+)
+
+
+# Resolvers
+def resolve_get_domain_by_id(self, info, **kwargs):
+	"""Return a domain by its row ID"""
+	group_id = kwargs.get('id', 1)
+	query = Domains.get_query(info).filter(
+		DomainModel.id == group_id
+	)
+	if not len(query.all()):
+		raise GraphQLError("Error, Invalid ID")
+	return query.all()
+
+
+def resolve_get_domain_by_domain(self, info, **kwargs):
+	"""Return a domain  by a url"""
+	domain = kwargs.get('url')
+	query = Domains.get_query(info).filter(
+		DomainModel.domain == domain
+	)
+	if not len(query.all()):
+		raise GraphQLError("Error, domain  does not exist")
+	return query.all()
+
+
+def resolve_get_domain_by_organization(self, info, **kwargs):
+	"""Return a list of domains by by their associated organization"""
+	organization = kwargs.get('org')
+
+	organization_id = Organizations.get_query(info).filter(
+		OrganizationsModel.organization == organization
+	).options(load_only('id'))
+
+	if not len(organization_id.all()):
+		raise GraphQLError("Error, no organization associated with that enum")
+
+	query = Domains.get_query(info).filter(
+		DomainModel.organization_id == organization_id
+	)
+
+	if not len(query.all()):
+		raise GraphQLError("Error, no domains associated with that organization")
+	return query.all()

api/scalars/email_address.py

@@ -15,8 +15,10 @@
 
 
 class EmailAddress(Scalar):
-	'''A field whose value conforms to the standard internet email address format as specified in RFC822:
-	https://www.w3.org/Protocols/rfc822/.'''
+	"""
+	A field whose value conforms to the standard internet email address format as specified in RFC822:
+	https://www.w3.org/Protocols/rfc822/.
+	"""
 
 	@staticmethod
 	def serialize(value):

api/scalars/url.py

@@ -1,25 +1,47 @@
+from re import compile
 from graphene.types import Scalar
 from graphql.language import ast
 from graphql import GraphQLError
 
 from functions.error_messages import *
 
+URL_REGEX = r'[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*)'
+
+URL_REGEX_CHECK = compile(URL_REGEX)
+
 
 class URL(Scalar):
-	'''A field whose value conforms to the standard URL format as specified in RFC3986:
-	https://www.ietf.org/rfc/rfc3986.txt.'''
+	"""
+	A field whose value conforms to the standard URL format as specified in RFC3986:
+	https://www.ietf.org/rfc/rfc3986.txt.
+	"""
 
 	@staticmethod
 	def serialize(value):
-		return str(value)
+		if not isinstance(value, str):
+			raise GraphQLError(scalar_error_type("String", value))
+
+		if not URL_REGEX_CHECK.search(value):
+			raise GraphQLError(scalar_error_type("URL", value))
+
+		return value
 
 	@staticmethod
 	def parse_value(value):
-		return str(value)
+		if not isinstance(value, str):
+			raise GraphQLError(scalar_error_type("String", value))
+
+		if not URL_REGEX_CHECK.search(value):
+			raise GraphQLError(scalar_error_type("URL", value))
+
+		return value
 
 	@staticmethod
 	def parse_literal(node):
 		if not isinstance(node, ast.StringValue):
 			raise GraphQLError(scalar_error_only_types("strings", "URLs", str(ast.Type)))
 
-		return str(node.value)
+		if not URL_REGEX_CHECK.search(node.value):
+			raise GraphQLError(scalar_error_type("URL", node.value))
+
+		return node.value

api/tests/test_domains_resolver.py

@@ -0,0 +1,218 @@
+import sys
+import os
+from os.path import dirname, join, expanduser, normpath, realpath
+
+import pytest
+from graphene.test import Client
+
+from unittest import TestCase
+
+import model_enums
+model_enums._called_from_test = True
+
+from app import app
+from db import db
+from models import Sectors, Groups, Organizations, Domains
+from queries import schema
+
+
+# This is the only way I could get imports to work for unit testing.
+PACKAGE_PARENT = '..'
+SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__))))
+sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT)))
+
+
+@pytest.fixture(scope='class')
+def domain_test_resolver_db_init():
+	"""Build database for domain resolver testing"""
+	db.init_app(app)
+
+	sectors_added = False
+	groups_added = False
+	org_added = False
+	domain_added = False
+
+	with app.app_context():
+		if Sectors.query.first() is None:
+			sector = Sectors(
+				id=2,
+				zone="GC",
+				sector="GC_BF",
+				description="Banking and Finance"
+			)
+			db.session.add(sector)
+			db.session.commit()
+			sectors_added = True
+
+		if Groups.query.first() is None:
+			group = Groups(
+				id=2,
+				s_group='GC_BF',
+				description='Banking and Finance',
+				sector_id=2
+			)
+			db.session.add(group)
+			db.session.commit()
+			groups_added = True
+
+		if Organizations.query.first() is None:
+			org = Organizations(
+				id=6,
+				organization='BOC',
+				description='BOC - Bank of Canada',
+				group_id=2
+			)
+			db.session.add(org)
+			db.session.commit()
+			org_added = True
+
+		if Domains.query.first() is None:
+			domain = Domains(
+				id=15,
+				domain='bankofcanada.ca',
+				organization_id=6
+			)
+			db.session.add(domain)
+			db.session.commit()
+			domain_added = True
+
+	yield
+
+	with app.app_context():
+		if domain_added:
+			Domains.query.filter(Domains.id == 15).delete()
+			db.session.commit()
+		if org_added:
+			Organizations.query.filter(Organizations.id == 6).delete()
+			db.session.commit()
+		if groups_added:
+			Groups.query.filter(Groups.id == 2).delete()
+			db.session.commit()
+		if sectors_added:
+			Sectors.query.filter(Sectors.id == 2).delete()
+			db.session.commit()
+
+
+@pytest.mark.usefixtures('domain_test_resolver_db_init')
+class TestOrgResolver(TestCase):
+	def test_get_domain_resolvers_by_id(self):
+		"""Test get_domain_by_id resolver"""
+		with app.app_context():
+			client = Client(schema)
+			query = """
+				{
+					getDomainById(id: 15){
+						domain
+					}
+				}"""
+
+			result_refr = {
+				"data": {
+					"getDomainById": [
+						{
+							"domain": "bankofcanada.ca"
+						}
+					]
+				}
+			}
+
+			result_eval = client.execute(query)
+		self.assertDictEqual(result_refr, result_eval)
+
+	def test_get_domain_resolvers_by_domain(self):
+		""""Test get_domain_by_domain resolver"""
+		with app.app_context():
+			client = Client(schema)
+			query = """
+				{
+					getDomainByDomain(url: "bankofcanada.ca"){
+						domain
+					}
+				}"""
+
+			result_refr = {
+				"data": {
+					"getDomainByDomain": [
+						{
+							"domain": "bankofcanada.ca"
+						}
+					]
+				}
+			}
+
+			result_eval = client.execute(query)
+		self.assertDictEqual(result_refr, result_eval)
+
+	def test_get_domain_resolvers_by_org(self):
+		"""Test get_domain_by_org_enum resolver"""
+		with app.app_context():
+			client = Client(schema)
+			query = """
+				{
+					getDomainByOrganization(org: BOC){
+						domain
+					}
+				}"""
+			result_refr = {
+				"data": {
+					"getDomainByOrganization": [
+						{
+							"domain": "bankofcanada.ca"
+						}
+					]
+				}
+			}
+
+			result_eval = client.execute(query)
+		self.assertDictEqual(result_refr, result_eval)
+
+	def test_domain_resolver_by_id_invalid(self):
+		"""Test get_domain_by_id invalid ID error handling"""
+		with app.app_context():
+			client = Client(schema)
+			query = """
+				{
+					getDomainById(id: 9999){
+						domain
+					}
+				}
+			"""
+			executed = client.execute(query)
+
+		assert executed['errors']
+		assert executed['errors'][0]
+		assert executed['errors'][0]['message'] == "Error, Invalid ID"
+
+	def test_domain_resolver_by_org_invalid(self):
+		"""Test get_domain_by_domain invalid sector error handling"""
+		with app.app_context():
+			client = Client(schema)
+			query = """
+				{
+					getDomainByDomain(url: "google.ca"){
+						domain
+					}
+				}
+			"""
+			executed = client.execute(query)
+
+		assert executed['errors']
+		assert executed['errors'][0]
+		assert executed['errors'][0]['message'] == 'Error, domain  does not exist'
+
+	def test_domain_resolver_by_org_invalid(self):
+		"""Test get_domain_by_org invalid Zone error handling"""
+		with app.app_context():
+			client = Client(schema)
+			query = """
+				{
+					getDomainByOrganization(org: fds){
+						domain
+					}
+				}
+			"""
+			executed = client.execute(query)
+
+		assert executed['errors']
+		assert executed['errors'][0]
+		assert executed['errors'][0]['message'] == f'Argument "org" has invalid value fds.\nExpected type "OrganizationsEnum", found fds.'