diff --git a/conntrackt/tests/test_views.py b/conntrackt/tests/test_views.py --- a/conntrackt/tests/test_views.py +++ b/conntrackt/tests/test_views.py @@ -12,7 +12,7 @@ from django.test import RequestFactory from django.test import TestCase # Application imports -from conntrackt.models import Project, Location, Entity, Interface +from conntrackt.models import Project, Location, Entity, Interface, Communication from conntrackt.views import IndexView from conntrackt.views import entity_iptables, project_iptables @@ -21,6 +21,7 @@ from conntrackt.views import ProjectView from conntrackt.views import LocationCreateView, LocationUpdateView, LocationDeleteView from conntrackt.views import EntityView, EntityCreateView, EntityUpdateView, EntityDeleteView from conntrackt.views import InterfaceCreateView, InterfaceUpdateView, InterfaceDeleteView +from conntrackt.views import CommunicationCreateView, CommunicationUpdateView, CommunicationDeleteView from helpers import PermissionTestMixin, create_get_request, generate_get_response, FakeMessages @@ -957,3 +958,483 @@ class InterfaceDeleteViewTest(Permission response = view(request, pk=1) self.assertEqual(response["Location"], reverse("entity", args=(1,))) + + +class CommunicationCreateViewTest(PermissionTestMixin, TestCase): + + view_class = CommunicationCreateView + sufficient_permissions = ("add_communication",) + + def setUp(self): + """ + Sets-up some data necessary for testing. + """ + + # Set-up some data for testing. + project1 = Project.objects.create(name="Test Project 1", description="This is test project 1.") + project2 = Project.objects.create(name="Test Project 2", description="This is test project 2.") + location = Location.objects.create(name="Test Location", description="This is test location.") + entity1 = Entity.objects.create(name="Test Entity 1", description="This is test entity 1.", project=project1, location=location) + entity2 = Entity.objects.create(name="Test Entity 2", description="This is test entity 2.", project=project1, location=location) + entity3 = Entity.objects.create(name="Test Entity 3", description="This is test entity 3.", project=project2, location=location) + Interface.objects.create(name="eth0", description="Main interface", entity=entity1, address="192.168.1.1", netmask="255.255.255.255") + Interface.objects.create(name="eth0", description="Main interface", entity=entity2, address="192.168.1.2", netmask="255.255.255.255") + Interface.objects.create(name="eth0", description="Main interface", entity=entity3, address="192.168.1.3", netmask="255.255.255.255") + + def test_interface_limit_from_entity(self): + """ + Tests if the queryset is properly limitted if GET parameter is passed. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?from_entity=1") + view.object = None + + # Get the form. + form = view.get_form(view.get_form_class()) + + # Set-up expected interfaces. + expected_interfaces = ["", + ""] + + self.assertQuerysetEqual(form.fields["source"].queryset, expected_interfaces) + self.assertQuerysetEqual(form.fields["destination"].queryset, expected_interfaces) + + def test_interface_limit_to_entity(self): + """ + Tests if the queryset is properly limitted if GET parameter is passed. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?to_entity=1") + view.object = None + + # Get the form. + form = view.get_form(view.get_form_class()) + + # Set-up expected interfaces. + expected_interfaces = ["", + ""] + + self.assertQuerysetEqual(form.fields["source"].queryset, expected_interfaces) + self.assertQuerysetEqual(form.fields["destination"].queryset, expected_interfaces) + + def test_interface_limit_project(self): + """ + Tests if the queryset is properly limitted if GET parameter is passed. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?project=1") + view.object = None + + # Get the form. + form = view.get_form(view.get_form_class()) + + # Set-up expected interfaces. + expected_interfaces = ["", + ""] + + self.assertQuerysetEqual(form.fields["source"].queryset, expected_interfaces) + self.assertQuerysetEqual(form.fields["destination"].queryset, expected_interfaces) + + def test_initial_from_entity(self): + """ + Tests if the choice field for interface is defaulted to first interface + of entity passed as part of GET parameters. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?from_entity=1") + view.object = None + + # Get the expected interface ID. + interface = Entity.objects.get(pk=1).interface_set.all()[0] + + # Fetch the initial values. + initial = view.get_initial() + + self.assertDictContainsSubset({"source": interface.pk}, initial) + + def test_initial_to_entity(self): + """ + Tests if the choice field for interface is defaulted to first interface + of entity passed as part of GET parameters. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?to_entity=1") + view.object = None + + # Get the expected interface ID. + interface = Entity.objects.get(pk=1).interface_set.all()[0] + + # Fetch the initial value. + initial = view.get_initial() + + self.assertDictContainsSubset({"destination": interface.pk}, initial) + + def test_initial_invalid_from_entity(self): + """ + Tests if the choice fields for source and destination interfaces are not + defaulted in case invalid entity ID is passed as GET parameter. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?from_entity=10") + view.object = None + + # Get the initial values. + initial = view.get_initial() + + self.assertEqual(len(initial), 0) + + def test_initial_invalid_to_entity(self): + """ + Tests if the choice fields for source and destination interfaces are not + defaulted in case invalid entity ID is passed as GET parameter. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?to_entity=10") + view.object = None + + # Get the initial values. + initial = view.get_initial() + + self.assertEqual(len(initial), 0) + + def test_initial_invalid_project(self): + """ + Tests if the choice fields for source and destination interfaces are not + defaulted in case invalid project ID is passed as GET parameter. + """ + + # Set-up the view. + view = CommunicationCreateView() + view.request = RequestFactory().get("/fake-path?project=10") + view.object = None + + # Get the initial values. + initial = view.get_initial() + + self.assertEqual(len(initial), 0) + + def test_success_url_from_entity(self): + """ + Validate that the success URL is set properly after communication is + created if origin entity is provided. + """ + + # Get the view. + view = CommunicationCreateView.as_view() + + # Generate the request. + source = Interface.objects.get(pk=1) + destination = Interface.objects.get(pk=2) + post_data = {"source": source.pk, + "destination": destination.pk, + "protocol": "TCP", + "port": "22", + "description": "SSH."} + request = RequestFactory().post("/fake-path?from_entity=1", data=post_data) + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + # Get the response. + response = view(request) + + self.assertEqual(response["Location"], reverse("entity", args=(1,))) + self.assertEqual(response.status_code, 302) + + def test_success_url_to_entity(self): + """ + Validate that the success URL is set properly after communication is + created if destination entity is provided. + """ + + # Get the view. + view = CommunicationCreateView.as_view() + + # Generate the request. + source = Interface.objects.get(pk=1) + destination = Interface.objects.get(pk=2) + post_data = {"source": source.pk, + "destination": destination.pk, + "protocol": "TCP", + "port": "22", + "description": "SSH."} + request = RequestFactory().post("/fake-path?to_entity=2", data=post_data) + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + # Get the response. + response = view(request) + + self.assertEqual(response["Location"], reverse("entity", args=(2,))) + self.assertEqual(response.status_code, 302) + + def test_success_url_no_entity(self): + """ + Validate that the success URL is set properly after communication is + created if no entity source/destinantion is passed on. + """ + + # Get the view. + view = CommunicationCreateView.as_view() + + # Generate the request. + source = Interface.objects.get(pk=1) + destination = Interface.objects.get(pk=2) + post_data = {"source": source.pk, + "destination": destination.pk, + "protocol": "TCP", + "port": "22", + "description": "SSH."} + request = RequestFactory().post("/fake-path", data=post_data) + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + # Get the response. + response = view(request) + + self.assertEqual(response["Location"], reverse("entity", args=(1,))) + self.assertEqual(response.status_code, 302) + + +class CommunicationUpdateViewTest(PermissionTestMixin, TestCase): + + fixtures = ['test-data.json'] + + view_class = CommunicationUpdateView + sufficient_permissions = ("change_communication",) + permission_test_view_kwargs = {"pk": 1} + + def test_context(self): + """ + Verifies that the context is properly set-up. + """ + + # Get the view. + view = CommunicationUpdateView.as_view() + + # Get the response. + response = generate_get_response(view, None, pk=1) + + # Set-up expected interface. + communication = Communication.objects.get(pk=1) + + self.assertEqual(response.context_data["communication"], communication) + + def test_form_interface_limit(self): + """ + Tests if the queryset is properly limitted to specific project's + entity interfaces. + """ + + # Set-up the view. + view = CommunicationUpdateView() + view.request = RequestFactory().get("/fake-path/1") + view.object = Communication.objects.get(pk=1) + + # Get the form. + form = view.get_form(view.get_form_class()) + + expected_interfaces = ["", + "", + "", + ""] + + self.assertQuerysetEqual(form.fields["source"].queryset, expected_interfaces) + self.assertQuerysetEqual(form.fields["destination"].queryset, expected_interfaces) + + def test_success_url_from_entity(self): + """ + Validate that the success URL is set properly after update. + """ + + # Get the view. + view = CommunicationUpdateView.as_view() + + # Get the communication object. + communication = Communication.objects.get(pk=1) + + # Generate the request. + post_data = {"source": communication.source.pk, + "destination": communication.destination.pk, + "protocol": communication.protocol, + "port": communication.port} + request = RequestFactory().post("/fake-path?from_entity=1", data=post_data) + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + # Get the response. + response = view(request, pk=1) + + self.assertEqual(response["Location"], reverse("entity", args=(1,))) + self.assertEqual(response.status_code, 302) + + def test_success_url_to_entity(self): + """ + Validate that the success URL is set properly after update. + """ + + # Get the view. + view = CommunicationUpdateView.as_view() + + # Get the communication object. + communication = Communication.objects.get(pk=1) + + # Generate the request. + post_data = {"source": communication.source.pk, + "destination": communication.destination.pk, + "protocol": communication.protocol, + "port": communication.port} + request = RequestFactory().post("/fake-path?to_entity=1", data=post_data) + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + # Get the response. + response = view(request, pk=1) + + self.assertEqual(response["Location"], reverse("entity", args=(1,))) + self.assertEqual(response.status_code, 302) + + def test_success_url_no_entity(self): + """ + Validate that the success URL is set properly after update. + """ + + # Get the view. + view = CommunicationUpdateView.as_view() + + # Get the communication object. + communication = Communication.objects.get(pk=1) + + # Generate the request. + post_data = {"source": communication.source.pk, + "destination": communication.destination.pk, + "protocol": communication.protocol, + "port": communication.port} + request = RequestFactory().post("/fake-path/", data=post_data) + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + # Get the response. + response = view(request, pk=1) + + self.assertEqual(response["Location"], reverse("entity", args=(communication.source.entity.pk,))) + self.assertEqual(response.status_code, 302) + + +class CommunicationDeleteViewTest(PermissionTestMixin, TestCase): + + fixtures = ['test-data.json'] + + view_class = CommunicationDeleteView + sufficient_permissions = ("delete_communication",) + permission_test_view_kwargs = {"pk": 1} + + def test_context(self): + """ + Verifies that the context is properly set-up when the view is called for + specific communication. + """ + + # Get the expected entity. + communication = Communication.objects.get(pk=1) + + # Get the view. + view = CommunicationDeleteView.as_view() + + # Get the response. + response = generate_get_response(view, None, pk=1) + + self.assertEqual(response.context_data["communication"], communication) + + def test_message(self): + """ + Tests if the message gets added when the communication is deleted. + """ + + # Get the view. + view = CommunicationDeleteView.as_view() + + # Generate the request. + request = RequestFactory().post("/fake-path/") + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + + request._messages = FakeMessages() + + # Get the response. + response = view(request, pk=1) + + self.assertIn("Communication Test Entity 2 -> Test Entity 1 (TCP:22) has been removed.", request._messages.messages) + + def test_success_url_from_entity(self): + """ + Validate that the success URL is set properly after delete. + """ + + # Get the view. + view = CommunicationDeleteView.as_view() + + # Generate the request + request = RequestFactory().post("/fake-path?from_entity=1") + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + request._messages = FakeMessages() + + # Get the response. + response = view(request, pk=1) + + self.assertEqual(response["Location"], reverse("entity", args=(1,))) + + def test_success_url_to_entity(self): + """ + Validate that the success URL is set properly after delete. + """ + + # Get the view. + view = CommunicationDeleteView.as_view() + + # Generate the request + request = RequestFactory().post("/fake-path?to_entity=1") + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + request._messages = FakeMessages() + + # Get the response. + response = view(request, pk=1) + + self.assertEqual(response["Location"], reverse("entity", args=(1,))) + + def test_success_url_no_entity(self): + """ + Validate that the success URL is set properly after delete. + """ + + # Get the view. + view = CommunicationDeleteView.as_view() + + # Get the communication object. + communication = Communication.objects.get(pk=1) + + # Generate the request + request = RequestFactory().post("/fake-path") + request.user = mock.Mock() + request._dont_enforce_csrf_checks = True + request._messages = FakeMessages() + + # Get the response. + response = view(request, pk=1) + + self.assertEqual(response["Location"], reverse("entity", args=(communication.source.entity.pk,)))