/usr/lib/python2.7/dist-packages/twext/python/test/test_launchd.py is in python-twext 0.1.b2.dev15059-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | ##
# Copyright (c) 2013-2015 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##
"""
Tests for L{twext.python.launchd}.
"""
import json
import os
import plistlib
import socket
import sys
if __name__ == '__main__':
# This module is loaded as a launchd job by test-cases below; the following
# code looks up an appropriate function to run.
testID = sys.argv[1]
a, b = testID.rsplit(".", 1)
from twisted.python.reflect import namedAny
try:
namedAny(".".join([a, b.replace("test_", "job_")]))()
finally:
sys.stdout.flush()
sys.stderr.flush()
skt = socket.socket()
skt.connect(("127.0.0.1", int(os.environ["TESTING_PORT"])))
sys.exit(0)
try:
from twext.python.launchd import launchActivateSocket
except ImportError:
skip = "LaunchD not available."
else:
skip = False
from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
class CheckInTests(TestCase):
"""
Integration tests making sure that actual checkin with launchd results in
the expected values.
"""
def setUp(self):
fp = FilePath(self.mktemp())
fp.makedirs()
from twisted.internet.protocol import Protocol, Factory
from twisted.internet import reactor, defer
d = defer.Deferred()
class JustLetMeMoveOn(Protocol):
def connectionMade(self):
d.callback(None)
self.transport.abortConnection()
f = Factory()
f.protocol = JustLetMeMoveOn
port = reactor.listenTCP(0, f, interface="127.0.0.1")
@self.addCleanup
def goodbyePort():
return port.stopListening()
env = dict(os.environ)
env["TESTING_PORT"] = repr(port.getHost().port)
self.stdout = fp.child("stdout.txt")
self.stderr = fp.child("stderr.txt")
self.launchLabel = ("org.calendarserver.UNIT-TESTS." +
str(os.getpid()) + "." + self.id())
plist = {
"Label": self.launchLabel,
"ProgramArguments": [sys.executable, "-m", __name__, self.id()],
"EnvironmentVariables": env,
"KeepAlive": False,
"StandardOutPath": self.stdout.path,
"StandardErrorPath": self.stderr.path,
"Sockets": {
"Awesome": [{"SecureSocketWithKey": "GeneratedSocket"}]
},
"RunAtLoad": True,
}
self.job = fp.child("job.plist")
self.job.setContent(plistlib.writePlistToString(plist))
os.spawnlp(os.P_WAIT, "launchctl", "launchctl", "load", self.job.path)
return d
@staticmethod
def job_test():
"""
Do something observable in a subprocess.
"""
sys.stdout.write("Sample Value.")
sys.stdout.flush()
def test_test(self):
"""
Since this test framework is somewhat finicky, let's just make sure
that a test can complete.
"""
self.assertEquals("Sample Value.", self.stdout.getContent())
@staticmethod
def job_getFDs():
"""
Check-in via the high-level C{getLaunchDSocketFDs} API, that just gives
us listening FDs.
"""
sys.stdout.write(json.dumps(launchActivateSocket("Awesome")))
def test_getFDs(self):
"""
L{getLaunchDSocketFDs} returns a Python dictionary mapping the names of
sockets specified in the property list to lists of integers
representing FDs.
"""
sockets = json.loads(self.stdout.getContent())
self.assertEquals(len(sockets), 1)
self.assertIsInstance(sockets[0], int)
def tearDown(self):
"""
Un-load the launchd job and report any errors it encountered.
"""
os.spawnlp(os.P_WAIT, "launchctl",
"launchctl", "unload", self.job.path)
err = self.stderr.getContent()
if 'Traceback' in err:
self.fail(err)
|