Qubes/core-agent-linux
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests/firewall: some code refactoring

Browse Source
This commit is contained in:
3hhh 2021-05-16 07:32:57 +02:00
parent 0993115bdc
commit 3230f471b0
No known key found for this signature in database
GPG Key ID: EB03A691DB2F0833
1 changed files with 18 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
qubesagent/test_firewall.py
View File

@ -166,8 +166,20 @@ class NftablesWorker(qubesagent.firewall.NftablesWorker):
else: else:
return ['2001::1', '2001::2'] return ['2001::1', '2001::2']
class WorkerTestCase(TestCase):
def assertPrepareRulesDnsRet(self, dns_ret, expected_domain, family):
self.assertEqual(dns_ret.keys(), {expected_domain})
self.assertIsInstance(dns_ret[expected_domain], set)
if family == 4:
self.assertIsNotNone(re.match('^\d+\.\d+\.\d+\.\d+/32$',
dns_ret[expected_domain].pop()))
elif family == 6:
self.assertIsNotNone(re.match('^[0-9a-f:]+/\d+$',
dns_ret[expected_domain].pop()))
else:
raise ValueError()
class TestIptablesWorker(TestCase): class TestIptablesWorker(WorkerTestCase):
def setUp(self): def setUp(self):
super(TestIptablesWorker, self).setUp() super(TestIptablesWorker, self).setUp()
self.obj = IptablesWorker() self.obj = IptablesWorker()
@ -226,10 +238,7 @@ class TestIptablesWorker(TestCase):
) )
ret = self.obj.prepare_rules('chain', rules, 4) ret = self.obj.prepare_rules('chain', rules, 4)
self.assertEqual(ret[0], expected_iptables) self.assertEqual(ret[0], expected_iptables)
self.assertEqual(ret[1].keys(), {'yum.qubes-os.org'}) self.assertPrepareRulesDnsRet(ret[1], 'yum.qubes-os.org', 4)
self.assertIsInstance(ret[1]['yum.qubes-os.org'], set)
self.assertIsNotNone(re.match('^\d+\.\d+\.\d+\.\d+/32$',
ret[1]['yum.qubes-os.org'].pop()))
with self.assertRaises(qubesagent.firewall.RuleParseError): with self.assertRaises(qubesagent.firewall.RuleParseError):
self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4) self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
with self.assertRaises(qubesagent.firewall.RuleParseError): with self.assertRaises(qubesagent.firewall.RuleParseError):
@ -268,10 +277,7 @@ class TestIptablesWorker(TestCase):
) )
ret = self.obj.prepare_rules('chain', rules, 6) ret = self.obj.prepare_rules('chain', rules, 6)
self.assertEqual(ret[0], expected_iptables) self.assertEqual(ret[0], expected_iptables)
self.assertEqual(ret[1].keys(), {'ripe.net'}) self.assertPrepareRulesDnsRet(ret[1], 'ripe.net', 6)
self.assertIsInstance(ret[1]['ripe.net'], set)
self.assertIsNotNone(re.match('^[0-9a-f:]+/\d+$',
ret[1]['ripe.net'].pop()))
def test_004_apply_rules4(self): def test_004_apply_rules4(self):
rules = [{'action': 'accept'}] rules = [{'action': 'accept'}]
@ -393,7 +399,7 @@ class TestIptablesWorker(TestCase):
]) ])
class TestNftablesWorker(TestCase): class TestNftablesWorker(WorkerTestCase):
def setUp(self): def setUp(self):
super(TestNftablesWorker, self).setUp() super(TestNftablesWorker, self).setUp()
self.obj = NftablesWorker() self.obj = NftablesWorker()
@ -462,10 +468,7 @@ class TestNftablesWorker(TestCase):
) )
ret = self.obj.prepare_rules('chain', rules, 4) ret = self.obj.prepare_rules('chain', rules, 4)
self.assertEqual(ret[0], expected_nft) self.assertEqual(ret[0], expected_nft)
self.assertEqual(ret[1].keys(), {'yum.qubes-os.org'}) self.assertPrepareRulesDnsRet(ret[1], 'yum.qubes-os.org', 4)
self.assertIsInstance(ret[1]['yum.qubes-os.org'], set)
self.assertIsNotNone(re.match('^\d+\.\d+\.\d+\.\d+/32$',
ret[1]['yum.qubes-os.org'].pop()))
with self.assertRaises(qubesagent.firewall.RuleParseError): with self.assertRaises(qubesagent.firewall.RuleParseError):
self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4) self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
with self.assertRaises(qubesagent.firewall.RuleParseError): with self.assertRaises(qubesagent.firewall.RuleParseError):
@ -503,10 +506,7 @@ class TestNftablesWorker(TestCase):
) )
ret = self.obj.prepare_rules('chain', rules, 6) ret = self.obj.prepare_rules('chain', rules, 6)
self.assertEqual(ret[0], expected_nft) self.assertEqual(ret[0], expected_nft)
self.assertEqual(ret[1].keys(), {'ripe.net'}) self.assertPrepareRulesDnsRet(ret[1], 'ripe.net', 6)
self.assertIsInstance(ret[1]['ripe.net'], set)
self.assertIsNotNone(re.match('^[0-9a-f:]+/\d+$',
ret[1]['ripe.net'].pop()))
def test_004_apply_rules4(self): def test_004_apply_rules4(self):
rules = [{'action': 'accept'}] rules = [{'action': 'accept'}]