"host",
"port",
"compress",
+ "magic",
"connect_timeout",
"connect_retries",
"cmd_prefix",
comprcmd = "gzip"
for compress in [constants.IEC_NONE, constants.IEC_GZIP]:
- for host in ["localhost", "1.2.3.4", "192.0.2.99"]:
+ for magic in [None, 10 * "-", "HelloWorld", "J9plh4nFo2",
+ "24A02A81-2264-4B51-A882-A2AB9D85B420"]:
+ opts = CmdBuilderConfig(magic=magic, compress=compress)
+ builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3)
+
+ magic_cmd = builder._GetMagicCommand()
+ dd_cmd = builder._GetDdCommand()
+
+ if magic:
+ self.assert_(("M=%s" % magic) in magic_cmd)
+ self.assert_(("M=%s" % magic) in dd_cmd)
+ else:
+ self.assertFalse(magic_cmd)
+
+ for host in ["localhost", "198.51.100.4", "192.0.2.99"]:
for port in [0, 1, 1234, 7856, 45452]:
for cmd_prefix in [None, "PrefixCommandGoesHere|",
"dd if=/dev/hda bs=1048576 |"]:
builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3)
self.assertRaises(errors.GenericError, builder.GetCommand)
+ def testOptionLengthError(self):
+ testopts = [
+ CmdBuilderConfig(bind="0.0.0.0" + ("A" * impexpd.SOCAT_OPTION_MAXLEN),
+ port=1234, ca="/tmp/ca"),
+ CmdBuilderConfig(host="localhost", port=1234,
+ ca="/tmp/ca" + ("B" * impexpd.SOCAT_OPTION_MAXLEN)),
+ CmdBuilderConfig(host="localhost", port=1234,
+ key="/tmp/key" + ("B" * impexpd.SOCAT_OPTION_MAXLEN)),
+ ]
+
+ for opts in testopts:
+ for mode in [constants.IEM_IMPORT, constants.IEM_EXPORT]:
+ builder = impexpd.CommandBuilder(mode, opts, 1, 2, 3)
+ self.assertRaises(errors.GenericError, builder.GetCommand)
+
+ opts.host = "localhost" + ("A" * impexpd.SOCAT_OPTION_MAXLEN)
+ builder = impexpd.CommandBuilder(constants.IEM_EXPORT, opts, 1, 2, 3)
+ self.assertRaises(errors.GenericError, builder.GetCommand)
+
def testModeError(self):
mode = "foobarbaz"