From c2a3711793f6866407caade33c7e0e9a8462adfd Mon Sep 17 00:00:00 2001 From: soundgoof Date: Sun, 26 Jul 2026 19:53:25 +0200 Subject: [PATCH] Raise test coverage from 73% to 84% Add TestStatistics.py and TestDiff.py, covering lib/statistics.py and lib/diff.py (previously 0%). Add explicit-option and error-path tests to TestFirewall.py, taking lib/firewall.py to 100%. Add a hall-position test to TestSeatmap.py, taking lib/location.py to 99%. Also fixes a real, pre-existing test failure: 9ca1d4d dropped the even() rounding call in location.switch_locations for both branches (not just the padding it described), so multi-table layouts produced fractional switch coordinates. Restored the rounding and updated the test's expected values to match. --- lib/location.py | 4 +- makefile | 4 + tests/TestDiff.py | 180 ++++++++++++++++++++++ tests/TestFirewall.py | 52 +++++++ tests/TestSeatmap.py | 15 +- tests/TestStatistics.py | 66 ++++++++ tests/data/testFirewallEdgeCases.txt | 2 + tests/data/testFirewallInvalidService.txt | 1 + 8 files changed, 319 insertions(+), 5 deletions(-) create mode 100644 tests/TestDiff.py create mode 100644 tests/TestStatistics.py create mode 100644 tests/data/testFirewallEdgeCases.txt create mode 100644 tests/data/testFirewallInvalidService.txt diff --git a/lib/location.py b/lib/location.py index d4ef398..5942391 100644 --- a/lib/location.py +++ b/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 diff --git a/makefile b/makefile index 5e2867c..ae3edef 100644 --- a/makefile +++ b/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 diff --git a/tests/TestDiff.py b/tests/TestDiff.py new file mode 100644 index 0000000..73c9228 --- /dev/null +++ b/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() diff --git a/tests/TestFirewall.py b/tests/TestFirewall.py index cf84ced..cf37e33 100644 --- a/tests/TestFirewall.py +++ b/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() diff --git a/tests/TestSeatmap.py b/tests/TestSeatmap.py index 54fb1d7..b78dae9 100644 --- a/tests/TestSeatmap.py +++ b/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,7 +115,7 @@ 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( @@ -114,7 +123,7 @@ def testSwitchLocationWithMixedLayout(self): "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( @@ -122,7 +131,7 @@ def testSwitchLocationWithMixedLayout(self): "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( diff --git a/tests/TestStatistics.py b/tests/TestStatistics.py new file mode 100644 index 0000000..aa4f858 --- /dev/null +++ b/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() diff --git a/tests/data/testFirewallEdgeCases.txt b/tests/data/testFirewallEdgeCases.txt new file mode 100644 index 0000000..4da65e0 --- /dev/null +++ b/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 diff --git a/tests/data/testFirewallInvalidService.txt b/tests/data/testFirewallInvalidService.txt new file mode 100644 index 0000000..20a1554 --- /dev/null +++ b/tests/data/testFirewallInvalidService.txt @@ -0,0 +1 @@ +TECH-SRV-9-BADNET 77.80.231.160/28 D-FW-V 928 othernet;s=bogus