[CalendarServer-changes] [11415] CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/ python
source_changes at macosforge.org
source_changes at macosforge.org
Wed Jun 26 01:14:52 PDT 2013
Revision: 11415
http://trac.calendarserver.org//changeset/11415
Author: glyph at apple.com
Date: 2013-06-26 01:14:52 -0700 (Wed, 26 Jun 2013)
Log Message:
-----------
Add some unit tests, some bindings for allocation functions to create some test data.
Modified Paths:
--------------
CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/launchd.py
Added Paths:
-----------
CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/test/test_launchd.py
Modified: CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/launchd.py
===================================================================
--- CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/launchd.py 2013-06-26 08:14:39 UTC (rev 11414)
+++ CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/launchd.py 2013-06-26 08:14:52 UTC (rev 11415)
@@ -1,3 +1,4 @@
+# -*- test-case-name: twext.python.test.test_launchd -*-
##
# Copyright (c) 2013 Apple Inc. All rights reserved.
#
@@ -17,3 +18,149 @@
"""
CFFI bindings for launchd check-in API.
"""
+
+from __future__ import print_function
+
+from cffi import FFI
+
+ffi = FFI()
+
+ffi.cdef("""
+
+static const char* LAUNCH_KEY_CHECKIN;
+static const char* LAUNCH_JOBKEY_LABEL;
+static const char* LAUNCH_JOBKEY_SOCKETS;
+
+typedef enum {
+ LAUNCH_DATA_DICTIONARY = 1,
+ LAUNCH_DATA_ARRAY,
+ LAUNCH_DATA_FD,
+ LAUNCH_DATA_INTEGER,
+ LAUNCH_DATA_REAL,
+ LAUNCH_DATA_BOOL,
+ LAUNCH_DATA_STRING,
+ LAUNCH_DATA_OPAQUE,
+ LAUNCH_DATA_ERRNO,
+ LAUNCH_DATA_MACHPORT,
+} launch_data_type_t;
+
+typedef struct _launch_data *launch_data_t;
+
+bool launch_data_dict_insert(launch_data_t, const launch_data_t, const char *);
+
+launch_data_t launch_data_alloc(launch_data_type_t);
+launch_data_t launch_data_new_string(const char *);
+launch_data_t launch_msg(const launch_data_t);
+launch_data_type_t launch_data_get_type(const launch_data_t);
+launch_data_t launch_data_dict_lookup(const launch_data_t, const char *);
+size_t launch_data_dict_get_count(const launch_data_t);
+
+const char * launch_data_get_string(const launch_data_t);
+
+size_t launch_data_array_get_count(const launch_data_t);
+launch_data_t launch_data_array_get_index(const launch_data_t, size_t);
+
+void launch_data_free(launch_data_t);
+""")
+
+lib = ffi.verify("""
+#include <launch.h>
+""")
+
+class NullPointerException(Exception):
+ """
+ Python doesn't have one of these.
+ """
+
+
+class LaunchArray(object):
+ def __init__(self, launchdata):
+ self.launchdata = launchdata
+
+
+ def __len__(self):
+ return lib.launch_data_array_get_count(self.launchdata)
+
+
+ def __getitem__(self, index):
+ return lib.launch_data_array_get_index(self.launchdata, index)
+
+
+
+class LaunchDictionary(object):
+ def __init__(self, launchdata):
+ self.launchdata = launchdata
+
+
+ def keys(self):
+ """
+ Keys.
+ """
+ return []
+
+
+ def __getitem__(self, key):
+ launchvalue = lib.launch_data_dict_lookup(self.launchdata, key)
+ try:
+ return _launchify(launchvalue)
+ except LaunchErrno:
+ raise KeyError(key)
+
+
+ def __len__(self):
+ return lib.launch_data_dict_get_count(self.launchdata)
+
+
+
+class LaunchErrno(Exception):
+ """
+ Error from launchd.
+ """
+
+
+
+def _launchify(launchvalue):
+ dtype = lib.launch_data_get_type(launchvalue)
+ if dtype == lib.LAUNCH_DATA_ERRNO:
+ raise LaunchErrno(launchvalue)
+ elif dtype == lib.LAUNCH_DATA_STRING:
+ cvalue = lib.launch_data_get_string(launchvalue)
+ pybytes = ffi.string(cvalue)
+ pyunicode = pybytes.decode('utf-8')
+ return pyunicode
+ elif dtype == lib.LAUNCH_DATA_ARRAY:
+ return LaunchArray(launchvalue)
+ elif dtype == lib.LAUNCH_DATA_DICTIONARY:
+ return LaunchDictionary(launchvalue)
+ elif dtype in lib.LAUNCH_DATA_FD:
+ return lib.launch_data_get_fd(launchvalue)
+
+import sys
+
+def getLaunchDSocketFDs():
+ result = {}
+ req = lib.launch_data_new_string(lib.LAUNCH_KEY_CHECKIN)
+ if req == ffi.NULL:
+ # Good luck reproducing this case.
+ raise NullPointerException()
+ response = lib.launch_msg(req)
+ if response == ffi.NULL:
+ raise NullPointerException()
+ if lib.launch_data_get_type(response) == lib.LAUNCH_DATA_ERRNO:
+ raise NullPointerException()
+ response = LaunchDictionary(response)
+ label = response[lib.LAUNCH_JOBKEY_LABEL]
+ skts = response[lib.LAUNCH_JOBKEY_SOCKETS]
+ result['label'] = label
+ result['sockets'] = list(skts['TestSocket'])
+ return result
+
+if __name__ == '__main__':
+ # Unit tests :-(
+ import traceback
+ try:
+ print(getLaunchDSocketFDs())
+ except:
+ traceback.print_exc()
+ sys.stdout.flush()
+ sys.stderr.flush()
Added: CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/test/test_launchd.py
===================================================================
--- CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/test/test_launchd.py (rev 0)
+++ CalendarServer/branches/users/glyph/launchd-wrapper-bis/twext/python/test/test_launchd.py 2013-06-26 08:14:52 UTC (rev 11415)
@@ -0,0 +1,58 @@
+##
+# Copyright (c) 2013 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}.
+"""
+
+from twext.python.launchd import lib, ffi, LaunchDictionary
+
+from twisted.trial.unittest import TestCase
+
+alloc = lib.launch_data_alloc
+
+
+
+class WrapperTests(TestCase):
+ """
+ Tests for all wrapper objects.
+ """
+
+ def setUp(self):
+ """
+ Assemble some test data structures.
+ """
+ self.testDict = ffi.gc(
+ lib.launch_data_alloc(lib.LAUNCH_DATA_DICTIONARY),
+ lib.launch_data_free
+ )
+ key1 = lib.launch_data_new_string("alpha")
+ val1 = lib.launch_data_new_string("alpha-value")
+ key2 = lib.launch_data_new_string("beta")
+ val2 = lib.launch_data_new_string("beta-value")
+ lib.launch_data_dict_insert(self.testDict, key1, val1)
+ lib.launch_data_dict_insert(self.testDict, key2, val2)
+
+
+ def test_launchDictionaryKeys(self):
+ """
+ L{LaunchDictionary.keys} returns a key.
+ """
+ dictionary = LaunchDictionary(self.testDict)
+ self.assertEquals(dictionary.keys(), [u"alpha", u"beta"])
+
+
+
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20130626/60396813/attachment-0001.html>
More information about the calendarserver-changes
mailing list