forked from saschpe/libvirt-hook-qemu
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_qemu.py
113 lines (95 loc) · 4.08 KB
/
test_qemu.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import imp
import json
import os
import socket
import sys
import textwrap
import unittest
qemu = imp.load_source('qemu', 'qemu')
class QemuTestCase(unittest.TestCase):
def test_host_ip(self):
host_ip = qemu.host_ip()
try:
socket.inet_aton(host_ip)
except socket.error:
pass # Not legal
def test_config(self):
conf = qemu.config(validate=False)
del qemu.config._conf # Revert closure
def test_config_schema_validation(self):
conf = qemu.config()
del qemu.config._conf # Revert closure
def capture_output(self, func):
outfile = "/tmp/libvirt-hook-qemu-test-output.txt"
orig_binary = qemu.IPTABLES_BINARY
orig_out = os.dup(sys.stdout.fileno())
try:
qemu.IPTABLES_BINARY = '/bin/echo'
to = open(outfile, "w")
os.dup2(to.fileno(), sys.stdout.fileno())
func()
finally:
sys.stdout.flush()
os.dup2(orig_out, sys.stdout.fileno())
qemu.IPTABLES_BINARY = orig_binary
output = open(outfile).read()
os.remove(outfile)
return output
def dedent(self, str):
return textwrap.dedent(str[1:])
def test_setup(self):
port_map = {
"udp": [53],
"tcp": [[80, 8080], 443]
}
# in this test, public_ip is 192.168.1.1 and private_ip is 127.0.0.1
expected_output = self.dedent("""
-t nat -N DNAT-test
-t nat -N SNAT-test
-t filter -N FWD-test
-t nat -A DNAT-test -p udp -d 192.168.1.1 --dport 53 -j DNAT --to 127.0.0.1:53
-t nat -A SNAT-test -p udp -s 127.0.0.1 -d 127.0.0.1 --dport 53 -j MASQUERADE
-t filter -A FWD-test -p udp -d 127.0.0.1 --dport 53 -j ACCEPT -o virbr0
-t nat -A DNAT-test -p tcp -d 192.168.1.1 --dport 80 -j DNAT --to 127.0.0.1:8080
-t nat -A SNAT-test -p tcp -s 127.0.0.1 -d 127.0.0.1 --dport 80 -j MASQUERADE
-t filter -A FWD-test -p tcp -d 127.0.0.1 --dport 8080 -j ACCEPT -o virbr0
-t nat -A DNAT-test -p tcp -d 192.168.1.1 --dport 443 -j DNAT --to 127.0.0.1:443
-t nat -A SNAT-test -p tcp -s 127.0.0.1 -d 127.0.0.1 --dport 443 -j MASQUERADE
-t filter -A FWD-test -p tcp -d 127.0.0.1 --dport 443 -j ACCEPT -o virbr0
-t nat -I OUTPUT -d 192.168.1.1 -j DNAT-test
-t nat -I PREROUTING -d 192.168.1.1 -j DNAT-test
-t nat -I POSTROUTING -s 127.0.0.1 -d 127.0.0.1 -j SNAT-test
-t filter -I FORWARD -d 127.0.0.1 -j FWD-test
""")
# stub out the disable_bridge_filtering call
self.bridge_nf = False
def disable_bridge_filtering():
self.bridge_nf = True
self.old_bridge_filtering = qemu.disable_bridge_filtering
qemu.disable_bridge_filtering = disable_bridge_filtering
def test_func():
domain = { "port_map": port_map, "interface": "virbr0" }
qemu.start_forwarding("DNAT-test", "SNAT-test", "FWD-test", "192.168.1.1", "127.0.0.1", domain)
output = self.capture_output(test_func)
self.maxDiff = None
self.assertMultiLineEqual(output, expected_output)
self.assertTrue(self.bridge_nf)
qemu.disable_bridge_filtering = self.old_bridge_filtering
def test_teardown(self):
expected_output = self.dedent("""
-t nat -D OUTPUT -d 192.168.1.1 -j DNAT-test
-t nat -D PREROUTING -d 192.168.1.1 -j DNAT-test
-t nat -D POSTROUTING -s 127.0.0.1 -d 127.0.0.1 -j SNAT-test
-t filter -D FORWARD -d 127.0.0.1 -j FWD-test
-t nat -F DNAT-test
-t nat -X DNAT-test
-t nat -F SNAT-test
-t nat -X SNAT-test
-t filter -F FWD-test
-t filter -X FWD-test
""")
def test_func():
qemu.stop_forwarding("DNAT-test", "SNAT-test", "FWD-test", "192.168.1.1", "127.0.0.1")
output = self.capture_output(test_func)
self.maxDiff = None
self.assertMultiLineEqual(output, expected_output)