Don't catch SystemExit from a subprocess

Now an entire process tree is killed in /process/remove, as long as
each child hasn't changed its process group.

.coveragerc

@@ -0,0 +1,10 @@
+# -*- conf -*-
+
+[run]
+# branch = True
+
+[report]
+exclude_lines =
+	pragma: no cover
+	if 0:
+omit = scripts,nilmrun/_version.py,nilmrun/filters/*

.gitignore

@@ -1,3 +1,4 @@
+.coverage
 build/
 *.pyc
 dist/

Makefile

@@ -1,18 +1,15 @@
-URL="http://localhost/nilmdb"
+# By default, run the tests.
+all: test
 
-all:
+test2:
-ifeq ($(INSIDE_EMACS), t)
+	nilmrun/trainola.py data.js
-	@make test
-else
-	@echo "Try 'make install'"
-endif
-
-test:
-	echo No tests yet
 
 version:
 	python setup.py version
 
+build:
+	python setup.py build_ext --inplace
+
 dist: sdist
 sdist:
 	python setup.py sdist
@@ -23,11 +20,29 @@ install:
 develop:
 	python setup.py develop
 
+docs:
+	make -C docs
+
+lint:
+	pylint --rcfile=.pylintrc nilmdb
+
+test:
+ifeq ($(INSIDE_EMACS), t)
+# Use the slightly more flexible script
+	python setup.py build_ext --inplace
+	python tests/runtests.py
+else
+# Let setup.py check dependencies, build stuff, and run the test
+	python setup.py nosetests
+endif
+
 clean::
+	rm -f .coverage
 	find . -name '*pyc' | xargs rm -f
 	rm -rf nilmtools.egg-info/ build/ MANIFEST.in
+	make -C docs clean
 
 gitclean::
 	git clean -dXf
 
-.PHONY: all test version dist sdist install clean gitclean
+.PHONY: all version dist sdist install docs lint test clean gitclean

data.js

@@ -0,0 +1,28 @@
+{ "url": "http://bucket.mit.edu/nilmdb",
+  "stream": "/sharon/prep-a",
+  "start": 1366111383280463,
+  "end": 1366126163457797,
+  "columns": [ { "name": "P1", "index": 0 },
+               { "name": "Q1", "index": 1 },
+               { "name": "P3", "index": 2 } ],
+  "exemplars": [
+      { "name": "Boiler Pump ON",
+        "url": "http://bucket.mit.edu/nilmdb",
+        "stream": "/sharon/prep-a",
+        "start": 1366260494269078,
+        "end": 1366260608185031,
+        "columns": [ { "name": "P1", "index": 0 },
+                     { "name": "Q1", "index": 1 }
+                   ]
+      },
+      { "name": "Boiler Pump OFF",
+        "url": "http://bucket.mit.edu/nilmdb",
+        "stream": "/sharon/prep-a",
+        "start": 1366260864215764,
+        "end": 1366260870882998,
+        "columns": [ { "name": "P1", "index": 0 },
+                     { "name": "Q1", "index": 1 }
+                   ]
+      }
+  ]
+}

docs/wsgi.md

@@ -13,9 +13,13 @@ First, create a WSGI script `/home/nilm/nilmrun.wsgi` containing:
 
 The first parameter is the path part of the URL.
 
-Then, set up Apache with a configuration like:
+Then, set up Apache with a configuration like below.  SSL and access
+control/authentication are strongly recommended since this can execute
+arbitrary commands.
+
+    <VirtualHost *:443>
+        SSLEngine On
 
-    <VirtualHost>
         WSGIScriptAlias /nilmrun /home/nilm/nilmrun.wsgi
         WSGIApplicationGroup nilmrun-appgroup
         WSGIProcessGroup nilmrun-procgroup
@@ -23,6 +27,7 @@ Then, set up Apache with a configuration like:
 
         # Access control example:
         <Location /nilmrun>
+            SSLRequireSSL
             Order deny,allow
             Deny from all
             Allow from 1.2.3.4

nilmrun/__init__.py

@@ -1,3 +1,4 @@
+import nilmrun.processmanager
 
 from ._version import get_versions
 __version__ = get_versions()['version']

nilmrun/_version.py

@@ -181,7 +181,7 @@ def versions_from_parentdir(parentdir_prefix, versionfile_source, verbose=False)
 
 tag_prefix = "nilmrun-"
 parentdir_prefix = "nilmrun-"
-versionfile_source = "src/_version.py"
+versionfile_source = "nilmrun/_version.py"
 
 def get_versions(default={"version": "unknown", "full": ""}, verbose=False):
     variables = { "refnames": git_refnames, "full": git_full }

nilmrun/processmanager.py

@@ -0,0 +1,247 @@
+#!/usr/bin/python
+
+from nilmdb.utils.printf import *
+
+import threading
+import multiprocessing
+import cStringIO
+import sys
+import os
+import signal
+import time
+import uuid
+import psutil
+import imp
+import traceback
+
+class LogReceiver(object):
+    """Spawn a thread that listens to a pipe for log messages,
+    and stores them locally."""
+    def __init__(self, pipe):
+        self.pipe = pipe
+        self.log = cStringIO.StringIO()
+        self.thread = threading.Thread(target = self.run)
+        self.thread.start()
+
+    def run(self):
+        while True:
+            data = os.read(self.pipe, 65536)
+            if not data:
+                os.close(self.pipe)
+                return
+            self.log.write(data)
+
+    def getvalue(self):
+        return self.log.getvalue()
+
+    def clear(self):
+        self.log = cStringIO.StringIO()
+
+class Process(object):
+    """Spawn and manage a process that calls a Python function"""
+    def __init__(self, name, function, parameters):
+        self.parameters = parameters
+        self.start_time = None
+        self.name = name
+
+        # Use a pipe for communicating log data
+        (rpipe, wpipe) = os.pipe()
+        self._log = LogReceiver(rpipe)
+
+        # Start the function in a new process
+        self._process = multiprocessing.Process(
+            target = self._trampoline, name = name,
+            args = (function, rpipe, wpipe, parameters))
+        self._process.daemon = True
+        self._process.start()
+
+        # Close the writer end of the pipe, get process info
+        os.close(wpipe)
+        self.start_time = time.time()
+        self.pid = str(uuid.uuid1(self._process.pid or 0))
+
+    def _trampoline(self, func, rpipe, wpipe, param): # pragma: no cover
+        # No coverage report for this, because it's executed in a subprocess
+        """Trampoline function to set up stdio and call the real function."""
+        # Close the reader end of the pipe
+        os.close(rpipe)
+
+        # Like os.close() but ignores errors
+        def tryclose(fd):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+
+        # Remap stdio to go to the pipe.  We do this at the OS level,
+        # replacing FDs, so that future spawned processes do the right thing.
+
+        # stdin
+        sys.stdin.close()
+        tryclose(0)
+        fd = os.open(os.devnull, os.O_RDONLY) # 0
+        sys.stdin = os.fdopen(fd, 'r', 0)
+
+        # stdout
+        sys.stdout.close()
+        tryclose(1)
+        fd = os.dup(wpipe) # 1
+        sys.stdout = os.fdopen(fd, 'w', 0)
+
+        # stdout
+        sys.stderr.close()
+        tryclose(2)
+        fd = os.dup(wpipe) # 2
+        sys.stderr = os.fdopen(fd, 'w', 0)
+
+        # Don't need this extra fd
+        os.close(wpipe)
+
+        # Ready to go -- call the function, exit when it's done
+        func(param)
+        sys.exit(0)
+
+    def terminate(self, timeout = 1.0):
+        """Terminate a process, and all of its children that are in the same
+        process group."""
+        # First give it some time to die on its own
+        self._process.join(timeout)
+        if not self.alive:
+            return True
+
+        def getpgid(pid):
+            try:
+                return os.getpgid(pid)
+            except OSError: # pragma: no cover
+                return None
+
+        # Find all children
+        group = getpgid(self._process.pid)
+        main = psutil.Process(self._process.pid)
+        allproc = [ main ] + main.get_children(recursive = True)
+
+        # Kill with SIGTERM, if they're still in this process group
+        for proc in allproc:
+            if getpgid(proc.pid) == group:
+                os.kill(proc.pid, signal.SIGTERM)
+
+        # Wait for it to die again
+        self._process.join(timeout)
+        if not self.alive:
+            return True
+
+        # One more try with SIGKILL
+        for proc in allproc:
+            if getpgid(proc.pid) == group:
+                os.kill(proc.pid, signal.SIGKILL)
+
+        # See if it worked
+        self._process.join(timeout)
+        return not self.alive
+
+    def clear_log(self):
+        self._log.clear()
+
+    @property
+    def log(self):
+        return self._log.getvalue()
+
+    @property
+    def alive(self):
+        return self._process.is_alive()
+
+    @property
+    def exitcode(self):
+        return self._process.exitcode
+
+def _exec_user_code(codeargs): # pragma: no cover (runs in subprocess)
+    """Execute 'code' as if it were placed into a file and executed"""
+    (code, args) = codeargs
+    # This is split off into a separate function because the Python3
+    # syntax of "exec" triggers a SyntaxError in Python2, if it's within
+    # a nested function.
+    imp.acquire_lock()
+    try:
+        module = imp.new_module("__main__")
+    finally:
+        imp.release_lock()
+    module.__file__ = "<user-code>"
+    sys.argv = [''] + args
+    # Wrap the compile and exec in a try/except so we can format the
+    # exception more nicely.
+    try:
+        codeobj = compile(code, '<user-code>', 'exec',
+                          flags = 0, dont_inherit = 1)
+        exec(codeobj, module.__dict__, {})
+    except Exception:
+        try:
+            # Pull out the exception
+            info = sys.exc_info()
+            tblist = traceback.extract_tb(info[2])
+
+            # First entry is probably this code; get rid of it
+            if len(tblist) and tblist[0][2] == '_exec_user_code':
+                tblist = tblist[1:]
+
+            # Add the user's source code to every line that's missing it
+            lines = code.splitlines()
+            maxline = len(lines)
+            for (n, (name, line, func, text)) in enumerate(tblist):
+                if name == '<user-code>' and text is None and line <= maxline:
+                    tblist[n] = (name, line, func, lines[line-1].strip())
+
+            # Format it in the usual manner
+            out = ['Traceback (most recent call last):\n']
+            out.extend(traceback.format_list(tblist))
+            out.extend(traceback.format_exception_only(info[0], info[1]))
+        finally:
+            # Need to explicitly delete traceback object to avoid ref cycle
+            del info
+        sys.stderr.write("".join(out))
+        sys.stderr.flush()
+        sys.exit(1)
+    sys.exit(0)
+
+class ProcessManager(object):
+    """Track and manage a collection of Process objects"""
+    def __init__(self):
+        self.processes = {}
+
+    def __iter__(self):
+        return iter(self.processes.keys())
+
+    def __getitem__(self, key):
+        return self.processes[key]
+
+    def run_function(self, procname, function, parameters):
+        """Run a Python function that already exists"""
+        new = Process(procname, function, parameters)
+        self.processes[new.pid] = new
+        return new.pid
+
+    def run_code(self, procname, code, args):
+        """Evaluate 'code' as if it were placed into a Python file and
+        executed.  The arguments will be accessible in the code as
+        sys.argv[1:]."""
+        return self.run_function(procname, _exec_user_code, (code, args))
+
+    def run_command(self, procname, argv):
+        """Execute a command line program"""
+        def spwan_user_command(argv): # pragma: no cover (runs in subprocess)
+            try:
+                maxfd = os.sysconf("SC_OPEN_MAX")
+            except Exception:
+                maxfd = 256
+            os.closerange(3, maxfd)
+            try:
+                os.chdir("/tmp")
+            except OSError:
+                pass
+            os.execvp(argv[0], argv)
+        return self.run_function(procname, spwan_user_command, argv)
+
+    def terminate(self, pid):
+        return self.processes[pid].terminate()
+
+    def remove(self, pid):
+        del self.processes[pid]

nilmrun/server.py

@@ -9,6 +9,7 @@ import decorator
 import psutil
 import traceback
 import argparse
+import time
 
 import nilmdb
 from nilmdb.utils.printf import *
@@ -24,12 +25,13 @@ from nilmdb.server.serverutil import (
         cherrypy_stop,
         )
 import nilmrun
+import nilmrun.testfilter
 
 # Add CORS_allow tool
 cherrypy.tools.CORS_allow = cherrypy.Tool('on_start_resource', CORS_allow)
 
 # CherryPy apps
-class NilmRunApp(object):
+class App(object):
     """Root application for NILM runner"""
 
     def __init__(self):
@@ -40,7 +42,7 @@ class NilmRunApp(object):
     def index(self):
         cherrypy.response.headers['Content-Type'] = 'text/plain'
         msg = sprintf("This is NilmRun version %s, running on host %s.\n",
-                      nilmdb.__version__, socket.getfqdn())
+                      nilmrun.__version__, socket.getfqdn())
         return msg
 
     # /favicon.ico
@@ -54,6 +56,95 @@ class NilmRunApp(object):
     def version(self):
         return nilmrun.__version__
 
+class AppProcess(object):
+
+    def __init__(self, manager):
+        self.manager = manager
+
+    def process_status(self, pid):
+        return {
+            "pid": pid,
+            "alive": self.manager[pid].alive,
+            "exitcode": self.manager[pid].exitcode,
+            "name": self.manager[pid].name,
+            "start_time": self.manager[pid].start_time,
+            "parameters": self.manager[pid].parameters,
+            "log": self.manager[pid].log,
+            }
+
+    # /process/status
+    @cherrypy.expose
+    @cherrypy.tools.json_out()
+    def status(self, pid, clear = False):
+        """Return status about a process.  If clear = True, also clear
+        the log."""
+        if pid not in self.manager:
+            raise cherrypy.HTTPError("404 Not Found", "No such PID")
+        status = self.process_status(pid)
+        if clear:
+            self.manager[pid].clear_log()
+        return status
+
+    # /process/list
+    @cherrypy.expose
+    @cherrypy.tools.json_out()
+    def list(self):
+        """Return a list of processes in the manager."""
+        return list(self.manager)
+
+    # /process/remove
+    @cherrypy.expose
+    @cherrypy.tools.json_in()
+    @cherrypy.tools.json_out()
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
+    def remove(self, pid):
+        """Remove a process from the manager, killing it if necessary."""
+        if pid not in self.manager:
+            raise cherrypy.HTTPError("404 Not Found", "No such PID")
+        if not self.manager.terminate(pid): # pragma: no cover
+            raise cherrypy.HTTPError("503 Service Unavailable",
+                                     "Failed to stop process")
+        status = self.process_status(pid)
+        self.manager.remove(pid)
+        return status
+
+class AppRun(object):
+    def __init__(self, manager):
+        self.manager = manager
+
+    # /run/command
+    @cherrypy.expose
+    @cherrypy.tools.json_in()
+    @cherrypy.tools.json_out()
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
+    def command(self, argv):
+        """Execute an arbitrary program on the server.  argv is a
+        list of the program and its arguments: 'argv[0]' is the program
+        and 'argv[1:]' are arguments"""
+        return self.manager.run_command("command", argv)
+
+    # /run/code
+    @cherrypy.expose
+    @cherrypy.tools.json_in()
+    @cherrypy.tools.json_out()
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
+    def code(self, code, args):
+        """Execute arbitrary Python code.  'code' is a formatted string.
+        It will be run as if it were written into a Python file and
+        executed, with the arguments in 'args' passed on the command line
+        (i.e., they end up in sys.argv[1:])"""
+        return self.manager.run_code("usercode", code, args)
+
+    # /run/testfilter
+    @cherrypy.expose
+    @cherrypy.tools.json_in()
+    @cherrypy.tools.json_out()
+    @exception_to_httperror(KeyError, ValueError)
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
+    def testfilter(self, data):
+        return self.manager.run_function(
+            "dummy", nilmrun.testfilter.test, data)
+
 class Server(object):
     def __init__(self, host = '127.0.0.1', port = 8080,
                  embedded = True,         # hide diagnostics and output, etc
@@ -104,7 +195,10 @@ class Server(object):
         cherrypy._cperror._ie_friendly_error_sizes = {}
 
         # Build up the application and mount it
-        root = NilmRunApp()
+        manager = nilmrun.processmanager.ProcessManager()
+        root = App()
+        root.process = AppProcess(manager)
+        root.run = AppRun(manager)
         cherrypy.tree.apps = {}
         cherrypy.tree.mount(root, basepath, config = { "/" : app_config })
 

nilmrun/testfilter.py

@@ -0,0 +1,22 @@
+#!/usr/bin/python
+
+from nilmdb.utils.printf import *
+import time
+import signal
+import sys
+
+# This is just for testing the process management.
+def test(n):
+    n = int(n)
+    if n < 0: # raise an exception
+        raise Exception("test exception")
+    if n == 0: # ignore SIGTERM and count to 100
+        n = 100
+        signal.signal(signal.SIGTERM, signal.SIG_IGN)
+    for x in range(n):
+        s = sprintf("dummy %d\n", x)
+        if x & 1:
+            sys.stdout.write(s)
+        else:
+            sys.stderr.write(s)
+        time.sleep(0.1)

scripts/nilmrun_server.py

@@ -20,7 +20,7 @@ def main():
                        help = 'Only listen on the given address',
                        default = '0.0.0.0')
     group.add_argument('-p', '--port', help = 'Listen on the given port',
-                       type = int, default = 12380)
+                       type = int, default = 12381)
     group.add_argument('-q', '--quiet', help = 'Silence output',
                        action = 'store_true')
     group.add_argument('-t', '--traceback',
@@ -41,12 +41,14 @@ def main():
 
     # Print info
     if not args.quiet:
-        print "Version: %s" % nilmrun.__version__
+        print "NilmRun version: %s" % nilmrun.__version__
+        print ("Note: This server does not do any authentication! " +
+               "Anyone who can connect can run arbitrary commands.")
         if args.address == '0.0.0.0' or args.address == '::':
             host = socket.getfqdn()
         else:
             host = args.address
-        print "NilmRun Server URL: http://%s:%d/" % ( host, args.port)
+        print "Server URL: http://%s:%d/" % ( host, args.port)
         print "----"
 
     server.start(blocking = True)

setup.cfg

@@ -0,0 +1,22 @@
+[aliases]
+test = nosetests
+
+[nosetests]
+# Note: values must be set to 1, and have no comments on the same line,
+# for "python setup.py nosetests" to work correctly.
+nocapture=1
+# Comment this out to see CherryPy logs on failure:
+nologcapture=1
+with-coverage=1
+cover-inclusive=1
+cover-package=nilmrun
+cover-erase=1
+# this works, puts html output in cover/ dir:
+# cover-html=1
+# need nose 1.1.3 for this:
+# cover-branches=1
+#debug=nose
+#debug-log=nose.log
+stop=1
+verbosity=2
+tests=tests

setup.py

@@ -30,7 +30,7 @@ except ImportError:
 # Versioneer manages version numbers from git tags.
 # https://github.com/warner/python-versioneer
 import versioneer
-versioneer.versionfile_source = 'src/_version.py'
+versioneer.versionfile_source = 'nilmrun/_version.py'
 versioneer.versionfile_build = 'nilmrun/_version.py'
 versioneer.tag_prefix = 'nilmrun-'
 versioneer.parentdir_prefix = 'nilmrun-'
@@ -69,12 +69,13 @@ setup(name='nilmrun',
       packages = [ 'nilmrun',
                    'nilmrun.scripts',
                    ],
-      package_dir = { 'nilmrun': 'src',
+      package_dir = { 'nilmrun': 'nilmrun',
                       'nilmrun.scripts': 'scripts',
                       },
       entry_points = {
           'console_scripts': [
               'nilmrun-server = nilmrun.scripts.nilmrun_server:main',
+              'nilm-trainola = nilmrun.trainola:main',
               ],
           },
       zip_safe = False,

tests/runtests.py

@@ -0,0 +1,49 @@
+#!/usr/bin/python
+
+import nose
+import os
+import sys
+import glob
+from collections import OrderedDict
+
+# Change into parent dir
+os.chdir(os.path.dirname(os.path.realpath(__file__)) + "/..")
+
+class JimOrderPlugin(nose.plugins.Plugin):
+    """When searching for tests and encountering a directory that
+    contains a 'test.order' file, run tests listed in that file, in the
+    order that they're listed.  Globs are OK in that file and duplicates
+    are removed."""
+    name = 'jimorder'
+    score = 10000
+
+    def prepareTestLoader(self, loader):
+        def wrap(func):
+            def wrapper(name, *args, **kwargs):
+                addr = nose.selector.TestAddress(
+                    name, workingDir=loader.workingDir)
+                try:
+                    order = os.path.join(addr.filename, "test.order")
+                except Exception:
+                    order = None
+                if order and os.path.exists(order):
+                    files = []
+                    for line in open(order):
+                        line = line.split('#')[0].strip()
+                        if not line:
+                            continue
+                        fn = os.path.join(addr.filename, line.strip())
+                        files.extend(sorted(glob.glob(fn)) or [fn])
+                    files = list(OrderedDict.fromkeys(files))
+                    tests = [ wrapper(fn, *args, **kwargs) for fn in files ]
+                    return loader.suiteClass(tests)
+                return func(name, *args, **kwargs)
+            return wrapper
+        loader.loadTestsFromName = wrap(loader.loadTestsFromName)
+        return loader
+
+# Use setup.cfg for most of the test configuration.  Adding
+# --with-jimorder here means that a normal "nosetests" run will
+# still work, it just won't support test.order.
+nose.main(addplugins = [ JimOrderPlugin() ],
+          argv = sys.argv + ["--with-jimorder"])

tests/test.order

@@ -0,0 +1,3 @@
+test_nilmrun.py
+
+test_*.py

tests/test_nilmrun.py

@@ -0,0 +1,325 @@
+# -*- coding: utf-8 -*-
+
+import nilmrun.server
+
+from nilmdb.client.httpclient import HTTPClient, ClientError, ServerError
+
+from nilmdb.utils.printf import *
+
+from nose.plugins.skip import SkipTest
+from nose.tools import *
+from nose.tools import assert_raises
+
+import itertools
+import distutils.version
+import os
+import sys
+import threading
+import cStringIO
+import simplejson as json
+import unittest
+import warnings
+import time
+import re
+import urllib2
+from urllib2 import urlopen, HTTPError
+import requests
+import pprint
+import textwrap
+
+from testutil.helpers import *
+
+testurl = "http://localhost:32181/"
+
+def setup_module():
+    global test_server
+
+    # Start web app on a custom port
+    test_server = nilmrun.server.Server(host = "127.0.0.1",
+                                        port = 32181,
+                                        force_traceback = True)
+    test_server.start(blocking = False)
+
+def teardown_module():
+    global test_server
+    # Close web app
+    test_server.stop()
+
+class TestClient(object):
+
+    def wait_kill(self, client, pid, timeout = 1):
+        time.sleep(timeout)
+        status = client.get("/process/status", { "pid": pid })
+        if not status["alive"]:
+            raise AssertionError("died before we could kill it")
+        status = client.post("/process/remove", { "pid": pid })
+        if status["alive"]:
+            raise AssertionError("didn't get killed")
+        return status
+
+    def wait_end(self, client, pid, timeout = 5, remove = True):
+        start = time.time()
+        status = None
+        while (time.time() - start) < timeout:
+            status = client.get("/process/status", { "pid": pid })
+            if status["alive"] == False:
+                break
+        else:
+            raise AssertionError("process " + str(pid) + " didn't die in " +
+                                 str(timeout) + " seconds: " + repr(status))
+        if remove:
+            status = client.post("/process/remove", { "pid": pid })
+        return status
+
+    def test_client_01_basic(self):
+        client = HTTPClient(baseurl = testurl)
+        version = client.get("/version")
+        eq_(distutils.version.LooseVersion(version),
+            distutils.version.LooseVersion(nilmrun.__version__))
+
+        in_("This is NilmRun", client.get("/"))
+
+        with assert_raises(ClientError):
+            client.get("/favicon.ico")
+
+    def test_client_02_manager(self):
+        client = HTTPClient(baseurl = testurl)
+
+        eq_(client.get("/process/list"), [])
+
+        with assert_raises(ClientError) as e:
+            client.get("/process/status", { "pid": 12345 })
+        in_("No such PID", str(e.exception))
+        with assert_raises(ClientError):
+            client.get("/process/remove", { "pid": 12345 })
+        in_("No such PID", str(e.exception))
+
+    def test_client_03_process_basic(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        # start dummy filter
+        pid = client.post("/run/testfilter", { "data": 30 })
+        eq_(client.get("/process/list"), [pid])
+        time.sleep(1)
+
+        # Verify that status looks OK
+        status = client.get("/process/status", { "pid": pid, "clear": True })
+        for x in [ "pid", "alive", "exitcode", "name",
+                   "start_time", "parameters", "log" ]:
+            in_(x, status)
+        in_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
+        eq_(status["alive"], True)
+        eq_(status["exitcode"], None)
+
+        # Check that the log got cleared
+        status = client.get("/process/status", { "pid": pid })
+        nin_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
+
+        # See that it ended properly
+        status = self.wait_end(client, pid, remove = False)
+        in_("dummy 27\ndummy 28\ndummy 29\n", status["log"])
+        eq_(status["exitcode"], 0)
+
+        # Remove it
+        killstatus = client.post("/process/remove", { "pid": pid })
+        eq_(status, killstatus)
+        eq_(client.get("/process/list"), [])
+        with assert_raises(ClientError) as e:
+            client.post("/process/remove", { "pid": pid })
+        in_("No such PID", str(e.exception))
+
+    def test_client_04_process_terminate(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        # Trigger exception in filter
+        pid = client.post("/run/testfilter", { "data": -1 })
+        time.sleep(0.5)
+        status = client.get("/process/status", { "pid": pid })
+        eq_(status["alive"], False)
+        eq_(status["exitcode"], 1)
+        in_("Exception: test exception", status["log"])
+        client.post("/process/remove", { "pid": pid })
+
+        # Kill a running filter by removing it early
+        newpid = client.post("/run/testfilter", { "data": 50 })
+        ne_(newpid, pid)
+        time.sleep(0.5)
+        start = time.time()
+        status = client.post("/process/remove", { "pid": newpid })
+        elapsed = time.time() - start
+        # Should have died in slightly over 1 second
+        assert(0.5 < elapsed < 2)
+        eq_(status["alive"], False)
+        ne_(status["exitcode"], 0)
+
+        # No more
+        eq_(client.get("/process/list"), [])
+
+        # Try to remove a running filter that ignored SIGTERM
+        pid = client.post("/run/testfilter", { "data": 0 })
+        start = time.time()
+        status = client.post("/process/remove", { "pid": pid })
+        elapsed = time.time() - start
+        # Should have died in slightly over 2 seconds
+        assert(1.5 < elapsed < 3)
+        eq_(status["alive"], False)
+        ne_(status["exitcode"], 0)
+
+    @unittest.skip("trainola moving to nilmtools")
+    def test_client_05_trainola_simple(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        pid = client.post("/run/trainola", { "data": {} })
+        status = self.wait_end(client, pid, remove = False)
+        ne_(status["exitcode"], 0)
+        status = client.post("/process/remove", { "pid": pid })
+
+    @unittest.skip("needs a running nilmdb; trainola moved to nilmtools")
+    def test_client_06_trainola(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        data = { "url": "http://bucket.mit.edu/nilmdb",
+                 "stream": "/sharon/prep-a",
+                 "start": 1366111383280463,
+                 "end": 1366126163457797,
+                 "columns": [ { "name": "P1", "index": 0 },
+                              { "name": "Q1", "index": 1 },
+                              { "name": "P3", "index": 2 } ],
+                 "exemplars": [
+                     { "name": "Boiler Pump ON",
+                       "url": "http://bucket.mit.edu/nilmdb",
+                       "stream": "/sharon/prep-a",
+                       "start": 1366260494269078,
+                       "end": 1366260608185031,
+                       "columns": [ { "name": "P1", "index": 0 },
+                                    { "name": "Q1", "index": 1 }
+                                    ]
+                       },
+                     { "name": "Boiler Pump OFF",
+                       "url": "http://bucket.mit.edu/nilmdb",
+                       "stream": "/sharon/prep-a",
+                       "start": 1366260864215764,
+                       "end": 1366260870882998,
+                       "columns": [ { "name": "P1", "index": 0 },
+                                    { "name": "Q1", "index": 1 }
+                                    ]
+                       }
+                     ]
+                 }
+
+        # start trainola
+        pid = client.post("/run/trainola", { "data": data })
+
+        # wait for it to finish
+        for i in range(60):
+            time.sleep(1)
+            if i == 2:
+                status = client.get("/process/status", { "pid": pid,
+                                                         "clear": True })
+                in_("Loading stream data", status['log'])
+            elif i == 3:
+                status = client.get("/process/status", { "pid": pid })
+                nin_("Loading stream data", status['log'])
+            else:
+                status = client.get("/process/status", { "pid": pid })
+            if status["alive"] == False:
+                break
+        else:
+            client.post("/process/remove", {"pid": pid })
+            raise AssertionError("took too long")
+        if i < 3:
+            raise AssertionError("too fast?")
+
+    def test_client_07_run_command(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        eq_(client.get("/process/list"), [])
+
+        def do(argv, kill):
+            pid = client.post("/run/command", { "argv": argv } )
+            eq_(client.get("/process/list"), [pid])
+            if kill:
+                return self.wait_kill(client, pid)
+            return self.wait_end(client, pid)
+
+        # Simple command
+        status = do(["pwd"], False)
+        eq_(status["exitcode"], 0)
+        eq_("/tmp\n", status["log"])
+
+        # Command with args
+        status = do(["expr", "1", "+", "2"], False)
+        eq_(status["exitcode"], 0)
+        eq_("3\n", status["log"])
+
+        # Missing command
+        status = do(["/no-such-command-blah-blah"], False)
+        ne_(status["exitcode"], 0)
+
+        # Kill a slow command
+        status = do(["sleep", "60"], True)
+        ne_(status["exitcode"], 0)
+
+    def test_client_08_run_code(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        eq_(client.get("/process/list"), [])
+
+        def do(code, args, kill):
+            pid = client.post("/run/code", { "code": code, "args": args } )
+            eq_(client.get("/process/list"), [pid])
+            if kill:
+                return self.wait_kill(client, pid)
+            return self.wait_end(client, pid)
+
+        # basic code snippet
+        code=textwrap.dedent("""
+        print 'hello'
+        def foo(arg):
+            print 'world'
+        """)
+        status = do(code, [], False)
+        eq_("hello\n", status["log"])
+        eq_(status["exitcode"], 0)
+
+        # compile error
+        code=textwrap.dedent("""
+        def foo(arg:
+            print 'hello'
+        """)
+        status = do(code, [], False)
+        in_("SyntaxError", status["log"])
+        eq_(status["exitcode"], 1)
+
+        # traceback in user code should be formatted nicely
+        code=textwrap.dedent("""
+        def foo(arg):
+            raise Exception(arg)
+        foo(123)
+        """)
+        status = do(code, [], False)
+        eq_('Traceback (most recent call last):\n' +
+        '  File "<user-code>", line 4, in <module>\n' +
+        '    foo(123)\n' +
+        '  File "<user-code>", line 3, in foo\n' +
+        '    raise Exception(arg)\n' +
+        'Exception: 123\n', status["log"])
+        eq_(status["exitcode"], 1)
+
+        # argument handling (strings come in as unicode)
+        code=textwrap.dedent("""
+        import sys
+        print sys.argv[1].encode('ascii'), sys.argv[2]
+        sys.exit(0)  # also test raising SystemExit
+        """)
+        status = do(code, ["hello", 123], False)
+        eq_(status["log"], "hello 123\n")
+        eq_(status["exitcode"], 0)
+
+        # try killing a long-running process
+        code=textwrap.dedent("""
+        import time
+        print 'hello'
+        time.sleep(60)
+        print 'world'
+        """)
+        status = do(code, [], True)
+        eq_(status["log"], "hello\n")
+        ne_(status["exitcode"], 0)

tests/testutil/__init__.py

@@ -0,0 +1 @@
+# empty

tests/testutil/helpers.py

@@ -0,0 +1,41 @@
+# Just some helpers for test functions
+
+def myrepr(x):
+    if isinstance(x, basestring):
+        return '"' + x + '"'
+    else:
+        return repr(x)
+
+def eq_(a, b):
+    if not a == b:
+        raise AssertionError("%s != %s" % (myrepr(a), myrepr(b)))
+
+def lt_(a, b):
+    if not a < b:
+        raise AssertionError("%s is not less than %s" % (myrepr(a), myrepr(b)))
+
+def in_(a, b):
+    if a not in b:
+        raise AssertionError("%s not in %s" % (myrepr(a), myrepr(b)))
+
+def nin_(a, b):
+    if a in b:
+        raise AssertionError("unexpected %s in %s" % (myrepr(a), myrepr(b)))
+
+def in2_(a1, a2, b):
+    if a1 not in b and a2 not in b:
+        raise AssertionError("(%s or %s) not in %s" % (myrepr(a1), myrepr(a2),
+                                                       myrepr(b)))
+
+def ne_(a, b):
+    if not a != b:
+        raise AssertionError("unexpected %s == %s" % (myrepr(a), myrepr(b)))
+
+def lines_(a, n):
+    l = a.count('\n')
+    if not l == n:
+        if len(a) > 5000:
+            a = a[0:5000] + " ... truncated"
+        raise AssertionError("wanted %d lines, got %d in output: '%s'"
+                             % (n, l, a))
+