Skip to content

Commit

Permalink
Move tests back. Use explicit pid for tests.
Browse files Browse the repository at this point in the history
  • Loading branch information
adiroiban committed Oct 13, 2023
1 parent cb4f3da commit a9aacc9
Showing 1 changed file with 116 additions and 112 deletions.
228 changes: 116 additions & 112 deletions src/twisted/test/test_twistd.py
Original file line number Diff line number Diff line change
Expand Up @@ -551,6 +551,116 @@ def test_preAndPostApplication(self):
self.assertTrue(s.hadApplicationLogObserver)
self.assertEqual(s.order, ["pre", "log", "post"])

def _applicationStartsWithConfiguredID(self, argv, uid, gid):
"""
Assert that given a particular command line, an application is started
as a particular UID/GID.
@param argv: A list of strings giving the options to parse.
@param uid: An integer giving the expected UID.
@param gid: An integer giving the expected GID.
"""
self.config.parseOptions(argv)

events = []

class FakeUnixApplicationRunner(twistd._SomeApplicationRunner):
def setupEnvironment(self, chroot, rundir, nodaemon, umask, pidfile):
events.append("environment")

def shedPrivileges(self, euid, uid, gid):
events.append(("privileges", euid, uid, gid))

def startReactor(self, reactor, oldstdout, oldstderr):
events.append("reactor")

def removePID(self, pidfile):
pass

@implementer(service.IService, service.IProcess)
class FakeService:
parent = None
running = None
name = None
processName = None
uid = None
gid = None

def setName(self, name):
pass

def setServiceParent(self, parent):
pass

def disownServiceParent(self):
pass

def privilegedStartService(self):
events.append("privilegedStartService")

def startService(self):
events.append("startService")

def stopService(self):
pass

application = FakeService()
verifyObject(service.IService, application)
verifyObject(service.IProcess, application)

runner = FakeUnixApplicationRunner(self.config)
runner.preApplication()
runner.application = application
runner.postApplication()

self.assertEqual(
events,
[
"environment",
"privilegedStartService",
("privileges", False, uid, gid),
"startService",
"reactor",
],
)

@skipIf(
not getattr(os, "setuid", None),
"Platform does not support --uid/--gid twistd options.",
)
def test_applicationStartsWithConfiguredNumericIDs(self):
"""
L{postApplication} should change the UID and GID to the values
specified as numeric strings by the configuration after running
L{service.IService.privilegedStartService} and before running
L{service.IService.startService}.
"""
uid = 1234
gid = 4321
self._applicationStartsWithConfiguredID(
["--uid", str(uid), "--gid", str(gid)], uid, gid
)

@skipIf(
not getattr(os, "setuid", None),
"Platform does not support --uid/--gid twistd options.",
)
def test_applicationStartsWithConfiguredNameIDs(self):
"""
L{postApplication} should change the UID and GID to the values
specified as user and group names by the configuration after running
L{service.IService.privilegedStartService} and before running
L{service.IService.startService}.
"""
user = "foo"
uid = 1234
group = "bar"
gid = 4321
patchUserDatabase(self.patch, user, uid, group, gid)
self._applicationStartsWithConfiguredID(
["--uid", user, "--gid", group], uid, gid
)

def test_startReactorRunsTheReactor(self):
"""
L{startReactor} calls L{reactor.run}.
Expand Down Expand Up @@ -862,7 +972,7 @@ def switchUIDFail(uid, gid, euid):
exc = self.assertRaises(SystemExit, runner.shedPrivileges, 35, 200, None)
self.assertEqual(exc.code, 1)

def _setUID(self, wantedUser, wantedUid, wantedGroup, wantedGid):
def _setUID(self, wantedUser, wantedUid, wantedGroup, wantedGid, pidFile):
"""
Common code for tests which try to pass the the UID to
L{UnixApplicationRunner}.
Expand All @@ -885,7 +995,7 @@ def setgid(gid):

options = twistd.ServerOptions()
options.parseOptions(
["--nodaemon", "--uid", str(wantedUid), "--pidfile=setuid.pid"]
["--nodaemon", "--uid", str(wantedUid), "--pidfile={}".format(pidFile)]
)
application = service.Application("test_setupEnvironment")
self.runner = UnixApplicationRunner(options)
Expand All @@ -898,15 +1008,17 @@ def test_setUidWithoutGid(self):
with a UID and no GUID will result in the GUID being
set to the default GUID for that UID.
"""
self._setUID("foo", 5151, "bar", 4242)
self._setUID("foo", 5151, "bar", 4242, "test_setUidWithoutGid.pid")

def test_setUidSameAsCurrentUid(self):
"""
If the specified UID is the same as the current UID of the process,
then a warning is displayed.
"""
currentUid = os.getuid()
self._setUID("morefoo", currentUid, "morebar", 4343)
self._setUID(
"morefoo", currentUid, "morebar", 4343, "test_setUidSameAsCurrentUid.pid"
)

warningsShown = self.flushWarnings()
expectedWarning = (
Expand All @@ -916,114 +1028,6 @@ def test_setUidSameAsCurrentUid(self):
self.assertEqual(expectedWarning, warningsShown[0]["message"])
self.assertEqual(1, len(warningsShown), warningsShown)

def _applicationStartsWithConfiguredID(self, argv, uid, gid):
"""
Assert that given a particular command line, an application is started
as a particular UID/GID.
@param argv: A list of strings giving the options to parse.
@param uid: An integer giving the expected UID.
@param gid: An integer giving the expected GID.
"""
config = twistd.ServerOptions()
self.serviceMaker = MockServiceMaker()
# Set up a config object like it's been parsed with a subcommand
config.loadedPlugins = {"test_command": self.serviceMaker}
config.subOptions = object()
config.subCommand = "test_command"
config.parseOptions(argv)

events = []

class FakeUnixApplicationRunner(twistd._SomeApplicationRunner):
def setupEnvironment(self, chroot, rundir, nodaemon, umask, pidfile):
events.append("environment")

def shedPrivileges(self, euid, uid, gid):
events.append(("privileges", euid, uid, gid))

def startReactor(self, reactor, oldstdout, oldstderr):
events.append("reactor")

def removePID(self, pidfile):
pass

@implementer(service.IService, service.IProcess)
class FakeService:
parent = None
running = None
name = None
processName = None
uid = None
gid = None

def setName(self, name):
pass

def setServiceParent(self, parent):
pass

def disownServiceParent(self):
pass

def privilegedStartService(self):
events.append("privilegedStartService")

def startService(self):
events.append("startService")

def stopService(self):
pass

application = FakeService()
verifyObject(service.IService, application)
verifyObject(service.IProcess, application)

runner = FakeUnixApplicationRunner(config)
runner.preApplication()
runner.application = application
runner.postApplication()

self.assertEqual(
events,
[
"environment",
"privilegedStartService",
("privileges", False, uid, gid),
"startService",
"reactor",
],
)

def test_applicationStartsWithConfiguredNameIDs(self):
"""
L{postApplication} should change the UID and GID to the values
specified as user and group names by the configuration after running
L{service.IService.privilegedStartService} and before running
L{service.IService.startService}.
"""
user = "foo"
uid = 1234
group = "bar"
gid = 4321
patchUserDatabase(self.patch, user, uid, group, gid)
self._applicationStartsWithConfiguredID(
["--uid", user, "--gid", group], uid, gid
)

def test_applicationStartsWithConfiguredNumericIDs(self):
"""
L{postApplication} should change the UID and GID to the values
specified as numeric strings by the configuration after running
L{service.IService.privilegedStartService} and before running
L{service.IService.startService}.
"""
uid = 1234
gid = 4321
self._applicationStartsWithConfiguredID(
["--uid", str(uid), "--gid", str(gid)], uid, gid
)


@skipIf(not _twistd_unix, "twistd unix not available")
class UnixApplicationRunnerRemovePIDTests(TestCase):
Expand Down

0 comments on commit a9aacc9

Please sign in to comment.