Revision 58bb385c test/ganeti.impexpd_unittest.py
| b/test/ganeti.impexpd_unittest.py | ||
|---|---|---|
| 25 | 25 |
import sys |
| 26 | 26 |
import re |
| 27 | 27 |
import unittest |
| 28 |
import socket |
|
| 28 | 29 |
|
| 29 | 30 |
from ganeti import constants |
| 30 | 31 |
from ganeti import objects |
| ... | ... | |
| 44 | 45 |
"ca", |
| 45 | 46 |
"host", |
| 46 | 47 |
"port", |
| 48 |
"ipv4", |
|
| 49 |
"ipv6", |
|
| 47 | 50 |
"compress", |
| 48 | 51 |
"magic", |
| 49 | 52 |
"connect_timeout", |
| ... | ... | |
| 101 | 104 |
self.assert_(CheckCmdWord(cmd, comprcmd)) |
| 102 | 105 |
|
| 103 | 106 |
if cmd_prefix is not None: |
| 104 |
self.assert_(cmd_prefix in i for i in cmd)
|
|
| 107 |
self.assert_(compat.any(cmd_prefix in i for i in cmd))
|
|
| 105 | 108 |
|
| 106 | 109 |
if cmd_suffix is not None: |
| 107 |
self.assert_(cmd_suffix in i for i in cmd)
|
|
| 110 |
self.assert_(compat.any(cmd_suffix in i for i in cmd))
|
|
| 108 | 111 |
|
| 109 | 112 |
# Check socat command |
| 110 | 113 |
socat_cmd = builder._GetSocatCommand() |
| ... | ... | |
| 118 | 121 |
|
| 119 | 122 |
self.assert_("verify=1" in ssl_addr)
|
| 120 | 123 |
|
| 124 |
def testIPv6(self): |
|
| 125 |
for mode in [constants.IEM_IMPORT, constants.IEM_EXPORT]: |
|
| 126 |
opts = CmdBuilderConfig(host="localhost", port=6789, |
|
| 127 |
ipv4=False, ipv6=False) |
|
| 128 |
builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3) |
|
| 129 |
cmd = builder._GetSocatCommand() |
|
| 130 |
self.assert_(compat.all("pf=" not in i for i in cmd))
|
|
| 131 |
|
|
| 132 |
# IPv4 |
|
| 133 |
opts = CmdBuilderConfig(host="localhost", port=6789, |
|
| 134 |
ipv4=True, ipv6=False) |
|
| 135 |
builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3) |
|
| 136 |
cmd = builder._GetSocatCommand() |
|
| 137 |
self.assert_(compat.any(",pf=ipv4" in i for i in cmd))
|
|
| 138 |
|
|
| 139 |
# IPv6 |
|
| 140 |
opts = CmdBuilderConfig(host="localhost", port=6789, |
|
| 141 |
ipv4=False, ipv6=True) |
|
| 142 |
builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3) |
|
| 143 |
cmd = builder._GetSocatCommand() |
|
| 144 |
self.assert_(compat.any(",pf=ipv6" in i for i in cmd))
|
|
| 145 |
|
|
| 146 |
# IPv4 and IPv6 |
|
| 147 |
opts = CmdBuilderConfig(host="localhost", port=6789, |
|
| 148 |
ipv4=True, ipv6=True) |
|
| 149 |
builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3) |
|
| 150 |
self.assertRaises(AssertionError, builder._GetSocatCommand) |
|
| 151 |
|
|
| 121 | 152 |
def testCommaError(self): |
| 122 | 153 |
opts = CmdBuilderConfig(host="localhost", port=1234, |
| 123 | 154 |
ca="/some/path/with,a/,comma") |
| ... | ... | |
| 155 | 186 |
self.assertRaises(errors.GenericError, builder.GetCommand) |
| 156 | 187 |
|
| 157 | 188 |
|
| 189 |
class TestVerifyListening(unittest.TestCase): |
|
| 190 |
def test(self): |
|
| 191 |
self.assertEqual(impexpd._VerifyListening(socket.AF_INET, |
|
| 192 |
"192.0.2.7", 1234), |
|
| 193 |
("192.0.2.7", 1234))
|
|
| 194 |
self.assertEqual(impexpd._VerifyListening(socket.AF_INET6, "::1", 9876), |
|
| 195 |
("::1", 9876))
|
|
| 196 |
self.assertEqual(impexpd._VerifyListening(socket.AF_INET6, "[::1]", 4563), |
|
| 197 |
("::1", 4563))
|
|
| 198 |
self.assertEqual(impexpd._VerifyListening(socket.AF_INET6, |
|
| 199 |
"[2001:db8::1:4563]", 4563), |
|
| 200 |
("2001:db8::1:4563", 4563))
|
|
| 201 |
|
|
| 202 |
def testError(self): |
|
| 203 |
for family in [socket.AF_UNIX, socket.AF_INET, socket.AF_INET6]: |
|
| 204 |
self.assertRaises(errors.GenericError, impexpd._VerifyListening, |
|
| 205 |
family, "", 1234) |
|
| 206 |
self.assertRaises(errors.GenericError, impexpd._VerifyListening, |
|
| 207 |
family, "192", 999) |
|
| 208 |
|
|
| 209 |
for family in [socket.AF_UNIX, socket.AF_INET6]: |
|
| 210 |
self.assertRaises(errors.GenericError, impexpd._VerifyListening, |
|
| 211 |
family, "192.0.2.7", 1234) |
|
| 212 |
self.assertRaises(errors.GenericError, impexpd._VerifyListening, |
|
| 213 |
family, "[2001:db8::1", 1234) |
|
| 214 |
self.assertRaises(errors.GenericError, impexpd._VerifyListening, |
|
| 215 |
family, "2001:db8::1]", 1234) |
|
| 216 |
|
|
| 217 |
for family in [socket.AF_UNIX, socket.AF_INET]: |
|
| 218 |
self.assertRaises(errors.GenericError, impexpd._VerifyListening, |
|
| 219 |
family, "::1", 1234) |
|
| 220 |
|
|
| 221 |
|
|
| 158 | 222 |
class TestCalcThroughput(unittest.TestCase): |
| 159 | 223 |
def test(self): |
| 160 | 224 |
self.assertEqual(impexpd._CalcThroughput([]), None) |
Also available in: Unified diff