Merge pull request #948 from nofusscomputing/2025-08-14
This commit is contained in:
@ -0,0 +1,104 @@
|
||||
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'] )
|
||||
|
||||
kwargs = self.kwargs_create_item.copy()
|
||||
kwargs.update({
|
||||
'organization': api_request_permissions['tenancy']['user'],
|
||||
'model': api_request_permissions['tenancy']['user']
|
||||
})
|
||||
|
||||
change_item = model_instance(
|
||||
kwargs_create = kwargs,
|
||||
)
|
||||
|
||||
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_delete(self, model_instance, api_request_permissions):
|
||||
""" Check correct permission for delete
|
||||
|
||||
Delete item as user with delete permission
|
||||
"""
|
||||
|
||||
client = Client()
|
||||
|
||||
client.force_login( api_request_permissions['user']['delete'] )
|
||||
|
||||
kwargs = self.kwargs_create_item
|
||||
kwargs.update({
|
||||
'organization': api_request_permissions['tenancy']['user'],
|
||||
'model': api_request_permissions['tenancy']['user']
|
||||
})
|
||||
|
||||
delete_item = model_instance(
|
||||
kwargs_create = kwargs
|
||||
)
|
||||
|
||||
response = client.delete(
|
||||
path = delete_item.get_url( many = False ),
|
||||
)
|
||||
|
||||
if response.status_code == 405:
|
||||
pytest.xfail( reason = 'ViewSet does not have this request method.' )
|
||||
|
||||
assert response.status_code == 204, 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'] )
|
||||
|
||||
kwargs = self.kwargs_create_item
|
||||
kwargs.update({
|
||||
'organization': api_request_permissions['tenancy']['user'],
|
||||
'model': api_request_permissions['tenancy']['user']
|
||||
})
|
||||
|
||||
view_item = model_instance(
|
||||
kwargs_create = kwargs
|
||||
)
|
||||
|
||||
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
|
||||
|
||||
|
||||
@pytest.mark.xfail( reason = 'model is not global based')
|
||||
def test_returned_data_from_user_and_global_organizations_only(self ):
|
||||
assert False
|
@ -94,6 +94,12 @@ class ModelSerializer(
|
||||
]
|
||||
|
||||
|
||||
def validate(self, attrs):
|
||||
|
||||
attrs['created_by'] = self._context['request'].user
|
||||
|
||||
return super().validate(attrs)
|
||||
|
||||
|
||||
def is_valid(self, *, raise_exception=False) -> bool:
|
||||
|
||||
|
@ -1,4 +1,6 @@
|
||||
from django.apps import apps
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
from django.db.models.signals import (
|
||||
post_migrate,
|
||||
)
|
||||
@ -14,6 +16,31 @@ def centurion_model_migrate(sender, **kwargs):
|
||||
if sender.label != 'core':
|
||||
return
|
||||
|
||||
try:
|
||||
|
||||
print('\n\nFetching System User.\n')
|
||||
|
||||
user = apps.get_model(settings.AUTH_USER_MODEL).objects.get(
|
||||
username = 'system'
|
||||
)
|
||||
|
||||
if user.is_active:
|
||||
print(' System user is set as "Active", disabling.\n')
|
||||
user.is_active = False
|
||||
|
||||
user.save()
|
||||
|
||||
except ObjectDoesNotExist:
|
||||
|
||||
print(' System user not found, creating.\n')
|
||||
|
||||
user = apps.get_model(settings.AUTH_USER_MODEL).objects.create(
|
||||
username = 'system',
|
||||
first_name = 'System',
|
||||
last_name = 'User',
|
||||
is_active = False,
|
||||
)
|
||||
|
||||
print('\n\nCenturion Model Migration Signal.....\n')
|
||||
|
||||
models: list[ dict ] = [
|
||||
@ -261,7 +288,7 @@ def centurion_model_migrate(sender, **kwargs):
|
||||
model_name = model.get_history_model_name( model )
|
||||
)
|
||||
|
||||
history = original_history.objects.filter().exclude( user = None )
|
||||
history = original_history.objects.filter()
|
||||
|
||||
print(f' Found {len(history)} history entries to migrate.')
|
||||
|
||||
@ -269,18 +296,28 @@ def centurion_model_migrate(sender, **kwargs):
|
||||
|
||||
try:
|
||||
|
||||
after = {}
|
||||
if entry.after:
|
||||
after = entry.after
|
||||
|
||||
entry_model = entry.model
|
||||
if hasattr(entry, 'child_model'):
|
||||
entry_model = entry.child_model
|
||||
|
||||
entry_user = entry.user
|
||||
|
||||
if not entry_user:
|
||||
|
||||
entry_user = user
|
||||
|
||||
migrated_history = audit_history.objects.create(
|
||||
organization = entry.organization,
|
||||
content_type = entry.content_type,
|
||||
model = entry_model,
|
||||
before = entry.before,
|
||||
after = entry.after,
|
||||
after = after,
|
||||
action = entry.action,
|
||||
user = entry.user,
|
||||
user = entry_user,
|
||||
created = entry.created
|
||||
)
|
||||
|
||||
|
@ -0,0 +1,218 @@
|
||||
import pytest
|
||||
|
||||
from django.apps import apps
|
||||
from django.conf import settings
|
||||
from django.db import models
|
||||
from django.utils.module_loading import import_string
|
||||
|
||||
from core.models.centurion_notes import CenturionModelNote
|
||||
|
||||
from api.tests.functional.test_functional_permissions_api import (
|
||||
APIPermissionsInheritedCases
|
||||
)
|
||||
|
||||
|
||||
|
||||
def get_models( excludes: list[ str ] = [] ) -> list[ tuple ]:
|
||||
"""Fetch models from Centurion Apps
|
||||
|
||||
Args:
|
||||
excludes (list[ str ]): Words that may be in a models name to exclude
|
||||
|
||||
Returns:
|
||||
list[ tuple ]: Centurion ERP Only models
|
||||
"""
|
||||
|
||||
models: list = []
|
||||
|
||||
model_apps: list = []
|
||||
|
||||
exclude_model_apps = [
|
||||
'django',
|
||||
'django_celery_results',
|
||||
'django_filters',
|
||||
'drf_spectacular',
|
||||
'drf_spectacular_sidecar',
|
||||
'coresheaders',
|
||||
'corsheaders',
|
||||
'rest_framework',
|
||||
'rest_framework_json_api',
|
||||
'social_django',
|
||||
]
|
||||
|
||||
for app in settings.INSTALLED_APPS:
|
||||
|
||||
app = app.split('.')[0]
|
||||
|
||||
if app in exclude_model_apps:
|
||||
continue
|
||||
|
||||
model_apps += [ app ]
|
||||
|
||||
|
||||
for model in apps.get_models():
|
||||
|
||||
if model._meta.app_label not in model_apps:
|
||||
continue
|
||||
|
||||
skip = False
|
||||
|
||||
for exclude in excludes:
|
||||
|
||||
if exclude in str(model._meta.model_name):
|
||||
skip = True
|
||||
break
|
||||
|
||||
if skip:
|
||||
continue
|
||||
|
||||
models += [ model ]
|
||||
|
||||
return models
|
||||
|
||||
|
||||
|
||||
class ModelNotesMetaAPIPermissionsTestCases(
|
||||
APIPermissionsInheritedCases
|
||||
):
|
||||
"""AuditHistory Meta Model Test Cases
|
||||
|
||||
This test suite is the base for the dynamic tests that are created
|
||||
during pytest discover.
|
||||
"""
|
||||
|
||||
@pytest.fixture( scope = 'class' )
|
||||
def note_model(self, request):
|
||||
|
||||
return request.cls.note_model_class
|
||||
|
||||
|
||||
@pytest.fixture( scope = 'class', autouse = True)
|
||||
def model_kwargs(self, django_db_blocker,
|
||||
request, note_model, kwargs_centurionmodelnotemeta
|
||||
):
|
||||
|
||||
model_kwargs = kwargs_centurionmodelnotemeta.copy()
|
||||
|
||||
with django_db_blocker.unblock():
|
||||
|
||||
note_model_kwargs = request.getfixturevalue('kwargs_' + note_model._meta.model_name)
|
||||
|
||||
kwargs = {}
|
||||
|
||||
many_field = {}
|
||||
|
||||
for field, value in note_model_kwargs.items():
|
||||
|
||||
if not hasattr(getattr(note_model, field), 'field'):
|
||||
continue
|
||||
|
||||
if isinstance(getattr(note_model, field).field, models.ManyToManyField):
|
||||
|
||||
if field in many_field:
|
||||
|
||||
many_field[field] += [ value ]
|
||||
|
||||
elif isinstance(value, list):
|
||||
|
||||
value_list = []
|
||||
|
||||
for list_value in value:
|
||||
|
||||
value_list += [ list_value ]
|
||||
|
||||
|
||||
value = value_list
|
||||
|
||||
else:
|
||||
|
||||
many_field.update({
|
||||
field: [
|
||||
value
|
||||
]
|
||||
})
|
||||
|
||||
continue
|
||||
|
||||
kwargs.update({
|
||||
field: value
|
||||
})
|
||||
|
||||
|
||||
model = note_model.objects.create(
|
||||
**kwargs
|
||||
)
|
||||
|
||||
|
||||
for field, values in many_field.items():
|
||||
|
||||
for value in values:
|
||||
|
||||
getattr(model, field).add( value )
|
||||
|
||||
|
||||
model_kwargs.update({
|
||||
'model': model
|
||||
})
|
||||
request.cls.kwargs_create_item = model_kwargs
|
||||
|
||||
yield model_kwargs
|
||||
|
||||
with django_db_blocker.unblock():
|
||||
|
||||
model.delete()
|
||||
|
||||
|
||||
@pytest.fixture( scope = 'class' )
|
||||
def model(self, request):
|
||||
|
||||
return request.cls.model_class
|
||||
|
||||
|
||||
@pytest.mark.skip( reason = 'ToDo: Figure out how to dynomagic add note_model instance' )
|
||||
def test_model_creation(self, model, user):
|
||||
pass
|
||||
|
||||
|
||||
|
||||
for model in get_models():
|
||||
|
||||
if(
|
||||
not issubclass(model, CenturionModelNote)
|
||||
or model == CenturionModelNote
|
||||
):
|
||||
continue
|
||||
|
||||
|
||||
cls_name: str = f"{model._meta.object_name}MetaAPIPermissionsPyTest"
|
||||
|
||||
inc_classes = (ModelNotesMetaAPIPermissionsTestCases,)
|
||||
try:
|
||||
|
||||
additional_testcases = import_string(
|
||||
model._meta.app_label + '.tests.functional.additional_meta_model_note_' +
|
||||
str( model._meta.object_name ).replace('CenturionModelNote', '').lower() + '_permissions_api.AdditionalTestCases'
|
||||
)
|
||||
|
||||
inc_classes = (additional_testcases, *inc_classes)
|
||||
|
||||
except Exception as ex:
|
||||
additional_testcases = None
|
||||
|
||||
dynamic_class = type(
|
||||
cls_name,
|
||||
inc_classes,
|
||||
{
|
||||
'note_model_class': apps.get_model(
|
||||
app_label = model._meta.app_label,
|
||||
model_name = str( model._meta.object_name ).replace('CenturionModelNote', '')
|
||||
),
|
||||
'model_class': model
|
||||
}
|
||||
)
|
||||
|
||||
dynamic_class = pytest.mark.__getattr__(
|
||||
'model_' + str(model._meta.model_name).replace('centurionmodelnote', ''))(dynamic_class)
|
||||
dynamic_class = pytest.mark.__getattr__('module_' + model._meta.app_label)(dynamic_class)
|
||||
|
||||
globals()[cls_name] = dynamic_class
|
@ -81,7 +81,7 @@ The device can also have configuration defined. this configuration is intended t
|
||||
|
||||
It's possible for a machine to be inventoried and have the report passed to the [inventory endpoint](../api.md#inventory-reports). This report will update the device within the interface and provides the option to use scheduled inventory gathering to keep the device up to date.
|
||||
|
||||
Inventory processing is conducted by a background worker. As soon as the inventory is uploaded, the inventory processing is added to the background worker queue. Further information about the background worker can be found within its [documentation](../core/index.md#background-worker)
|
||||
Inventory processing is conducted by a background worker. As soon as the inventory is uploaded, the inventory processing is added to the background worker queue. Further information about the background worker can be found within its [documentation](../core/index.md#background-worker). All inventory objects history entries will be added by a user called `System User` regardless of the user that was used to authenticated to upload the inventory.
|
||||
|
||||
!!! tip
|
||||
Inventory not uploading? review the task logs by navigating to `Settings -> Application -> Task Logs`
|
||||
|
Reference in New Issue
Block a user