Changeset - a69e9814cf16
[Not reviewed]
default
1 3 1
Branko Majic (branko) - 11 years ago 2013-09-19 13:10:22
branko@majic.rs
CONNT-18: Switched to using factory_boy and dynamic data generation instead of using fixtures for tests.
5 files changed with 334 insertions and 283 deletions:
0 comments (0 inline, 0 general)
conntrackt/fixtures/test-data.json
Show inline comments
 
deleted file
conntrackt/tests/factories.py
Show inline comments
 
new file 100644
 
# Third-party application imports.
 
import factory
 

	
 
# Application imports.
 
from conntrackt.models import Project, Location, Entity, Interface, Communication
 

	
 

	
 
class ProjectFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing projects where name is set to "Test Project N", and N
 
    is an increasing sequence. Description is based on name.
 
    """
 

	
 
    FACTORY_FOR = Project
 

	
 
    @factory.sequence
 
    def name(n):
 
        return "Test Project %d" % n
 

	
 
    description = factory.LazyAttribute(lambda d: "This is %s." % d.name)
 

	
 

	
 
class LocationFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing locations where name is set to "Test Location N", and
 
    N is an increasing sequence. Description is based on name.
 
    """
 

	
 
    FACTORY_FOR = Location
 

	
 
    @factory.sequence
 
    def name(n):
 
        return "Test Location %d" % n
 

	
 
    description = factory.LazyAttribute(lambda d: "This is %s." % d.name)
 

	
 

	
 
class ServerInterfaceFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing server interfaces. Interface IP address is set to
 
    192.168.1.N, where N is a sequence. The netmask is always set to
 
    "255.255.255.255". The interface name is always set to eth0.
 
    """
 

	
 
    FACTORY_FOR = Interface
 
    netmask = "255.255.255.255"
 
    name = "eth0"
 

	
 
    @factory.sequence
 
    def address(n):
 
        return "192.168.1.%d" % (n)
 

	
 

	
 
class SubnetInterfaceFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing subnet "interfaces". Interface IP address is set to
 
    10.10.N.0, where N is a sequence. Interface name is always set to
 
    switch0. Netmask is set to "255.255.255.0".
 
    """
 

	
 
    FACTORY_FOR = Interface
 
    netmask = "255.255.255.0"
 
    name = "switch0"
 

	
 
    @factory.sequence
 
    def address(n):
 
        return "10.10.%d.0" % (n)
 

	
 

	
 
class ServerEntityFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing server entities where name is set to "Test Entity N",
 
    and N is an increasing sequence.
 
    """
 

	
 
    FACTORY_FOR = Entity
 

	
 
    interface = factory.RelatedFactory(ServerInterfaceFactory, "entity")
 

	
 
    @factory.sequence
 
    def name(n):
 
        return "Test Entity %d" % n
 

	
 
    description = factory.LazyAttribute(lambda d: "This is %s." % d.name)
 

	
 

	
 
class SubnetEntityFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing subnet entities where name is set to "Test Subnet N",
 
    and N is an increasing sequence.
 
    """
 

	
 
    FACTORY_FOR = Entity
 

	
 
    interface = factory.RelatedFactory(SubnetInterfaceFactory, "entity")
 

	
 
    @factory.sequence
 
    def name(n):
 
        return "Test Subnet %d" % n
 

	
 
    description = factory.LazyAttribute(lambda d: "This is %s." % d.name)
 

	
 

	
 
class CommunicationFactory(factory.django.DjangoModelFactory):
 
    """
 
    Factory for producing communications. The descriptin of communication will
 
    be based on protocol and port.
 
    """
 

	
 
    FACTORY_FOR = Communication
 

	
 
    description = factory.LazyAttribute(lambda d: "Communicate over %s:%s" % (d.protocol, d.port))
 

	
 

	
 
def setup_test_data():
 
    """
 
    Sets-up some test data for testing more complex functionality.
 
    """
 

	
 
    for factory in ProjectFactory, LocationFactory, ServerEntityFactory, SubnetEntityFactory, ServerInterfaceFactory, SubnetInterfaceFactory, CommunicationFactory:
 
        factory.reset_sequence()
 

	
 
    project1 = ProjectFactory(pk=1)
 
    project2 = ProjectFactory(pk=2)
 

	
 
    location1 = LocationFactory(pk=1)
 
    location2 = LocationFactory(pk=2)
 

	
 
    entity1 = ServerEntityFactory(pk=1, project=project1, location=location1)
 
    entity2 = ServerEntityFactory(pk=2, project=project1, location=location1)
 
    entity3 = ServerEntityFactory(pk=3, project=project1, location=location2)
 
    entity4 = SubnetEntityFactory(pk=4, project=project1, location=location2)
 
    entity5 = ServerEntityFactory(pk=5, project=project2, location=location1)
 

	
 
    communication1 = CommunicationFactory(pk=1, source_id=2, destination_id=1, protocol="TCP", port="22")
 
    communication2 = CommunicationFactory(pk=2, source_id=2, destination_id=1, protocol="ICMP", port="8")
 
    communication3 = CommunicationFactory(pk=3, source_id=3, destination_id=1, protocol="TCP", port="3306")
 
    communication4 = CommunicationFactory(pk=4, source_id=1, destination_id=3, protocol="UDP", port="53")
 
    communication5 = CommunicationFactory(pk=5, source_id=4, destination_id=1, protocol="TCP", port="22")
 
    communication6 = CommunicationFactory(pk=6, source_id=1, destination_id=2, protocol="UDP", port="123")
conntrackt/tests/test_models.py
Show inline comments
 
# Django imports.
 
from django.core.exceptions import ValidationError
 
from django.db import IntegrityError
 
from django.test import TestCase
 

	
 
# Application imports.
 
from conntrackt.models import Project, Location, Entity, Interface, Communication
 

	
 
# Test imports.
 
from .factories import ProjectFactory, LocationFactory
 
from .factories import ServerEntityFactory, ServerInterfaceFactory
 
from .factories import SubnetEntityFactory, SubnetInterfaceFactory
 
from .factories import CommunicationFactory
 
from .factories import setup_test_data
 

	
 

	
 
class ProjectTest(TestCase):
 

	
 
    def test_unique_name(self):
 
        """
 
        Test if unique project name is enforced.
 
        """
 

	
 
        project1 = Project(name="Test Project", description="This is a test project.")
 
        project1.save()
 
        project = ProjectFactory()
 

	
 
        project2 = Project(name="Test Project", description="This is a test project.")
 
        self.assertRaises(IntegrityError, project2.save)
 
        self.assertRaises(IntegrityError, ProjectFactory, name=project.name)
 

	
 
    def test_representation(self):
 
        """
 
        Test the representation of project.
 
        """
 

	
 
        project = Project(name="Test Project", description="This is a test project.")
 
        project = ProjectFactory(name="Test Project")
 

	
 
        self.assertEqual(str(project), "Test Project")
 

	
 
    def test_absolute_url(self):
 
        """
 
        Tests if the absolute URL is generated properly.
 
        """
 

	
 
        project = Project(pk=1, name="Test Project", description="This is a test project.")
 
        project = ProjectFactory(pk=1)
 

	
 
        self.assertEqual(project.get_absolute_url(), "/conntrackt/project/1/")
 

	
 

	
 
class LocationTest(TestCase):
 

	
 
    def test_unique_name(self):
 
        """
 
        Test if unique locationn name is enforced.
 
        Test if unique location name is enforced.
 
        """
 

	
 
        location1 = Location(name="Test Location", description="This is a test location.")
 
        location1.save()
 
        location = LocationFactory()
 

	
 
        location2 = Location(name="Test Location", description="This is a test location.")
 
        self.assertRaises(IntegrityError, location2.save)
 
        self.assertRaises(IntegrityError, LocationFactory, name=location.name)
 

	
 
    def test_representation(self):
 
        """
 
        Test the representation of location.
 
        """
 

	
 
        project = Location(name="Test Location", description="This is a test location.")
 
        project = LocationFactory(name="Test Location")
 

	
 
        self.assertEqual(str(project), "Test Location")
 

	
 

	
 
class EntityTest(TestCase):
 
    fixtures = ['test-data.json']
 

	
 
    def setUp(cls):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_incoming_communications(self):
 
        """
 
        Test that we get correct list of incoming connections with the sample
 
        data.
 
        """
 

	
 
        entity = Entity.objects.get(name="Test Entity 1")
 
        entity = Entity.objects.get(pk=1)
 
        incoming = Communication.objects.filter(pk__in=(1, 2, 3, 5))
 

	
 
        self.assertItemsEqual(entity.incoming_communications(), incoming)
 

	
 
    def test_outgoing_communications(self):
 
        """
 
        Test that we get correct list of outgoing connections with the sample
 
        data.
 
        """
 

	
 
        entity = Entity.objects.get(name="Test Entity 1")
 
        entity = Entity.objects.get(pk=1)
 
        outgoing = Communication.objects.filter(pk__in=(4, 6))
 

	
 
        self.assertItemsEqual(entity.outgoing_communications(), outgoing)
 

	
 
    def test_representation(self):
 
        """
 
        Test the representation of entity.
 
        """
 

	
 
        ent = Entity.objects.get(name="Test Entity 1")
 
        ent = Entity.objects.get(pk=1)
 
        representation = "Test Entity 1 (Test Project 1 - Test Location 1)"
 

	
 
        self.assertEqual(str(ent), representation)
 

	
 
    def test_unique_name(self):
 
        """
 
        Test if unique entity name is enforced across same project.
 
        """
 

	
 
        entity1 = Entity.objects.get(pk=1)
 

	
 
        entity2 = Entity(name=entity1.name, description="Duplicate entity.", project=entity1.project, location=entity1.location)
 
        entity_dup = Entity(name=entity1.name, description="Duplicate entity.", project=entity1.project, location=entity1.location)
 

	
 
        self.assertRaises(IntegrityError, entity2.save)
 
        self.assertRaises(IntegrityError, entity_dup.save)
 

	
 
    def test_absolute_url(self):
 
        """
 
        Tests if the absolute URL is generated properly.
 
        """
 

	
 
        entity = Entity.objects.get(pk=1)
 

	
 
        self.assertEqual(entity.get_absolute_url(), "/conntrackt/entity/1/")
 

	
 
    def test_project_move_constraints(self):
 
        """
 
        Tests if entity is prevented from being moved to different project in
 
        case of existing communications.
 
        """
 

	
 
        entity = Entity.objects.get(pk=1)
 
        new_project = Project.objects.get(pk=2)
 

	
 
        entity.project = new_project
 
        self.assertRaisesRegexp(ValidationError, "The entity cannot be moved to different project as long as it has valid communications with entities in current project.", entity.clean)
 

	
 

	
 
class InterfaceTest(TestCase):
 
    fixtures = ['test-data.json']
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_unique_name(self):
 
        """
 
        Test if unique interface name is enforced across same entity.
 
        """
 

	
 
        entity = Entity.objects.get(pk=1)
 

	
 
        interface = entity.interface_set.get(pk=1)
 

	
 
        duplicate = Interface(name=interface.name, description="Duplicate interface.", entity=entity, address="10.10.10.10", netmask="255.255.255.255")
 

	
 
@@ -165,78 +180,84 @@ class InterfaceTest(TestCase):
 
        """
 

	
 
        interface = Entity.objects.get(name="Test Entity 1").interface_set.get(name="eth0")
 
        representation = "Test Entity 1 (192.168.1.1)"
 

	
 
        self.assertEqual(str(interface), representation)
 

	
 
    def test_representation_subnet(self):
 
        """
 
        Test representation of subnet.
 
        """
 

	
 
        interface = Entity.objects.get(name="Test Subnet").interface_set.get(name="eth0")
 
        representation = "Test Subnet (192.168.2.0/255.255.255.0)"
 
        interface = Entity.objects.get(pk=4).interface_set.get(name="switch0")
 
        representation = "Test Subnet 4 (10.10.4.0/255.255.255.0)"
 

	
 
        self.assertEqual(str(interface), representation)
 

	
 

	
 
class CommunicationTest(TestCase):
 
    fixtures = ['test-data.json']
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_unique_communication(self):
 
        """
 
        Test enforcement of unique communications.
 
        """
 

	
 
        comm = Communication.objects.get(pk=1)
 

	
 
        self.assertRaises(IntegrityError, Communication.objects.create, source=comm.source, destination=comm.destination, protocol=comm.protocol, port=comm.port, description="Duplicate communication.")
 

	
 
    def test_project_same(self):
 
        """
 
        Test enforcement of same project entities for communications.
 
        """
 

	
 
        ent1 = Entity.objects.get(name="Test Entity 1")
 
        ent1 = Entity.objects.get(pk=1)
 
        ent1_eth0 = ent1.interface_set.get(name="eth0")
 
        ent2 = Entity.objects.get(name="Other Project Test Entity")
 
        ent2_eth0 = ent2.interface_set.get(name="eth0")
 
        ent5 = Entity.objects.get(pk=5)
 
        ent5_eth0 = ent5.interface_set.get(name="eth0")
 

	
 
        # Set-up a communication between different projects.
 
        comm = Communication.objects.create(source=ent1_eth0, destination=ent2_eth0, protocol="ICMP", port="8", description="Ping.")
 
        comm = Communication.objects.create(source=ent1_eth0, destination=ent5_eth0, protocol="ICMP", port="8", description="Ping.")
 

	
 
        self.assertRaisesRegexp(ValidationError, 'Source and destination entities do not belong to the same project', comm.full_clean)
 

	
 
    def test_same_entity(self):
 
        """
 
        Test enforcement of differing entities for communication.
 
        """
 

	
 
        ent = Entity.objects.get(name="Test Entity 1")
 
        ent = Entity.objects.get(pk=1)
 
        ent_eth0 = ent.interface_set.get(name="eth0")
 

	
 
        # Set-up a communication between same entity.
 
        comm = Communication.objects.create(source=ent_eth0, destination=ent_eth0, protocol="ICMP", port="8", description="Ping.")
 

	
 
        self.assertRaisesRegexp(ValidationError, "Source and destination entities are identical.", comm.full_clean)
 

	
 
    def test_unsupported_protocol(self):
 
        """
 
        Test enforcement of supported protocol.
 
        """
 

	
 
        ent1 = Entity.objects.get(name="Test Entity 1")
 
        ent1 = Entity.objects.get(pk=1)
 
        ent1_eth0 = ent1.interface_set.get(name="eth0")
 
        ent2 = Entity.objects.get(name="Test Entity 2")
 
        ent2 = Entity.objects.get(pk=2)
 
        ent2_eth0 = ent2.interface_set.get(name="eth0")
 

	
 
        comm = Communication(source=ent1_eth0, destination=ent2_eth0, protocol="BOGUS", port="1234")
 

	
 
        self.assertRaisesRegexp(ValidationError, "BOGUS is not a supported protocol.", comm.full_clean)
 

	
 
    def test_edit_link(self):
 
        """
 
        Tests the function for getting the edit link string.
 
        """
 

	
 
        comm = Communication.objects.get(pk=1)
conntrackt/tests/test_utils.py
Show inline comments
 
# Django imports.
 
from django.test import TestCase
 

	
 
# Third-party Python library imports.
 
import palette
 
import pydot
 

	
 
# Application imports.
 
from conntrackt.models import Entity, Project, Communication
 
from conntrackt import utils
 
from .factories import setup_test_data
 

	
 

	
 
class GenerateEntityIptablesTest(TestCase):
 

	
 
    fixtures = ['test-data.json']
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_generated_iptables(self):
 
        """
 
        Tests if the entity's iptables are generated properly or not.
 
        """
 

	
 
        entity = Entity.objects.get(pk=1)
 
        generated = utils.generate_entity_iptables(entity)
 

	
 
        expected = """*filter
 
:INPUT DROP [0:0]
 
# Accept all incoming related traffic.
 
-A INPUT -m state --state RELATED,ESTABLISHED -j ACCEPT
 

	
 
# Accept all incoming traffic on loopback interface.
 
-A INPUT -i lo -j ACCEPT
 

	
 
# MySQL.
 
-A INPUT -s 192.168.1.3/255.255.255.255 -d 192.168.1.1/255.255.255.255 -p tcp -m tcp --dport 3306 -j ACCEPT
 

	
 
# Ping.
 
# Communicate over ICMP:8
 
-A INPUT -s 192.168.1.2/255.255.255.255 -d 192.168.1.1/255.255.255.255 -p icmp -m icmp --icmp-type 8 -j ACCEPT
 

	
 
# SSH.
 
# Communicate over TCP:22
 
-A INPUT -s 192.168.1.2/255.255.255.255 -d 192.168.1.1/255.255.255.255 -p tcp -m tcp --dport 22 -j ACCEPT
 
-A INPUT -s 192.168.2.0/255.255.255.0 -d 192.168.1.1/255.255.255.255 -p tcp -m tcp --dport 22 -j ACCEPT
 
-A INPUT -s 10.10.4.0/255.255.255.0 -d 192.168.1.1/255.255.255.255 -p tcp -m tcp --dport 22 -j ACCEPT
 

	
 
# Communicate over TCP:3306
 
-A INPUT -s 192.168.1.3/255.255.255.255 -d 192.168.1.1/255.255.255.255 -p tcp -m tcp --dport 3306 -j ACCEPT
 

	
 
:OUTPUT ACCEPT [0:0]
 
:FORWARD DROP [0:0]
 
COMMIT
 
*nat
 
:PREROUTING ACCEPT [0:0]
 
:INPUT ACCEPT [0:0]
 
:OUTPUT ACCEPT [0:0]
 
:POSTROUTING ACCEPT [0:0]
 
COMMIT
 
"""
 
        self.assertEqual(generated, expected)
 
@@ -108,25 +114,30 @@ class GetDistinctColorsTest(TestCase):
 
        self.assertEqual(equal.count(True), 12)
 

	
 
        # Check the difference between first and last colour.
 
        equal = abs(colors[0].hsl["h"] + 1 - colors[12].hsl["h"] - reference) < delta
 
        self.assertEqual(True, equal)
 

	
 

	
 
class GenerateProjectDiagramTest(TestCase):
 
    """
 
    Tests the generate_project_diagram function.
 
    """
 

	
 
    fixtures = ["test-data.json"]
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_unique_entity_colors(self):
 
        """
 
        Tests if each node/entity in the graph will have a unique colour.
 
        """
 

	
 
        # Get diagram for project
 
        project = Project.objects.get(pk=1)
 
        diagram = utils.generate_project_diagram(project)
 

	
 
        # Extract all nodes
 
        clusters = diagram.get_subgraphs()
 
@@ -162,25 +173,25 @@ class GenerateProjectDiagramTest(TestCas
 
            self.assertEqual(nodes[edge.get_source()].get_color(), edge.get_color())
 

	
 
    def test_entities_present(self):
 
        """
 
        Tests if all (and only) specific project entities are in the graph.
 
        """
 

	
 
        # Get diagram for project
 
        project = Project.objects.get(pk=1)
 
        diagram = utils.generate_project_diagram(project)
 

	
 
        # Set-up expected node names.
 
        expected_node_names = ["Test Entity 1", "Test Entity 2", "Test Entity 3", "Test Subnet"]
 
        expected_node_names = [u"Test Entity 1", u"Test Entity 2", u"Test Entity 3", u"Test Subnet 4"]
 

	
 
        # Get all nodes from diagram.
 
        clusters = diagram.get_subgraphs()
 
        nodes = []
 
        for cluster in clusters:
 
            nodes.extend(cluster.get_nodes())
 

	
 
        # Get the node names, strip the quotes from them.
 
        node_names = [n.get_name().replace('"', '') for n in nodes]
 

	
 
        # Validate that the two lists contain same elements.
 
        self.assertEqual(sorted(expected_node_names), sorted(node_names))
 
@@ -195,27 +206,30 @@ class GenerateProjectDiagramTest(TestCas
 
        project = Project.objects.get(pk=1)
 
        diagram = utils.generate_project_diagram(project)
 

	
 
        # Get all edges from the diagram.
 
        edges = diagram.get_edges()
 

	
 
        # Create list of edge labels.
 
        edge_labels = ["%s -> %s (%s)" % (e.get_source().replace('"', ''),
 
                                           e.get_destination().replace('"', ''),
 
                                           e.get_label().replace('"', '')) for e in edges]
 

	
 
        # Create list of expected edge labels
 
        expected_edge_labels = ['Test Entity 1 -> Test Entity 2 (UDP:123)', 'Test Entity 1 -> Test Entity 3 (UDP:53)',
 
                          'Test Entity 2 -> Test Entity 1 (ICMP:8)', 'Test Entity 2 -> Test Entity 1 (TCP:22)',
 
                          'Test Entity 3 -> Test Entity 1 (TCP:3306)', 'Test Subnet -> Test Entity 1 (TCP:22)']
 
        expected_edge_labels = [u'Test Entity 1 -> Test Entity 2 (UDP:123)',
 
                                u'Test Entity 1 -> Test Entity 3 (UDP:53)',
 
                                u'Test Entity 2 -> Test Entity 1 (ICMP:8)',
 
                                u'Test Entity 2 -> Test Entity 1 (TCP:22)',
 
                                u'Test Entity 3 -> Test Entity 1 (TCP:3306)',
 
                                u'Test Subnet 4 -> Test Entity 1 (TCP:22)']
 

	
 
        self.assertEqual(sorted(expected_edge_labels), sorted(edge_labels))
 

	
 
    def test_locations_present(self):
 
        """
 
        Tests if all (and only) specific project locations are in the graph (as
 
        clusters).
 
        """
 

	
 
        # Get diagram for project.
 
        project = Project.objects.get(pk=1)
 
        diagram = utils.generate_project_diagram(project)
conntrackt/tests/test_views.py
Show inline comments
 
@@ -18,33 +18,39 @@ from conntrackt.views import IndexView
 
from conntrackt.views import entity_iptables, project_iptables, project_diagram
 

	
 
from conntrackt.views import ProjectView, ProjectCreateView, ProjectUpdateView, ProjectDeleteView
 
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
 

	
 
# Test imports.
 
from .forms import FormWithWidgetCSSClassFormMixin, FormWithPlaceholderFormMixin
 
from .helpers import PermissionTestMixin, create_get_request, generate_get_response, FakeMessages
 
from .views import RedirectToNextMixinView
 
from .factories import setup_test_data
 

	
 

	
 
class IndexViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    sufficient_permissions = ("view",)
 
    view_class = IndexView
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context_no_projects(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called and
 
        no projects are available.
 
        """
 

	
 
        Project.objects.all().delete()
 

	
 
        # Get the view.
 
        view = IndexView.as_view()
 

	
 
        # Get the response.
 
@@ -93,30 +99,35 @@ class IndexViewTest(PermissionTestMixin,
 
        # Get the view.
 
        view = IndexView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view)
 

	
 
        # Validate the response.
 
        self.assertQuerysetEqual(response.context_data["locations"], ["<Location: Test Location 1>", "<Location: Test Location 2>"])
 

	
 

	
 
class ProjectViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    sufficient_permissions = ("view",)
 
    permission_test_view_kwargs = {"pk": "1"}
 
    view_class = ProjectView
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called.
 
        """
 

	
 
        # Get the view.
 
        view = ProjectView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view, pk=1)
 

	
 
        # Fetch context data from response.
 
@@ -126,82 +137,92 @@ class ProjectViewTest(PermissionTestMixi
 
        expected_entities = ["<Entity: Test Entity 1 (Test Project 1 - Test Location 1)>",
 
                             "<Entity: Test Entity 2 (Test Project 1 - Test Location 1)>"]
 

	
 
        # Validate context data.
 
        self.assertEqual(location.name, "Test Location 1")
 
        self.assertQuerysetEqual(entities, expected_entities)
 

	
 
        # Fetch context data from response.
 
        location, entities = response.context_data["location_entities"][1]
 

	
 
        # Set-up expected context data values.
 
        expected_entities = ["<Entity: Test Entity 3 (Test Project 1 - Test Location 2)>",
 
                             "<Entity: Test Subnet (Test Project 1 - Test Location 2)>"]
 
                             "<Entity: Test Subnet 4 (Test Project 1 - Test Location 2)>"]
 

	
 
        # Validate context data.
 
        self.assertEqual(location.name, "Test Location 2")
 
        self.assertQuerysetEqual(entities, expected_entities)
 

	
 
        # Validate context data.
 
        self.assertEqual(str(response.context_data["project"]), "Test Project 1")
 

	
 

	
 
class EntityViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = EntityView
 
    sufficient_permissions = ("view",)
 
    permission_test_view_kwargs = {"pk": "1"}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Tests if the form comes pre-populated with proper content.
 
        """
 

	
 
        # Get the view.
 
        view = EntityView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view, pk=1)
 

	
 
        # Set-up expected context data.
 
        expected_entity = Entity.objects.get(pk=1)
 

	
 
        expected_incoming_communications = ["<Communication: Test Entity 2 -> Test Entity 1 (TCP:22)>",
 
                                            "<Communication: Test Entity 2 -> Test Entity 1 (ICMP:8)>",
 
                                            "<Communication: Test Entity 3 -> Test Entity 1 (TCP:3306)>",
 
                                            "<Communication: Test Subnet -> Test Entity 1 (TCP:22)>"]
 
                                            "<Communication: Test Subnet 4 -> Test Entity 1 (TCP:22)>"]
 

	
 
        expected_outgoing_communications = ["<Communication: Test Entity 1 -> Test Entity 2 (UDP:123)>",
 
                                            "<Communication: Test Entity 1 -> Test Entity 3 (UDP:53)>"]
 

	
 
        expected_interfaces = ["<Interface: Test Entity 1 (192.168.1.1)>"]
 

	
 
        # Validate the response.
 
        self.assertQuerysetEqual(response.context_data["interfaces"], expected_interfaces)
 
        self.assertQuerysetEqual(response.context_data["incoming_communications"], expected_incoming_communications)
 
        self.assertQuerysetEqual(response.context_data["outgoing_communications"], expected_outgoing_communications)
 
        self.assertEqual(response.context_data["entity"], expected_entity)
 
        self.assertTrue("entity_iptables" in response.context_data)
 

	
 

	
 
class EntityIptablesTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_function = staticmethod(entity_iptables)
 
    sufficient_permissions = ("view",)
 
    permission_test_view_kwargs = {"pk": "1"}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_invalid_entity(self):
 
        """
 
        Tests if a 404 is returned if no entity was found (invalid ID).
 
        """
 

	
 
        # Set-up a request.
 
        request = create_get_request()
 

	
 
        # Get the view.
 
        view = entity_iptables
 

	
 
        # Validate the response.
 
@@ -242,30 +263,35 @@ class EntityIptablesTest(PermissionTestM
 
        view = entity_iptables
 

	
 
        # Get the response.
 
        response = generate_get_response(view, pk=1)
 

	
 
        self.assertContains(response, ":INPUT")
 
        self.assertContains(response, ":OUTPUT")
 
        self.assertContains(response, ":FORWARD")
 

	
 

	
 
class ProjectIptablesTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_function = staticmethod(project_iptables)
 
    sufficient_permissions = ("view",)
 
    permission_test_view_kwargs = {"project_id": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_invalid_project(self):
 
        """
 
        Tests if a 404 is returned if no project was found (invalid ID).
 
        """
 

	
 
        # Set-up a request.
 
        request = create_get_request()
 

	
 
        # Get the view.
 
        view = project_iptables
 

	
 
        # Request iptables for whole project.
 
@@ -325,25 +351,25 @@ class ProjectIptablesTest(PermissionTest
 
        # Get the view.
 
        view = project_iptables
 

	
 
        # Get the response.
 
        response = generate_get_response(project_iptables, None, 1)
 

	
 
        buff = StringIO(response.content)
 

	
 
        zipped_iptables = ZipFile(buff, "r", ZIP_DEFLATED)
 
        expected_zip_files = ["test_entity_1-iptables.conf",
 
                              "test_entity_2-iptables.conf",
 
                              "test_entity_3-iptables.conf",
 
                              "test_subnet-iptables.conf"]
 
                              "test_subnet_4-iptables.conf"]
 

	
 
        self.assertEqual(len(zipped_iptables.namelist()), 4)
 
        self.assertEqual(zipped_iptables.namelist(), expected_zip_files)
 

	
 
        for filename in expected_zip_files:
 
            iptables_file = zipped_iptables.read(filename)
 
            self.assertIn(":INPUT", iptables_file)
 
            self.assertIn(":OUTPUT", iptables_file)
 
            self.assertIn(":FORWARD", iptables_file)
 

	
 
        zipped_iptables.close()
 

	
 
@@ -376,54 +402,64 @@ class ProjectIptablesTest(PermissionTest
 

	
 
        zipped_iptables.close()
 

	
 

	
 
class ProjectCreateViewTest(PermissionTestMixin, TestCase):
 

	
 
    view_class = ProjectCreateView
 
    sufficient_permissions = ("add_project",)
 

	
 

	
 
class ProjectUpdateViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = ProjectUpdateView
 
    sufficient_permissions = ("change_project",)
 
    permission_test_view_kwargs = {"pk": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific project.
 
        """
 

	
 
        # Get the view.
 
        view = ProjectUpdateView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view, None, pk=1)
 

	
 
        self.assertEqual(response.context_data["project"].name, "Test Project 1")
 
        self.assertEqual(response.context_data["headline"], "Update project Test Project 1")
 

	
 

	
 
class ProjectDeleteViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = ProjectDeleteView
 
    sufficient_permissions = ("delete_project",)
 
    permission_test_view_kwargs = {"pk": "1"}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific project.
 
        """
 

	
 
        # Get the expected project.
 
        project = Project.objects.get(pk=1)
 

	
 
        # Get the view.
 
        view = ProjectDeleteView.as_view()
 

	
 
@@ -452,54 +488,64 @@ class ProjectDeleteViewTest(PermissionTe
 

	
 
        self.assertIn("Project Test Project 1 has been removed.", request._messages.messages)
 

	
 

	
 
class LocationCreateViewTest(PermissionTestMixin, TestCase):
 

	
 
    view_class = LocationCreateView
 
    sufficient_permissions = ("add_location",)
 

	
 

	
 
class LocationUpdateViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = LocationUpdateView
 
    sufficient_permissions = ("change_location",)
 
    permission_test_view_kwargs = {"pk": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific location.
 
        """
 

	
 
        # Get the view.
 
        view = LocationUpdateView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view, None, pk=1)
 

	
 
        self.assertEqual(response.context_data["location"].name, "Test Location 1")
 
        self.assertEqual(response.context_data["headline"], "Update location Test Location 1")
 

	
 

	
 
class LocationDeleteViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = LocationDeleteView
 
    sufficient_permissions = ("delete_location",)
 
    permission_test_view_kwargs = {"pk": "1"}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific location.
 
        """
 

	
 
        # Get the expected location.
 
        location = Location.objects.get(pk=1)
 

	
 
        # Get the view.
 
        view = LocationDeleteView.as_view()
 

	
 
@@ -600,30 +646,35 @@ class EntityCreateViewTest(PermissionTes
 

	
 
        view = EntityCreateView()
 
        view.request = RequestFactory().get("/fake-path?location=1")
 
        view.object = None
 

	
 
        initial = view.get_initial()
 

	
 
        self.assertDictContainsSubset({"location": "1"}, initial)
 

	
 

	
 
class EntityDeleteViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = EntityDeleteView
 
    sufficient_permissions = ("delete_entity",)
 
    permission_test_view_kwargs = {"pk": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific entity.
 
        """
 

	
 
        # Get the expected entity.
 
        entity = Entity.objects.get(pk=1)
 

	
 
        # Get the view.
 
        view = EntityDeleteView.as_view()
 

	
 
@@ -666,30 +717,35 @@ class EntityDeleteViewTest(PermissionTes
 
        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("project", args=(1,)))
 

	
 

	
 
class EntityUpdateViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = EntityUpdateView
 
    sufficient_permissions = ("change_entity",)
 
    permission_test_view_kwargs = {"pk": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific entity.
 
        """
 

	
 
        # Get the view.
 
        view = EntityUpdateView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view, None, pk=1)
 

	
 
@@ -760,30 +816,35 @@ class InterfaceCreateViewTest(Permission
 
        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)
 

	
 

	
 
class InterfaceUpdateViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = InterfaceUpdateView
 
    sufficient_permissions = ("change_interface",)
 
    permission_test_view_kwargs = {"pk": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific entity.
 
        """
 

	
 
        # Get the view.
 
        view = InterfaceUpdateView.as_view()
 

	
 
        # Get the response.
 
        response = generate_get_response(view, None, pk=1)
 

	
 
@@ -801,25 +862,25 @@ class InterfaceUpdateViewTest(Permission
 

	
 
        # Set-up the view.
 
        view = InterfaceUpdateView()
 
        view.request = RequestFactory().get("/fake-path/1")
 
        view.object = Interface.objects.get(pk=1)
 

	
 
        # Get the form.
 
        form = view.get_form(view.get_form_class())
 

	
 
        expected_entities = ["<Entity: Test Entity 1 (Test Project 1 - Test Location 1)>",
 
                             "<Entity: Test Entity 2 (Test Project 1 - Test Location 1)>",
 
                             "<Entity: Test Entity 3 (Test Project 1 - Test Location 2)>",
 
                             "<Entity: Test Subnet (Test Project 1 - Test Location 2)>"]
 
                             "<Entity: Test Subnet 4 (Test Project 1 - Test Location 2)>"]
 

	
 
        self.assertQuerysetEqual(form.fields["entity"].queryset, expected_entities)
 

	
 
    def test_success_url(self):
 
        """
 
        Validate that the success URL is set properly after update.
 
        """
 

	
 
        # Get the view.
 
        view = InterfaceUpdateView.as_view()
 

	
 
        # Get the interface object.
 
@@ -833,30 +894,35 @@ class InterfaceUpdateViewTest(Permission
 
        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)
 

	
 

	
 
class InterfaceDeleteViewTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_class = InterfaceDeleteView
 
    sufficient_permissions = ("delete_interface",)
 
    permission_test_view_kwargs = {"pk": 1}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_context(self):
 
        """
 
        Verifies that the context is properly set-up when the view is called for
 
        specific interface.
 
        """
 

	
 
        # Get the expected entity.
 
        interface = Interface.objects.get(pk=1)
 

	
 
        # Get the view.
 
        view = InterfaceDeleteView.as_view()
 

	
 
@@ -1121,30 +1187,35 @@ class CommunicationCreateViewTest(Permis
 
        request.user = mock.Mock()
 
        request._dont_enforce_csrf_checks = True
 

	
 
        # Get the response.
 
        response = view(request)
 

	
 
        self.assertEqual(response["Location"], reverse("project", 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 setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    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.
 
@@ -1161,25 +1232,25 @@ class CommunicationUpdateViewTest(Permis
 

	
 
        # 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)>"]
 
                               "<Interface: Test Subnet 4 (10.10.4.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_next(self):
 
        """
 
        Validate that the success URL is set properly after update if GET
 
        parameter is passed.
 
        """
 

	
 
        # Get the view.
 
        view = CommunicationUpdateView.as_view()
 
@@ -1223,30 +1294,35 @@ class CommunicationUpdateViewTest(Permis
 
        request.user = mock.Mock()
 
        request._dont_enforce_csrf_checks = True
 

	
 
        # Get the response.
 
        response = view(request, pk=1)
 

	
 
        self.assertEqual(response["Location"], reverse("project", args=(communication.source.entity.project.id,)))
 
        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 setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    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()
 

	
 
@@ -1330,30 +1406,35 @@ class CommunicationDeleteViewTest(Permis
 
        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,)))
 

	
 

	
 
class ProjectDiagramTest(PermissionTestMixin, TestCase):
 

	
 
    fixtures = ['test-data.json']
 

	
 
    view_function = staticmethod(project_diagram)
 
    sufficient_permissions = ("view",)
 
    permission_test_view_kwargs = {"pk": "1"}
 

	
 
    def setUp(self):
 
        """
 
        Set-up some test data.
 
        """
 

	
 
        setup_test_data()
 

	
 
    def test_invalid_project(self):
 
        """
 
        Tests if a 404 is returned if no project was found (invalid ID).
 
        """
 

	
 
        # Set-up a request.
 
        request = create_get_request()
 

	
 
        # Get the view.
 
        view = project_diagram
 

	
 
        # Validate the response.
0 comments (0 inline, 0 general)