|
@@ -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 = ["<Interface: Test Entity 1 (192.168.1.1)>",
|
|
|
"<Interface: Test Entity 2 (192.168.1.2)>"]
|
|
|
|
|
|
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 = ["<Interface: Test Entity 1 (192.168.1.1)>",
|
|
|
"<Interface: Test Entity 2 (192.168.1.2)>"]
|
|
|
|
|
|
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 = ["<Interface: Test Entity 1 (192.168.1.1)>",
|
|
|
"<Interface: Test Entity 2 (192.168.1.2)>"]
|
|
|
|
|
|
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 = ["<Interface: Test Entity 1 (192.168.1.1)>",
|
|
|
"<Interface: Test Entity 2 (192.168.1.2)>",
|
|
|
"<Interface: Test Entity 3 (192.168.1.3)>",
|
|
|
"<Interface: Test Subnet (192.168.2.0/255.255.255.0)>"]
|
|
|
|
|
|
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,)))
|