events: fix event creation with anonymous user
This commit is contained in:
parent
a81f981471
commit
ca4ead8fd8
|
@ -85,7 +85,7 @@ def sanitize_dict(source: dict[Any, Any]) -> dict[Any, Any]:
|
||||||
value = asdict(value)
|
value = asdict(value)
|
||||||
if isinstance(value, dict):
|
if isinstance(value, dict):
|
||||||
final_dict[key] = sanitize_dict(value)
|
final_dict[key] = sanitize_dict(value)
|
||||||
elif isinstance(value, User):
|
elif isinstance(value, (User, AnonymousUser)):
|
||||||
final_dict[key] = sanitize_dict(get_user(value))
|
final_dict[key] = sanitize_dict(get_user(value))
|
||||||
elif isinstance(value, models.Model):
|
elif isinstance(value, models.Model):
|
||||||
final_dict[key] = sanitize_dict(model_to_dict(value))
|
final_dict[key] = sanitize_dict(model_to_dict(value))
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
"""policy process tests"""
|
"""policy process tests"""
|
||||||
|
from django.contrib.auth.models import AnonymousUser
|
||||||
from django.core.cache import cache
|
from django.core.cache import cache
|
||||||
from django.test import RequestFactory, TestCase
|
from django.test import RequestFactory, TestCase
|
||||||
from guardian.shortcuts import get_anonymous_user
|
from guardian.shortcuts import get_anonymous_user
|
||||||
|
@ -149,6 +150,38 @@ class TestPolicyProcess(TestCase):
|
||||||
self.assertEqual(event.context["result"]["messages"], ["dummy"])
|
self.assertEqual(event.context["result"]["messages"], ["dummy"])
|
||||||
self.assertEqual(event.client_ip, "127.0.0.1")
|
self.assertEqual(event.client_ip, "127.0.0.1")
|
||||||
|
|
||||||
|
def test_execution_logging_anonymous(self):
|
||||||
|
"""Test policy execution creates event with anonymous user"""
|
||||||
|
policy = DummyPolicy.objects.create(
|
||||||
|
result=False, wait_min=0, wait_max=1, execution_logging=True
|
||||||
|
)
|
||||||
|
binding = PolicyBinding(
|
||||||
|
policy=policy, target=Application.objects.create(name="test")
|
||||||
|
)
|
||||||
|
|
||||||
|
user = AnonymousUser()
|
||||||
|
|
||||||
|
http_request = self.factory.get("/")
|
||||||
|
http_request.user = user
|
||||||
|
|
||||||
|
request = PolicyRequest(user)
|
||||||
|
request.set_http_request(http_request)
|
||||||
|
response = PolicyProcess(binding, request, None).execute()
|
||||||
|
self.assertEqual(response.passing, False)
|
||||||
|
self.assertEqual(response.messages, ("dummy",))
|
||||||
|
|
||||||
|
events = Event.objects.filter(
|
||||||
|
action=EventAction.POLICY_EXECUTION,
|
||||||
|
context__policy_uuid=policy.policy_uuid.hex,
|
||||||
|
)
|
||||||
|
self.assertTrue(events.exists())
|
||||||
|
self.assertEqual(len(events), 1)
|
||||||
|
event = events.first()
|
||||||
|
self.assertEqual(event.user["username"], "AnonymousUser")
|
||||||
|
self.assertEqual(event.context["result"]["passing"], False)
|
||||||
|
self.assertEqual(event.context["result"]["messages"], ["dummy"])
|
||||||
|
self.assertEqual(event.client_ip, "127.0.0.1")
|
||||||
|
|
||||||
def test_raises(self):
|
def test_raises(self):
|
||||||
"""Test policy that raises error"""
|
"""Test policy that raises error"""
|
||||||
policy_raises = ExpressionPolicy.objects.create(
|
policy_raises = ExpressionPolicy.objects.create(
|
||||||
|
|
Reference in New Issue