Skip to content
Permalink
Browse files

Start writing test for checkin, need to convert launchd structures to…

… python structures to get them between the processes again...

git-svn-id: https://svn.calendarserver.org/repository/calendarserver/CalendarServer/branches/users/glyph/launchd-wrapper-bis@11422 e27351fd-9f3e-4f54-a53b-843176b1656c
  • Loading branch information...
glyph committed Jun 26, 2013
1 parent 909f795 commit 913387046d5c10aa98eb2120c3f20416aadbf3e8
Showing with 46 additions and 6 deletions.
  1. +15 −0 twext/python/launchd.py
  2. +31 −6 twext/python/test/test_launchd.py
@@ -151,6 +151,21 @@ def __len__(self):



def plainPython(x):
"""
Convert a launchd python-like data structure into regular Python
dictionaries and lists.
"""
if isinstance(x, LaunchDictionary):
result = {}
result.update(x)
return result
elif isinstance(x, LaunchArray):
result = []




class LaunchErrno(Exception):
"""
Error from launchd.
@@ -24,14 +24,16 @@
testID = sys.argv[1]
a, b = testID.rsplit(".", 1)
from twisted.python.reflect import namedAny
namedAny(".".join([a, b.replace("test_", "job_")]))()
skt = socket.socket()
skt.connect(("127.0.0.1", int(os.environ["TESTING_PORT"])))
try:
namedAny(".".join([a, b.replace("test_", "job_")]))()
finally:
skt = socket.socket()
skt.connect(("127.0.0.1", int(os.environ["TESTING_PORT"])))
sys.exit(0)


from twext.python.launchd import (lib, ffi, LaunchDictionary, LaunchArray,
_managed, constants)
_managed, constants, plainPython)

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
@@ -98,6 +100,16 @@ def test_launchDictionaryItems(self):
(b"beta", b"beta-value"), (b"gamma", 3)]))


def test_launchDictionaryPlainPython(self):
"""
L{plainPython} will convert a L{LaunchDictionary} into a Python
dictionary.
"""
self.assertEquals({b"alpha": b"alpha-value", b"beta": b"beta-value",
b"gamma": 3},
plainPython(LaunchDictionary(self.testDict)))


class ArrayTests(TestCase):
"""
Tests for L{LaunchArray}
@@ -220,8 +232,8 @@ def goodbyePort():
return d


@classmethod
def job_test(self):
@staticmethod
def job_test():
"""
Do something observable in a subprocess.
"""
@@ -237,6 +249,19 @@ def test_test(self):
self.assertEquals("Sample Value.", self.stdout.getContent())


@staticmethod
def job_checkin():
"""
Check in in the subprocess.
"""
checkin()


def test_checkin(self):
"""
Checking in results
"""


def tearDown(self):
os.spawnlp(os.P_WAIT, "launchctl", "launchctl", "unload", self.job.path)

0 comments on commit 9133870

Please sign in to comment.
You can’t perform that action at this time.