x/vmware-nsx
Code Issues Proposed changes

Deal with intermittent OverflowError in gate-neutron-python26

Browse Source 
This occurence looks like is related to a failure in
utils.execute. This patch attempts to deal with the error
by mocking out Popen.communicate altogether.

Fixes bug #1202065

Change-Id: I4092efb17939b69dd2d1965aaf619a9b2be03ef4
This commit is contained in:
armando-migliaccio 2013-07-24 17:35:28 -07:00
parent 97046a78b0
commit 530a28a9b1
1 changed files with 25 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
neutron/tests/unit/test_agent_linux_utils.py
View File

@ -29,38 +29,54 @@ class AgentUtilsExecuteTest(base.BaseTestCase):
self.test_file = self.useFixture( self.test_file = self.useFixture(
fixtures.TempDir()).join("test_execute.tmp") fixtures.TempDir()).join("test_execute.tmp")
open(self.test_file, 'w').close() open(self.test_file, 'w').close()
instance = mock.patch("subprocess.Popen.communicate")
self.mock_popen = instance.start()
self.addCleanup(self.mock_popen.stop)
def test_without_helper(self): def test_without_helper(self):
expected = "%s\n" % self.test_file
self.mock_popen.return_value = [expected, ""]
result = utils.execute(["ls", self.test_file]) result = utils.execute(["ls", self.test_file])
self.assertEqual(result, "%s\n" % self.test_file) self.assertEqual(result, expected)
def test_with_helper(self): def test_with_helper(self):
expected = "ls %s\n" % self.test_file
self.mock_popen.return_value = [expected, ""]
result = utils.execute(["ls", self.test_file], result = utils.execute(["ls", self.test_file],
self.root_helper) self.root_helper)
self.assertEqual(result, "ls %s\n" % self.test_file) self.assertEqual(result, expected)
def test_stderr(self): def test_stderr_true(self):
stdout, stderr = utils.execute(["ls", self.test_file], expected = "%s\n" % self.test_file
return_stderr=True) self.mock_popen.return_value = [expected, ""]
self.assertEqual(stdout, "%s\n" % self.test_file) out = utils.execute(["ls", self.test_file], return_stderr=True)
self.assertEqual(stderr, "") self.assertIsInstance(out, tuple)
self.assertEqual(out, (expected, ""))
def test_check_exit_code(self): def test_check_exit_code(self):
self.mock_popen.return_value = ["", ""]
stdout = utils.execute(["ls", self.test_file[:-1]], stdout = utils.execute(["ls", self.test_file[:-1]],
check_exit_code=False) check_exit_code=False)
self.assertEqual(stdout, "") self.assertEqual(stdout, "")
def test_execute_raises(self):
self.mock_popen.side_effect = RuntimeError
self.assertRaises(RuntimeError, utils.execute, self.assertRaises(RuntimeError, utils.execute,
["ls", self.test_file[:-1]]) ["ls", self.test_file[:-1]])
def test_process_input(self): def test_process_input(self):
expected = "%s\n" % self.test_file[:-1]
self.mock_popen.return_value = [expected, ""]
result = utils.execute(["cat"], process_input="%s\n" % result = utils.execute(["cat"], process_input="%s\n" %
self.test_file[:-1]) self.test_file[:-1])
self.assertEqual(result, "%s\n" % self.test_file[:-1]) self.assertEqual(result, expected)
def test_with_addl_env(self): def test_with_addl_env(self):
expected = "%s\n" % self.test_file
self.mock_popen.return_value = [expected, ""]
result = utils.execute(["ls", self.test_file], result = utils.execute(["ls", self.test_file],
addl_env={'foo': 'bar'}) addl_env={'foo': 'bar'})
self.assertEqual(result, "%s\n" % self.test_file) self.assertEqual(result, expected)
class AgentUtilsGetInterfaceMAC(base.BaseTestCase): class AgentUtilsGetInterfaceMAC(base.BaseTestCase):