1 # Copyright 2013 Red Hat, Inc.
3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
4 # not use this file except in compliance with the License. You may obtain
5 # a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 # License for the specific language governing permissions and limitations
18 from neutron._i18n import _
19 from neutron.agent.linux import async_process
20 from neutron.agent.linux import utils
21 from neutron.tests import base
24 class AsyncProcessTestFramework(base.BaseTestCase):
27 super(AsyncProcessTestFramework, self).setUp()
28 self.test_file_path = self.get_temp_file_path('test_async_process.tmp')
29 self.data = [six.text_type(x) for x in range(4)]
30 with open(self.test_file_path, 'w') as f:
31 f.writelines('%s\n' % item for item in self.data)
33 def _check_stdout(self, proc):
34 # Ensure that all the output from the file is read
36 while output != self.data:
37 new_output = list(proc.iter_stdout())
43 class TestAsyncProcess(AsyncProcessTestFramework):
44 def _safe_stop(self, proc):
47 except async_process.AsyncProcessException:
50 def test_stopping_async_process_lifecycle(self):
51 proc = async_process.AsyncProcess(['tail', '-f',
53 self.addCleanup(self._safe_stop, proc)
54 proc.start(block=True)
55 self._check_stdout(proc)
58 # Ensure that the process and greenthreads have stopped
60 self.assertEqual(proc._process.returncode, -9)
61 for watcher in proc._watchers:
64 def test_async_process_respawns(self):
65 proc = async_process.AsyncProcess(['tail', '-f',
68 self.addCleanup(self._safe_stop, proc)
71 # Ensure that the same output is read twice
72 self._check_stdout(proc)
74 utils.execute(['kill', '-9', pid])
75 utils.wait_until_true(
76 lambda: proc.is_active() and pid != proc.pid,
79 exception=RuntimeError(_("Async process didn't respawn")))
80 self._check_stdout(proc)