2
# -*- coding: utf-8 -*-
3
#---------------------------------------------------------------------
4
# Copyright © 2013 Canonical Ltd.
6
# Author: James Hunt <james.hunt@canonical.com>
8
# This program is free software; you can redistribute it and/or modify
9
# it under the terms of the GNU General Public License version 2, as
10
# published by the Free Software Foundation.
12
# This program is distributed in the hope that it will be useful,
13
# but WITHOUT ANY WARRANTY; without even the implied warranty of
14
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15
# GNU General Public License for more details.
17
# You should have received a copy of the GNU General Public License along
18
# with this program; if not, write to the Free Software Foundation, Inc.,
19
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
20
#---------------------------------------------------------------------
22
#---------------------------------------------------------------------
23
# Description: System-level Upstart tests for the pyupstart module.
25
# Notes: Can only be run as the root user.
26
#---------------------------------------------------------------------
31
base_dir = os.path.abspath(os.path.dirname(__file__))
32
module_dir = os.path.normpath(os.path.realpath(base_dir + os.sep + '..'))
34
# top-level unpacked source directory
35
top_srcdir = os.path.normpath(os.path.realpath(module_dir + os.sep + '..'))
37
# Tell Python where the uninstalled module lives in the source tree
38
sys.path.append(module_dir)
39
from pyupstart import *
43
class TestSystemUpstart(unittest.TestCase):
46
raise unittest.SkipTest('Need root for System-level Upstart tests')
48
# Tests must not operate within a session
49
self.assertEqual(None, os.environ.get('UPSTART_SESSION', None))
51
self.upstart = SystemInit()
54
# Ensure no state file exists
55
state_file = '{}{}{}'.format(SYSTEM_LOG_DIR, os.sep, UPSTART_STATE_FILE)
56
self.assertFalse(os.path.exists(state_file))
58
class TestSystemInitReExec(TestSystemUpstart):
60
def test_pid1_reexec(self):
61
version = self.upstart.version()
62
self.assertTrue(version)
64
# create a job and start it, marking it such that the .conf file
65
# will be retained when object becomes unusable (after re-exec).
66
job = self.upstart.job_create('sleeper', 'exec sleep 123')
72
self.assertEqual(len(pids), 1)
77
# PID 1 Upstart is now in the process of starting, but we need to
78
# reconnect to it via D-Bus since it cannot yet retain client
79
# connections. However, since the re-exec won't be instantaneous,
81
self.upstart.polling_connect(force=True)
83
# check that we can still operate on the re-exec'd Upstart
84
version_postexec = self.upstart.version()
85
self.assertTrue(version_postexec)
86
self.assertEqual(version, version_postexec)
88
# Ensure the job is still running with the same PID
91
self.assertTrue(job.running('_'))
94
self.assertEqual(len(pids), 1)
95
self.assertTrue(pids['main'])
97
# Ensure pid remains the same
98
self.assertEqual(pid, pids['main'])
100
# Exceptions will be caught by the unittest framework
106
self.assertEqual(len(pids), 1)
107
self.assertTrue(pids['main'])
109
os.kill(pids['main'], 0)
110
self.assertTrue(job.running('_'))
112
# The pid should have changed after a restart
113
self.assertNotEqual(pid, pids['main'])
118
self.upstart.destroy()
120
class TestSystemInitChrootSession(TestSystemUpstart):
121
CHROOT_ENVVAR = 'UPSTART_TEST_CHROOT_PATH'
123
def test_chroot_session_reexec(self):
124
chroot_path = os.environ.get(self.CHROOT_ENVVAR, None)
127
raise unittest.SkipTest('{} variable not set'.format(self.CHROOT_ENVVAR))
129
# Ensure the chroot exists
130
self.assertTrue(os.path.exists(chroot_path))
132
# Ensure Upstart is installed in the chroot
133
chroot_initctl = '{}{}{}'.format(chroot_path, os.sep, INITCTL)
134
self.assertTrue(os.path.exists(chroot_initctl))
136
# No sessions should exist before the test starts
137
self.assertFalse(self.upstart.sessions_exist())
139
# Create an Upstart chroot session by talking from the chroot
141
ret = subprocess.call(['chroot', chroot_path, INITCTL, 'list'])
142
self.assertEqual(0, ret)
144
# Ensure a session now exists
145
self.assertTrue(self.upstart.sessions_exist())
148
self.upstart.reexec()
150
# Ensure Upstart responds
151
self.upstart.polling_connect(force=True)
152
self.assertTrue(self.upstart.version())
164
kwargs['format'] = format
166
# We want to see what's happening
167
kwargs['level'] = logging.DEBUG
169
logging.basicConfig(**kwargs)
172
testRunner=unittest.TextTestRunner(
180
if __name__ == '__main__':
added
removed
scripts/tests/test_pyupstart_system_init.py
