Merge pull request #849 from nofusscomputing/test-model-inheritance-refactor

This commit is contained in:
Jon
2025-07-04 19:51:58 +09:30
committed by GitHub
11 changed files with 269 additions and 131 deletions

11
.vscode/launch.json vendored
View File

@ -62,6 +62,17 @@
"autoStartBrowser": false,
"program": "${workspaceFolder}/app/manage.py"
},
{
"name": "Make Migrations",
"type": "debugpy",
"request": "launch",
"args": [
"makemigrations"
],
"django": true,
"autoStartBrowser": false,
"program": "${workspaceFolder}/app/manage.py"
},
{
"name": "Migrate",
"type": "debugpy",

View File

@ -153,7 +153,7 @@ for centurion_model in get_models(
centurion_model._meta.model_name + '_permissions_api.AdditionalTestCases'
)
inc_classes += (additional_testcases,)
inc_classes = (additional_testcases, *inc_classes)
except Exception as ex:
additional_testcases = None

View File

@ -101,10 +101,14 @@ class APIPermissionAddInheritedCases:
the_model = model_instance( kwargs_create = model_kwargs )
url = the_model.get_url( many = True )
# the_model.delete()
try:
response = client.post(
path = the_model.get_url( many = True ),
path = url,
data = kwargs_api_create
)
@ -451,6 +455,13 @@ class APIPermissionViewInheritedCases:
if response.status_code == 405:
pytest.xfail( reason = 'ViewSet does not have this request method.' )
elif IsAuthenticatedOrReadOnly in response.renderer_context['view'].permission_classes:
pytest.xfail( reason = 'ViewSet is public viewable, test is N/A' )
assert response.status_code == 200
contains_different_org: bool = False
for item in response.data['results']:

View File

@ -1,5 +1,7 @@
import django
import importlib
import logging
import rest_framework
from django.utils.safestring import mark_safe
@ -59,7 +61,7 @@ class Create(
if response.data['id'] is not None:
serializer = view_serializer(
self.get_queryset().get( pk = int(response.data['id']) ),
response.data.serializer.instance,
context = {
'request': request,
'view': self,
@ -88,21 +90,12 @@ class Create(
if not isinstance(e, APIException):
response = Response(
data = {
'server_error': str(e)
},
status = 501
)
e = self._django_to_api_exception(e)
self.get_log().exception(e)
else:
response = Response(
data = e.detail,
status = e.status_code
)
response = Response(
data = e.get_full_details(),
status = e.status_code
)
return response
@ -145,21 +138,12 @@ class Destroy(
if not isinstance(e, APIException):
response = Response(
data = {
'server_error': str(e)
},
status = 501
)
e = self._django_to_api_exception(e)
self.get_log().exception(e)
else:
response = Response(
data = e.detail,
status = e.status_code
)
response = Response(
data = e.get_full_details(),
status = e.status_code
)
return response
@ -203,21 +187,12 @@ class List(
if not isinstance(e, APIException):
response = Response(
data = {
'server_error': str(e)
},
status = 501
)
e = self._django_to_api_exception(e)
self.get_log().exception(e)
else:
response = Response(
data = e.detail,
status = e.status_code
)
response = Response(
data = e.get_full_details(),
status = e.status_code
)
return response
@ -264,21 +239,12 @@ class Retrieve(
if not isinstance(e, APIException):
response = Response(
data = {
'server_error': str(e)
},
status = 501
)
e = self._django_to_api_exception(e)
self.get_log().exception(e)
else:
response = Response(
data = e.detail,
status = e.status_code
)
response = Response(
data = e.get_full_details(),
status = e.status_code
)
return response
@ -324,7 +290,7 @@ class Update(
view_serializer = getattr(serializer_module, self.get_view_serializer_name())
serializer = view_serializer(
self.queryset.get( pk = int(self.kwargs['pk']) ),
response.data.serializer.instance,
context = {
'request': request,
'view': self,
@ -345,21 +311,12 @@ class Update(
if not isinstance(e, APIException):
response = Response(
data = {
'server_error': str(e)
},
status = 501
)
e = self._django_to_api_exception(e)
self.get_log().exception(e)
else:
response = Response(
data = e.detail,
status = e.status_code
)
response = Response(
data = e.get_full_details(),
status = e.status_code
)
return response
@ -400,7 +357,7 @@ class Update(
view_serializer = getattr(serializer_module, self.get_view_serializer_name())
serializer = view_serializer(
self.queryset.get( pk = int(self.kwargs['pk']) ),
response.data.serializer.instance,
context = {
'request': request,
'view': self,
@ -421,21 +378,12 @@ class Update(
if not isinstance(e, APIException):
response = Response(
data = {
'server_error': str(e)
},
status = 501
)
e = self._django_to_api_exception(e)
self.get_log().exception(e)
else:
response = Response(
data = e.detail,
status = e.status_code
)
response = Response(
data = e.get_full_details(),
status = e.status_code
)
return response
@ -454,6 +402,51 @@ class CommonViewSet(
viewsets (class): Django Rest Framework base class.
"""
def _django_to_api_exception( self, exc ):
"""Convert Django exception to DRF Exception
Args:
exc (Django.core.exceptions.*): Django exception to convert
Raises:
rest_framework.exceptions.ValidationError: Exception to return
Returns:
None: Exception not converted
"""
rtn_exception = None
if isinstance(exc, django.core.exceptions.ObjectDoesNotExist):
exc = rest_framework.exceptions.NotFound(exc.args)
elif isinstance(exc, django.core.exceptions.PermissionDenied):
exc = rest_framework.exceptions.PermissionDenied(exc.error_dict)
elif isinstance(exc, django.core.exceptions.ValidationError):
exc = rest_framework.exceptions.ValidationError(exc.error_dict)
else:
exc = ValueError('20250704-Unknown Exception Type. Unable to convert. Please report this error as a bug.')
try:
raise exc
except Exception as e:
return e
@property
def allowed_methods(self):
"""Allowed HTTP Methods

View File

@ -0,0 +1,17 @@
import pytest
class AdditionalTestCases:
def test_returned_data_from_user_and_global_organizations_only(
self
):
"""Check items returned
Items returned from the query Must be from the users organization and
global ONLY!
"""
pytest.xfail( reason = 'model does not use global org' )

View File

@ -227,43 +227,6 @@ class ViewSet(
return self.page_layout
def get_queryset(self):
if self.queryset is not None:
return self.queryset
if self.kwargs.get('model_name', '') == 'githubrepository':
self.queryset = GitHubRepository.objects.select_related(
'git_group',
).all()
elif self.kwargs.get('model_name', '') == 'gitlabrepository':
self.queryset = GitLabRepository.objects.select_related(
'git_group',
).all()
else:
self.queryset = self.model.objects.select_related(
'git_group',
'githubrepository',
'gitlabrepository',
).all()
if 'pk' in self.kwargs:
if self.kwargs['pk']:
self.queryset = self.queryset.filter( pk = int( self.kwargs['pk'] ) )
return self.queryset
def get_return_url(self) -> str:
if 'pk' in self.kwargs:

View File

@ -1,3 +1,5 @@
import pytest
class AdditionalTestCases:
@ -19,3 +21,25 @@ class AdditionalTestCases:
('different_organization_user_forbidden',
'Permission not restricted to orgs, they are app wide'),
]
def test_returned_results_only_user_orgs(self):
"""Returned results check
Ensure that a query to the viewset endpoint does not return
items that are not part of the users organizations.
"""
pytest.xfail( reason = 'model is not org based' )
def test_returned_data_from_user_and_global_organizations_only(
self
):
"""Check items returned
Items returned from the query Must be from the users organization and
global ONLY!
"""
pytest.xfail( reason = 'model is not org based' )

View File

@ -0,0 +1,91 @@
import pytest
from django.test import Client
class AdditionalTestCases:
def test_permission_change(self, model_instance, api_request_permissions):
""" Check correct permission for change
Make change with user who has change permission
"""
client = Client()
client.force_login( api_request_permissions['user']['change'] )
change_item = model_instance(
kwargs_create = {
'organization': api_request_permissions['tenancy']['user']
},
)
change_item.id = api_request_permissions['user']['change'].id
response = client.patch(
path = change_item.get_url( many = False ),
data = self.change_data,
content_type = 'application/json'
)
if response.status_code == 405:
pytest.xfail( reason = 'ViewSet does not have this request method.' )
assert response.status_code == 200, response.content
def test_permission_view(self, model_instance, api_request_permissions):
""" Check correct permission for view
Attempt to view as user with view permission
"""
client = Client()
client.force_login( api_request_permissions['user']['view'] )
view_item = model_instance(
kwargs_create = {
'organization': api_request_permissions['tenancy']['user']
}
)
view_item.id = api_request_permissions['user']['view'].id
response = client.get(
path = view_item.get_url( many = False )
)
if response.status_code == 405:
pytest.xfail( reason = 'ViewSet does not have this request method.' )
assert response.status_code == 200, response.content
def test_returned_results_only_user_orgs(self):
"""Returned results check
Ensure that a query to the viewset endpoint does not return
items that are not part of the users organizations.
"""
pytest.xfail( reason = 'model is not org based' )
def test_returned_data_from_user_and_global_organizations_only(
self
):
"""Check items returned
Items returned from the query Must be from the users organization and
global ONLY!
"""
pytest.xfail( reason = 'model is not org based' )

View File

@ -108,7 +108,14 @@ def model_instance(django_db_blocker, model_kwarg_data, model, model_kwargs):
for model_obj in model_objs:
if model_obj._meta.abstract:
is_abstract = False
if hasattr(model_obj, '_meta'):
is_abstract = model_obj._meta.abstract
if is_abstract:
del model_obj

View File

@ -1,4 +1,5 @@
import datetime
from django.core.exceptions import ValidationError
import pytest
from django.db import models
@ -66,9 +67,21 @@ def model_kwarg_data():
if create_instance:
instance =model.objects.create(
**kwargs
)
try:
instance =model.objects.create(
**kwargs
)
except ValidationError as e:
if '__all__' in e.error_dict:
if 'unique' in e.error_dict['__all__'][0].code:
instance = model.objects.get(
**kwargs
)
for field, values in many_field.items():

View File

@ -14,7 +14,8 @@ def model_service():
@pytest.fixture( scope = 'class')
def kwargs_service(django_db_blocker,
kwargs_centurionmodel,
kwargs_device, model_device
kwargs_device, model_device,
kwargs_port, model_port,
):
random_str = str(datetime.datetime.now(tz=datetime.timezone.utc))
@ -30,9 +31,14 @@ def kwargs_service(django_db_blocker,
device = model_device.objects.create( **kwargs )
port = model_port.objects.create( **kwargs_port )
kwargs = {
**kwargs_centurionmodel.copy(),
'name': 'service_' + random_str,
'device': device,
'config_key_variable': 'svc',
'port': port,
}
yield kwargs.copy()
@ -40,3 +46,5 @@ def kwargs_service(django_db_blocker,
with django_db_blocker.unblock():
device.delete()
port.delete()