Skip to content

Commit

Permalink
feat: basics of business logic for customer provisioning
Browse files Browse the repository at this point in the history 
ENT-10071
  • Loading branch information
@iloveagent57
iloveagent57 committed Feb 25, 2025
1 parent a951ee9 commit 9c6c2d3
Show file tree
Hide file tree
Showing 7 changed files with 463 additions and 19 deletions.
262 changes: 252 additions & 10 deletions enterprise_access/apps/api/v1/tests/test_provisioning_views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
Tests for the provisioning views.
"""
import uuid
from unittest import mock

import ddt
from edx_rbac.constants import ALL_ACCESS_CONTEXT
Expand Down Expand Up @@ -43,11 +44,6 @@ class TestProvisioningAuth(APITest):
{'system_wide_role': SYSTEM_ENTERPRISE_ADMIN_ROLE, 'context': ALL_ACCESS_CONTEXT},
status.HTTP_403_FORBIDDEN,
),
# Even operators can't provision
(
{'system_wide_role': SYSTEM_ENTERPRISE_OPERATOR_ROLE, 'context': ALL_ACCESS_CONTEXT},
status.HTTP_403_FORBIDDEN,
),
# No JWT based auth, no soup for you.
(
None,
Expand All @@ -66,15 +62,26 @@ def test_provisioning_create_view_forbidden(self, role_context_dict, expected_re
response = self.client.post(PROVISIONING_CREATE_ENDPOINT)
assert response.status_code == expected_response_code

def test_provisioning_create_allowed_for_provisioning_admins(self):
@ddt.data(
(
{'system_wide_role': SYSTEM_ENTERPRISE_OPERATOR_ROLE, 'context': ALL_ACCESS_CONTEXT},
status.HTTP_201_CREATED,
),
(
{'system_wide_role': SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE, 'context': ALL_ACCESS_CONTEXT},
status.HTTP_201_CREATED,
),
)
@ddt.unpack
@mock.patch('enterprise_access.apps.api.v1.views.provisioning.provisioning_api')
def test_provisioning_create_allowed_for_provisioning_admins(
self, role_context_dict, expected_response_code, mock_provisioning_api,
):
"""
Tests that we get expected 200 response for the provisioning create view when
the requesting user has the correct system role and provides a valid request payload.
"""
self.set_jwt_cookie([{
'system_wide_role': SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE,
'context': ALL_ACCESS_CONTEXT,
}])
self.set_jwt_cookie([role_context_dict])

request_payload = {
"enterprise_customer": {
Expand All @@ -89,4 +96,239 @@ def test_provisioning_create_allowed_for_provisioning_admins(self):
],
}
response = self.client.post(PROVISIONING_CREATE_ENDPOINT, data=request_payload)
assert response.status_code == expected_response_code

mock_provisioning_api.get_or_create_enterprise_customer.assert_called_once_with(
**request_payload['enterprise_customer'],
)

created_customer = mock_provisioning_api.get_or_create_enterprise_customer.return_value
mock_provisioning_api.get_or_create_enterprise_admin_users.assert_called_once_with(
enterprise_customer_uuid=created_customer['uuid'],
user_emails=['[email protected]'],
)


@ddt.ddt
class TestProvisioningEndToEnd(APITest):
"""
Tests end-to-end calls to provisioning endpoints through mocked-out calls
to downstream services.
"""
def setUp(self):
super().setUp()
self.set_jwt_cookie([
{
'system_wide_role': SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE,
'context': ALL_ACCESS_CONTEXT,
},
])

@ddt.data(
# Data representing the state where a net-new customer is created.
{
'existing_customer_data': None,
'created_customer_data': {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
'uuid': TEST_ENTERPRISE_UUID,
},
'expected_get_customer_kwargs': {
'enterprise_customer_slug': 'test-customer',
},
'create_customer_called': True,
'expected_create_customer_kwargs': {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
},
},
# Data representing the state where a customer with the given slug exists.
{
'existing_customer_data': {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
'uuid': TEST_ENTERPRISE_UUID,
},
'created_customer_data': None,
'expected_get_customer_kwargs': {
'enterprise_customer_slug': 'test-customer',
},
'create_customer_called': False,
'expected_create_customer_kwargs': None
},
)
@mock.patch('enterprise_access.apps.provisioning.api.LmsApiClient')
def test_get_or_create_customer_and_admins_created(self, test_data, mock_lms_api_client):
"""
Tests cases where admins don't exist and customer is fetched or created.
"""
mock_client = mock_lms_api_client.return_value
mock_client.get_enterprise_customer_data.return_value = test_data['existing_customer_data']
mock_client.get_enterprise_admin_users.return_value = []
mock_client.get_enterprise_pending_admin_users.return_value = []

if test_data['created_customer_data']:
mock_client.create_enterprise_customer.return_value = test_data['created_customer_data']

mock_client.create_enterprise_admin_user.side_effect = [
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
]

request_payload = {
"enterprise_customer": {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
},
"pending_admins": [
{"user_email": "[email protected]"},
{"user_email": "[email protected]"},
],
}
response = self.client.post(PROVISIONING_CREATE_ENDPOINT, data=request_payload)
assert response.status_code == status.HTTP_201_CREATED

mock_client.get_enterprise_customer_data.assert_called_once_with(
**test_data['expected_get_customer_kwargs'],
)
if test_data['create_customer_called']:
mock_client.create_enterprise_customer.assert_called_once_with(
**test_data['expected_create_customer_kwargs'],
)
else:
self.assertFalse(mock_client.create_enterprise_customer.called)

mock_client.get_enterprise_admin_users.assert_called_once_with(TEST_ENTERPRISE_UUID)
mock_client.get_enterprise_pending_admin_users.assert_called_once_with(TEST_ENTERPRISE_UUID)
mock_client.create_enterprise_admin_user.assert_has_calls([
mock.call(TEST_ENTERPRISE_UUID, '[email protected]'),
mock.call(TEST_ENTERPRISE_UUID, '[email protected]'),
], any_order=True)

@ddt.data(
# No admin users exist, two pending admins created.
{
'existing_admin_users': [],
'existing_pending_admin_users': [],
'create_pending_admins_called': True,
'create_admin_user_side_effect': [
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'expected_create_pending_admin_calls': [
mock.call(TEST_ENTERPRISE_UUID, '[email protected]'),
mock.call(TEST_ENTERPRISE_UUID, '[email protected]'),
],
},
# One pending admin exists, one new one created.
{
'existing_admin_users': [],
'existing_pending_admin_users': [
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'create_pending_admins_called': True,
'create_admin_user_side_effect': [
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'expected_create_pending_admin_calls': [
mock.call(TEST_ENTERPRISE_UUID, '[email protected]'),
],
},
# One full admin exists, one new pending admin created.
{
'existing_admin_users': [
{'email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'existing_pending_admin_users': [],
'create_pending_admins_called': True,
'create_admin_user_side_effect': [
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'expected_create_pending_admin_calls': [
mock.call(TEST_ENTERPRISE_UUID, '[email protected]'),
],
},
# One full admin exists, one pending exists, none created.
{
'existing_admin_users': [
{'email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'existing_pending_admin_users': [
{'user_email': '[email protected]', 'enterprise_customer_uuid': TEST_ENTERPRISE_UUID},
],
'create_pending_admins_called': False,
'create_admin_user_side_effect': None,
'expected_create_pending_admin_calls': [],
},
)
@mock.patch('enterprise_access.apps.provisioning.api.LmsApiClient')
def test_customer_fetched_admins_fetched_or_created(self, test_data, mock_lms_api_client):
"""
Tests cases where [pending]admins are fetched or created, but the customer
already exists
"""
mock_client = mock_lms_api_client.return_value
mock_client.get_enterprise_customer_data.return_value = {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
'uuid': TEST_ENTERPRISE_UUID,
}
mock_client.get_enterprise_admin_users.return_value = test_data['existing_admin_users']
mock_client.get_enterprise_pending_admin_users.return_value = test_data['existing_pending_admin_users']
mock_client.create_enterprise_admin_user.side_effect = test_data['create_admin_user_side_effect']

request_payload = {
"enterprise_customer": {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
},
"pending_admins": [
{"user_email": "[email protected]"},
{"user_email": "[email protected]"},
],
}
response = self.client.post(PROVISIONING_CREATE_ENDPOINT, data=request_payload)
assert response.status_code == status.HTTP_201_CREATED
expected_response_payload = {
"enterprise_customer": {
'name': 'Test Customer',
'slug': 'test-customer',
'country': 'US',
'uuid': str(TEST_ENTERPRISE_UUID),
},
"pending_admins": [
{"user_email": "[email protected]"},
{"user_email": "[email protected]"},
],
}
actual_response_payload = response.json()
self.assertEqual(
actual_response_payload['enterprise_customer'],
expected_response_payload['enterprise_customer'],
)
self.assertCountEqual(
actual_response_payload['pending_admins'],
expected_response_payload['pending_admins'],
)
self.assertEqual(actual_response_payload.keys(), expected_response_payload.keys())

mock_client.get_enterprise_customer_data.assert_called_once_with(
enterprise_customer_slug='test-customer',
)
self.assertFalse(mock_client.create_enterprise_customer.called)

mock_client.get_enterprise_admin_users.assert_called_once_with(TEST_ENTERPRISE_UUID)
mock_client.get_enterprise_pending_admin_users.assert_called_once_with(TEST_ENTERPRISE_UUID)
if test_data['create_pending_admins_called']:
mock_client.create_enterprise_admin_user.assert_has_calls(
test_data['expected_create_pending_admin_calls'],
any_order=True,
)
else:
self.assertFalse(mock_client.create_enterprise_admin_user.called)
52 changes: 50 additions & 2 deletions enterprise_access/apps/api/v1/views/provisioning.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,20 +3,31 @@
"""
import logging

import requests
from drf_spectacular.utils import extend_schema
from edx_rbac.mixins import PermissionRequiredMixin
from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
from rest_framework import generics, permissions, status
from rest_framework import exceptions, generics, permissions, status
from rest_framework.response import Response

from enterprise_access.apps.api import serializers
from enterprise_access.apps.core import constants
from enterprise_access.apps.provisioning import api as provisioning_api

logger = logging.getLogger(__name__)

PROVISIONING_API_TAG = 'Provisioning'


class ProvisioningException(exceptions.APIException):
"""
General provisioning-related API exception.
"""
status_code = 422
default_detail = 'Could not execute this provisioning request'
default_code = 'provisioning_error'


@extend_schema(
tags=[PROVISIONING_API_TAG],
summary='Create a new provisioning request.',
Expand All @@ -37,4 +48,41 @@ class ProvisioningCreateView(PermissionRequiredMixin, generics.CreateAPIView):
def create(self, request, *args, **kwargs):
request_serializer = serializers.ProvisioningRequestSerializer(data=request.data)
request_serializer.is_valid(raise_exception=True)
return Response('ack', status=status.HTTP_201_CREATED)

customer_request_data = request_serializer.validated_data['enterprise_customer']
try:
created_customer = provisioning_api.get_or_create_enterprise_customer(
name=customer_request_data['name'],
country=customer_request_data['country'],
slug=customer_request_data['slug'],
)
except requests.exceptions.HTTPError as exc:
raise ProvisioningException(
detail=f'Error get/creating customer record: {exc}',
code='customer_provisioning_error',
) from exc

admin_emails = [
record.get('user_email')
for record in request_serializer.validated_data['pending_admins']
]

try:
customer_admin_emails = provisioning_api.get_or_create_enterprise_admin_users(
enterprise_customer_uuid=created_customer['uuid'],
user_emails=admin_emails,
)
except requests.exceptions.HTTPError as exc:
raise ProvisioningException(
detail=f'Error get/creating admin records: {exc}',
code='admin_provisioning_error',
) from exc

response_serializer = serializers.ProvisioningResponseSerializer({
'enterprise_customer': created_customer,
'pending_admins': [{'user_email': email} for email in customer_admin_emails],
})
return Response(
response_serializer.data,
status=status.HTTP_201_CREATED,
)
Loading

0 comments on commit 9c6c2d3

Please sign in to comment.