Commit 7471adb5 authored by John Anderson's avatar John Anderson
Browse files

closes #4932 - added WebRequestContext for emulating HTTP requests for ORM operations

parent 1e1e2d5f
......@@ -201,3 +201,19 @@ To delete multiple objects at once, call `delete()` on a filtered queryset. It's
!!! warning
Deletions are immediate and irreversible. Always consider the impact of deleting objects carefully before calling `delete()` on an instance or queryset.
## Change Logging and Webhooks
Note that NetBox's change logging and webhook processing features operate under the context of an HTTP request. As such, these functions do not work automatically when using the ORM directly, either through the nbshell or otherwise. A special context manager is provided to allow these features to operate under an emulated HTTP request context. This context manager must be explicitly invoked for change log entries and webhooks to be created when interacting with objects through the ORM. Here is an example using the `WebRequestContext` context manager within the nbshell:
```
>>> from dcim.models import Site
>>> from extras.context_managers import WebRequestContext
>>> user = User.objects.get(username="andersonjd")
>>> with WebRequestContext(user):
... lax = Site(name="LAX")
... lax.clean()
... lax.save()
```
A `User` object must be provided. A `WSGIRequest` may optionally be passed and one will automatically be created if not provided.
import uuid
from contextlib import contextmanager
from django.contrib.auth.models import User
from django.core.handlers.wsgi import WSGIRequest
from django.db.models.signals import m2m_changed, pre_delete, post_save
from django.test.client import RequestFactory
from extras.signals import _handle_changed_object, _handle_deleted_object
from utilities.utils import curry
......@@ -30,3 +34,45 @@ def change_logging(request):
post_save.disconnect(handle_changed_object, dispatch_uid='handle_changed_object')
m2m_changed.disconnect(handle_changed_object, dispatch_uid='handle_changed_object')
pre_delete.disconnect(handle_deleted_object, dispatch_uid='handle_deleted_object')
@contextmanager
def WebRequestContext(user, request=None):
"""
Emulate the context of an HTTP request, which provides functions like change logging and webhook processing
in response to data changes. This context manager is for use with low level utility tooling, such as the
nbshell management command. By default, when working with the Django ORM, niether change logging nor webhook
processing occur unless manually invoked and this context manager handles those functions. A User object must be
provided and a WSGIRequest request object may optionally be passed. If not provided, the request object will
be created automatically.
Example usage:
>>> from dcim.models import Site
>>> from extras.context_managers import WebRequestContext
>>> user = User.objects.get(username="andersonjd")
>>> with WebRequestContext(user):
... lax = Site(name="LAX")
... lax.clean()
... lax.save()
:param user: User object
:param request: WSGIRequest object with an optional unique `id` set (one will be set if not present)
"""
if request is None:
request = RequestFactory().request(SERVER_NAME="WebRequestContext")
if not isinstance(request, WSGIRequest):
raise TypeError("The request object must be an instance of django.core.handlers.wsgi.WSGIRequest")
if not isinstance(user, User):
raise TypeError("The user object must be an instance of django.contrib.auth.models.User")
if not hasattr(request, "id"):
request.id = uuid.uuid4()
request.user = user
with change_logging(request):
yield
import django_rq
from django.contrib.contenttypes.models import ContentType
from django.contrib.auth.models import User
from django.test import TestCase
from dcim.models import Site
from extras.choices import *
from extras.context_managers import WebRequestContext
from extras.models import ObjectChange, Webhook
class WebRequestContextTestCase(TestCase):
def setUp(self):
self.user = User.objects.create_user(
username='jacob',
email='jacob@example.com',
password='top_secret'
)
site_ct = ContentType.objects.get_for_model(Site)
DUMMY_URL = "http://localhost/"
DUMMY_SECRET = "LOOKATMEIMASECRETSTRING"
webhooks = Webhook.objects.bulk_create((
Webhook(name='Site Create Webhook', type_create=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
))
for webhook in webhooks:
webhook.content_types.set([site_ct])
self.queue = django_rq.get_queue('default')
self.queue.empty() # Begin each test with an empty queue
def test_user_object_type_error(self):
with self.assertRaises(TypeError):
with WebRequestContext("a string is not a user object"):
pass
def test_request_object_type_error(self):
class NotARequest:
pass
with self.assertRaises(TypeError):
with WebRequestContext(self.user, NotARequest()):
pass
def test_change_log_created(self):
with WebRequestContext(self.user):
site = Site(name='Test Site 1')
site.save()
site = Site.objects.get(name='Test Site 1')
oc_list = ObjectChange.objects.filter(
changed_object_type=ContentType.objects.get_for_model(Site),
changed_object_id=site.pk
).order_by('pk')
self.assertEqual(len(oc_list), 1)
self.assertEqual(oc_list[0].changed_object, site)
self.assertEqual(oc_list[0].action, ObjectChangeActionChoices.ACTION_CREATE)
def test_change_webhook_enqueued(self):
with WebRequestContext(self.user):
site = Site(name='Test Site 2')
site.save()
# Verify that a job was queued for the object creation webhook
site = Site.objects.get(name='Test Site 2')
self.assertEqual(self.queue.count, 1)
job = self.queue.jobs[0]
self.assertEqual(job.args[0], Webhook.objects.get(type_create=True))
self.assertEqual(job.args[1]['id'], site.pk)
self.assertEqual(job.args[2], 'site')
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment