@@ -0,0 +1,99 @@
1+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
2+#
3+# Unity Autopilot Test Suite
4+# Copyright (C) 2013 Canonical
5+#
6+# This program is free software: you can redistribute it and/or modify
7+# it under the terms of the GNU General Public License as published by
8+# the Free Software Foundation, either version 3 of the License, or
9+# (at your option) any later version.
10+#
11+# This program is distributed in the hope that it will be useful,
12+# but WITHOUT ANY WARRANTY; without even the implied warranty of
13+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14+# GNU General Public License for more details.
15+#
16+# You should have received a copy of the GNU General Public License
17+# along with this program. If not, see <http://www.gnu.org/licenses/>.
18+#
19+
20+"""Tests for upstart integration."""
21+
22+import os
23+import signal
24+import subprocess
25+import time
26+
27+from testtools.matchers._basic import Equals
28+from autopilot.matchers import Eventually
29+from autopilot.introspection import get_proxy_object_for_existing_process
30+
31+from unity8 import get_binary_path
32+from unity8.shell.emulators import UnityEmulatorBase
33+from unity8.shell.tests import UnityTestCase, _get_device_emulation_scenarios
34+
35+import logging
36+
37+logger = logging.getLogger(__name__)
38+
39+
40+class UpstartIntegrationTests(UnityTestCase):
41+
42+ scenarios = _get_device_emulation_scenarios()
43+
44+ def _get_status(self):
45+ pid, status = os.waitpid(self.process.pid, os.WUNTRACED | os.WCONTINUED | os.WNOHANG)
46+ logger.debug("Got status: {0}; stopped: {1}; continued: {2}"\
47+ .format(status, os.WIFSTOPPED(status), os.WIFCONTINUED(status)))
48+ return status
49+
50+ def _launch_unity(self):
51+ self.patch_environment("QT_LOAD_TESTABILITY", "1")
52+ self.process = subprocess.Popen([get_binary_path()] + self.unity_geometry_args)
53+ def ensure_stopped():
54+ self.process.terminate()
55+ for i in xrange(10):
56+ try:
57+ self._get_status()
58+ except OSError:
59+ break
60+ else:
61+ time.sleep(1)
62+ try:
63+ self._get_status()
64+ except OSError:
65+ pass
66+ else:
67+ self.process.kill()
68+
69+ self.process.wait()
70+
71+ self.addCleanup(ensure_stopped)
72+
73+ def _set_proxy(self):
74+ super(UpstartIntegrationTests, self)._set_proxy(get_proxy_object_for_existing_process(
75+ pid=self.process.pid,
76+ emulator_base=UnityEmulatorBase,
77+ ))
78+
79+ def test_no_sigstop(self):
80+ self.patch_environment("UPSTART_JOB", "foo")
81+ self._launch_unity()
82+ self._set_proxy()
83+
84+ logger.debug("Unity started, waiting for it to be ready.")
85+ self.assertUnityReady()
86+ logger.debug("Unity loaded and ready.")
87+
88+ def test_expect_sigstop(self):
89+ self.patch_environment("UPSTART_JOB", "unity8")
90+ self._launch_unity()
91+ self.assertThat(lambda: os.WIFSTOPPED(self._get_status()), Eventually(Equals(True)), "Unity8 should raise SIGSTOP when ready")
92+
93+ self.process.send_signal(signal.SIGCONT)
94+ self.assertThat(lambda: os.WIFCONTINUED(self._get_status()), Eventually(Equals(True)), "Unity8 should have resumed")
95+
96+ logger.debug("Unity started, waiting for it to be ready.")
97+ self._set_proxy()
98+ self.assertUnityReady()
99+ logger.debug("Unity loaded and ready.")