Changeset - e10572a75d58
[Not reviewed]
default
0 1 0
Branko Majic (branko) - 11 years ago 2013-07-23 21:01:40
branko@majic.rs
CONNT-6: Implemented tests for the communication views.
1 file changed with 482 insertions and 1 deletions:
0 comments (0 inline, 0 general)
conntrackt/tests/test_views.py
Show inline comments
 
@@ -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,)))
0 comments (0 inline, 0 general)