cleanups and doc

Author: infothrill

.gitignore

@@ -1,3 +1,5 @@
-dist
 *.egg-info
 __pycache__
+.coverage
+dist
+htmlcov

CHANGELOG.rst

@@ -1,6 +1,6 @@
 Release history
 ---------------
 
-0.1 (October 2013)
-++++++++++++++++++
+0.1 (November 2013)
++++++++++++++++++++
 - Focus: initial public release

README.rst

@@ -59,7 +59,7 @@ Detect if a job exists:
    >>> launchd.LaunchdJob("com.example.fubar").exists()
    False
 
-launchd properties:
+launchd job properties (these come directly from launchd and NOT the .plist files):
 
 .. code-block:: python
 
@@ -79,6 +79,15 @@ Find all plist filenames of currently running jobs:
          continue
       print(job.plistfilename)
 
+Job properties of a given job (this uses the actual .plist file):
+
+.. code-block:: python
+   >>> launchd.plist.read("com.apple.kextd")
+   {'ProgramArguments': ['/usr/libexec/kextd'], 'KeepAlive': {'SuccessfulExit': False},
+   'POSIXSpawnType': 'Interactive', 'MachServices': {'com.apple.KernelExtensionServer':
+   {'HostSpecialPort': 15}}, 'Label': 'com.apple.kextd'}
+
+
 
 Installation
 ============

example.py

@@ -8,43 +8,48 @@
 
 import launchd
 
-myplist = dict(
-          Disabled=False,
-          Label="testlaunchdwrapper_python",
-          Nice=-15,
-          OnDemand=True,
-          ProgramArguments=["/bin/bash", "-c", "sleep 1 && echo 'Hello World' && exit 0"],
-          RunAtLoad=True,
-          ServiceDescription="runs a sample command",
-          ServiceIPC=False,
-          )
 
+def install(label, plist):
+    '''
+    Utility function to store a new .plist file and load it
 
-def install():
-    fname = launchd.plist.write(myplist, myplist['Label'], launchd.plist.USER)
+    :param label: job label
+    :param plist: a property list dictionary
+    '''
+    fname = launchd.plist.write(label, plist)
     launchd.load(fname)
 
 
-def uninstall():
-    fname = launchd.plist.discover_filename(myplist['Label'])
-    job = launchd.LaunchdJob(myplist['Label'])
-    try:
-        job.refresh()
-    except ValueError:
-        pass
-    else:
+def uninstall(label):
+    '''
+    Utility function to remove a .plist file and unload it
+
+    :param label: job label
+    '''
+    if launchd.LaunchdJob(label).exists():
+        fname = launchd.plist.discover_filename(label)
         launchd.unload(fname)
         os.unlink(fname)
 
 
 def main():
+    myplist = dict(
+              Disabled=False,
+              Label="testlaunchdwrapper_python",
+              Nice=-15,
+              OnDemand=True,
+              ProgramArguments=["/bin/bash", "-c", "sleep 1 && echo 'Hello World' && exit 0"],
+              RunAtLoad=True,
+              ServiceDescription="runs a sample command",
+              ServiceIPC=False,
+              )
+
     import time
     label = myplist['Label']
     job = launchd.LaunchdJob(label)
     if not job.exists():
         print("'%s' is not loaded in launchd. Installing..." % (label))
-        install()
-        job.refresh()
+        install(label, myplist)
         while job.pid is not None:
             print("Alive! PID = %s" % job.pid)
             job.refresh()
@@ -60,7 +65,7 @@ def main():
                 time.sleep(0.2)
 
     print("Uninstalling again...")
-    uninstall()
+    uninstall(label)
     return 0
 
 

launchd/cmd.py

@@ -1,71 +1,27 @@
 # -*- coding: utf-8 -*-
 
 import subprocess
-import plistlib
-import six
-
-_LAUNCHCTL_CMD = "launchctl"
-
-
-def jobs_cmd():
-    '''
-    Wrapper for `launchctl list`
-
-    Returns a generator for LaunchdJob
-    '''
-    if six.PY2:
-        stdout = launchctl("list")
-    else:
-        stdout = launchctl("list").decode("utf-8")
-    # PID, Status, Label
-    lines = iter(stdout.splitlines())
-    # skip first line
-    next(lines)
-    sep = "\t"
-    Ox = "0x"
-    for line in lines:
-        pid, _, last = line.strip().partition(sep)
-        status, _, label = last.strip().partition(sep)
-        if label.startswith(Ox):
-            continue
-        if pid.isdigit():
-            pid = int(pid)
-        else:
-            pid = None
-        if status.isdigit():
-            status = int(status)
-        else:
-            status = None
-        yield LaunchdJob(label, pid, status)
 
-
-def job_properties_cmd(joblabel):
-    '''
-    Wrapper for `launchctl -x LABEL`
-
-    Returns dictionary
-    :param job: string label or LaunchdJob
-    '''
-    if six.PY2:
-        return dict(plistlib.readPlistFromString(launchctl("list", "-x", joblabel.encode("utf-8"))))
-    else:
-        return dict(plistlib.readPlistFromBytes(launchctl("list", "-x", joblabel)))
+import six
 
 
 def launchctl(subcommand, *args):
     '''
     A minimal wrapper to call the launchctl binary and capture the output
     :param subcommand: string
     '''
+    if not isinstance(subcommand, six.string_types):
+        raise ValueError("Argument is invalid: %r" % repr(subcommand))
     if isinstance(subcommand, six.text_type):
         subcommand = subcommand.encode('utf-8')
-    else:
-        raise ValueError("Argument is invalid: %r" % repr(subcommand))
-    cmd = [_LAUNCHCTL_CMD, subcommand]
+
+    cmd = ["launchctl", subcommand]
     for arg in args:
-        if isinstance(arg, six.text_type):
-            cmd.append(arg.encode('utf-8'))
+        if isinstance(arg, six.string_types):
+            if isinstance(arg, six.text_type):
+                cmd.append(arg.encode('utf-8'))
+            else:
+                cmd.append(arg)
         else:
             raise ValueError("Argument is invalid: %r" % repr(arg))
-    output = subprocess.check_output(cmd, stdin=None, stderr=subprocess.STDOUT, shell=False)
-    return output
+    return subprocess.check_output(cmd, stdin=None, stderr=subprocess.STDOUT, shell=False)

launchd/launchctl.py

@@ -109,10 +109,10 @@ def jobs():
         except KeyError:
             pid = None
         try:
-            status = int(entry['LastExitStatus'])
+            laststatus = int(entry['LastExitStatus'])
         except KeyError:
-            status = None
-        job = LaunchdJob(label, pid, status)
+            laststatus = None
+        job = LaunchdJob(label, pid, laststatus)
         job._nsproperties = entry
         yield job
 
@@ -128,10 +128,8 @@ def stop(*args):
 
 
 def load(*args):
-    raise NotImplementedError()
     return launchctl("load", *args)
 
 
 def unload(*args):
-    raise NotImplementedError()
     return launchctl("unload", *args)

launchd/plist.py

@@ -3,7 +3,6 @@
 import os
 import plistlib
 
-
 USER = 1
 USER_ADMIN = 2
 DAEMON_ADMIN = 3
@@ -23,11 +22,18 @@ def compute_filename(label, scope):
     return os.path.expanduser(os.path.join(PLIST_LOCATIONS[scope], label + ".plist"))
 
 
-def discover_filename(label, scope=None):
-    if scope is not None:
-        scopes = [scope]
-    else:
+def discover_filename(label, scopes=None):
+    '''
+    Check the filesystem for the existence of a .plist file matching the job label.
+    Optionally specify one or more scopes to search (default all).
+
+    :param label: string
+    :param scope: tuple or list or oneOf(USER, USER_ADMIN, DAEMON_ADMIN, USER_OS, DAEMON_OS)
+    '''
+    if scopes is None:
         scopes = [k for k in PLIST_LOCATIONS]
+    elif not isinstance(scopes, (list, tuple)):
+        scopes = (scopes, )
     for thisscope in scopes:
         plistfilename = compute_filename(label, thisscope)
         if os.path.isfile(plistfilename):
@@ -36,11 +42,11 @@ def discover_filename(label, scope=None):
 
 
 def read(label, scope=None):
-    with open(discover_filename(label, scope)) as f:
+    with open(discover_filename(label, (scope,)), 'rb') as f:
         return plistlib.readPlist(f)
 
 
-def write(plist, label, scope):
+def write(label, plist, scope=USER):
     '''
     Writes the given property list to the appropriate file on disk and returns
     the absolute filename.

launchd/tests/plist.py

@@ -41,3 +41,41 @@ def test_compute_filename(self):
     def test_discover_filename(self):
         fname = plist.discover_filename("com.apple.kextd", plist.DAEMON_OS)
         self.assertTrue(os.path.isfile(fname))
+
+        # no scope specified
+        fname2 = plist.discover_filename("com.apple.kextd")
+        self.assertTrue(os.path.isfile(fname2))
+
+        fname = plist.discover_filename("com.apple.kextd", plist.USER)
+        self.assertEqual(None, fname)
+
+
+class PlistToolPersistencyTest(unittest.TestCase):
+    def setUp(self):
+        self.sample_label = "com.example.unittest"
+        self.sample_props = dict(Label="testlaunchdwrapper_python")
+        fname = plist.discover_filename(self.sample_label, plist.USER)
+        if fname is not None:
+            os.unlink(fname)
+        unittest.TestCase.setUp(self)
+
+    def tearDown(self):
+        fname = plist.discover_filename(self.sample_label, plist.USER)
+        if fname is not None:
+            os.unlink(fname)
+        unittest.TestCase.tearDown(self)
+
+    @unittest.skipUnless(sys.platform.startswith("darwin"), "requires OS X")
+    def test_read_write(self):
+        sample_label = "com.example.unittest"
+        sample_props = dict(Label="testlaunchdwrapper_python")
+
+        fname = plist.discover_filename(sample_label, plist.USER)
+        self.assertEqual(None, fname)
+        plist.write(sample_label, sample_props, plist.USER)
+
+        fname = plist.discover_filename(sample_label, plist.USER)
+        self.assertTrue(os.path.isfile(fname))
+        props = plist.read(sample_label, plist.USER)
+        self.assertEqual(sample_props, props)
+        os.unlink(fname)