Add tests for compression tool misuse

This patch extends opcode tests for both unacceptable tools, and the
use of tools which are not on the compression_tools parameter list.
Signed-off-by: Hrvoje Ribicic <riba@google.com>
Reviewed-by: Thomas Thrainer <thomasth@google.com>

test/py/cmdlib/backup_unittest.py

@@ -143,6 +143,16 @@ class TestLUBackupExportLocalExport(TestLUBackupExportBase):
     op = self.CopyOpCode(self.op, remove_instance=True)
     self.ExecOpCode(op)
 
+  def testValidCompressionTool(self):
+    op = self.CopyOpCode(self.op, compress="lzop")
+    self.cfg.SetCompressionTools(["gzip", "lzop"])
+    self.ExecOpCode(op)
+
+  def testInvalidCompressionTool(self):
+    op = self.CopyOpCode(self.op, compress="invalid")
+    self.cfg.SetCompressionTools(["gzip", "lzop"])
+    self.ExecOpCodeExpectOpPrereqError(op, "Compression tool not allowed")
+
 
 class TestLUBackupExportRemoteExport(TestLUBackupExportBase):
   def setUp(self):

test/py/cmdlib/cluster_unittest.py

@@ -28,6 +28,7 @@ import OpenSSL
 import copy
 import unittest
 import operator
+import re
 
 from ganeti.cmdlib import cluster
 from ganeti import constants
@@ -980,6 +981,23 @@ class TestLUClusterSetParams(CmdlibTestCase):
 
     self.mcpu.assertLogContainsRegex("Could not disable the master ip")
 
+  def testCompressionToolSuccess(self):
+    compression_tools = ["certainly_not_a_default", "gzip"]
+    op = opcodes.OpClusterSetParams(compression_tools=compression_tools)
+    self.ExecOpCode(op)
+    self.assertEqual(compression_tools, self.cluster.compression_tools)
+
+  def testCompressionToolCompatibility(self):
+    compression_tools = ["not_gzip", "not_not_not_gzip"]
+    op = opcodes.OpClusterSetParams(compression_tools=compression_tools)
+    self.ExecOpCodeExpectOpPrereqError(op, ".*the gzip utility must be.*")
+
+  def testCompressionToolForbiddenValues(self):
+    for value in ["none", "\"rm -rf all.all\"", "ls$IFS-la"]:
+      compression_tools = [value, "gzip"]
+      op = opcodes.OpClusterSetParams(compression_tools=compression_tools)
+      self.ExecOpCodeExpectOpPrereqError(op, re.escape(value))
+
 
 class TestLUClusterVerify(CmdlibTestCase):
   def testVerifyAllGroups(self):