Allow start==end in stream_insert_context, if no data was provided.

We were still calculating the maximum number of rows correctly,
so the extra data was really extra and would get re-written to the
beginning of the subsequent file.

The only case in which this would lead to database issues is if the
very last file was lengthened incorrectly, and the "nrows" calculation
would therefore be wrong when the database was reopened.  Still, even
in that case, it should just leave a small gap in the data, not cause
any errors.

docs/design.md

@@ -421,3 +421,20 @@ and has all of the same functions.  It adds three new functions:
 It is significantly faster!  It is about 20 times faster to decimate a
 stream with `nilm-decimate` when the filter code is using the new
 binary/numpy interface.
+
+
+WSGI interface & chunked requests
+---------------------------------
+
+mod_wsgi requires "WSGIChunkedRequest On" to handle
+"Transfer-encoding: Chunked" requests.  However, `/stream/insert`
+doesn't handle this correctly right now, because:
+
+- The `cherrpy.request.body.read()` call needs to be fixed for chunked requests
+
+- We don't want to just buffer endlessly in the server, and it will
+  require some thought on how to handle data in chunks (what to do about
+  interval endpoints).
+
+It is probably better to just keep the endpoint management on the client
+side, so leave "WSGIChunkedRequest off" for now.

extras/fix-oversize-files.py

@@ -0,0 +1,50 @@
+#!/usr/bin/python
+
+import os
+import sys
+import cPickle as pickle
+import argparse
+import fcntl
+import re
+from nilmdb.client.numpyclient import layout_to_dtype
+
+parser = argparse.ArgumentParser(
+    description = """
+Fix database corruption where binary writes caused too much data to be
+written to the file.  Truncates files to the correct length.  This was
+fixed by b98ff1331a515ad47fd3203615e835b529b039f9.
+""")
+parser.add_argument("path", action="store", help='Database root path')
+parser.add_argument("-y", "--yes", action="store_true", help='Fix them')
+args = parser.parse_args()
+
+lock = os.path.join(args.path, "data.lock")
+with open(lock, "w") as f:
+    fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+    fix = {}
+
+    for (path, dirs, files) in os.walk(args.path):
+        if "_format" in files:
+            with open(os.path.join(path, "_format")) as format:
+                fmt = pickle.load(format)
+                rowsize = layout_to_dtype(fmt["layout"]).itemsize
+                maxsize = rowsize * fmt["rows_per_file"]
+                fix[path] = maxsize
+                if maxsize < 128000000: # sanity check
+                    raise Exception("bad maxsize " + str(maxsize))
+
+    for fixpath in fix:
+        for (path, dirs, files) in os.walk(fixpath):
+            for fn in files:
+                if not re.match("^[0-9a-f]{4,}$", fn):
+                    continue
+                fn = os.path.join(path, fn)
+                size = os.path.getsize(fn)
+                maxsize = fix[fixpath]
+                if size > maxsize:
+                    diff = size - maxsize
+                    print diff, "too big:", fn
+                    if args.yes:
+                        with open(fn, "a+") as dbfile:
+                            dbfile.truncate(maxsize)

nilmdb/client/client.py

@@ -6,7 +6,6 @@ import nilmdb.utils
 import nilmdb.client.httpclient
 from nilmdb.client.errors import ClientError
 
-import re
 import time
 import simplejson as json
 import contextlib
@@ -66,12 +65,8 @@ class Client(object):
             params["layout"] = layout
         if extended:
             params["extended"] = 1
-        def sort_streams_nicely(x):
-            """Human-friendly sort (/stream/2 before /stream/10)"""
-            num = lambda t: int(t) if t.isdigit() else t
-            key = lambda k: [ num(c) for c in re.split('([0-9]+)', k[0]) ]
-            return sorted(x, key = key)
-        return sort_streams_nicely(self.http.get("stream/list", params))
+        streams = self.http.get("stream/list", params)
+        return nilmdb.utils.sort.sort_human(streams, key = lambda s: s[0])
 
     def stream_get_metadata(self, path, keys = None):
         params = { "path": path }
@@ -122,7 +117,10 @@ class Client(object):
             params["start"] = timestamp_to_string(start)
         if end is not None:
             params["end"] = timestamp_to_string(end)
-        return self.http.post("stream/remove", params)
+        total = 0
+        for count in self.http.post_gen("stream/remove", params):
+            total += int(count)
+        return total
 
     @contextlib.contextmanager
     def stream_insert_context(self, path, start = None, end = None):
@@ -146,6 +144,7 @@ class Client(object):
         ctx = StreamInserter(self, path, start, end)
         yield ctx
         ctx.finalize()
+        ctx.destroy()
 
     def stream_insert(self, path, data, start = None, end = None):
         """Insert rows of data into a stream.  data should be a string
@@ -295,6 +294,15 @@ class StreamInserter(object):
         self._block_data = []
         self._block_len = 0
 
+        self.destroyed = False
+
+    def destroy(self):
+        """Ensure this object can't be used again without raising
+        an error"""
+        def error(*args, **kwargs):
+            raise Exception("don't reuse this context object")
+        self._send_block = self.insert = self.finalize = self.send = error
+
     def insert(self, data):
         """Insert a chunk of ASCII formatted data in string form.  The
         overall data must consist of lines terminated by '\\n'."""
@@ -441,7 +449,7 @@ class StreamInserter(object):
             self._interval_start = end_ts
 
         # Double check endpoints
-        if start_ts is None or end_ts is None:
+        if (start_ts is None or end_ts is None) or (start_ts == end_ts):
             # If the block has no non-comment lines, it's OK
             try:
                 self._get_first_noncomment(block)

nilmdb/client/httpclient.py

@@ -137,5 +137,14 @@ class HTTPClient(object):
         """Simple GET (parameters in URL) returning a generator"""
         return self._req_gen("GET", url, params, binary = binary)
 
+    def post_gen(self, url, params = None):
+        """Simple POST (parameters in body) returning a generator"""
+        if self.post_json:
+            return self._req_gen("POST", url, None,
+                                 json.dumps(params),
+                                 { 'Content-type': 'application/json' })
+        else:
+            return self._req_gen("POST", url, None, params)
+
     # Not much use for a POST or PUT generator, since they don't
     # return much data.

nilmdb/client/numpyclient.py

@@ -98,6 +98,7 @@ class NumpyClient(nilmdb.client.client.Client):
         ctx = StreamInserterNumpy(self, path, start, end, dtype)
         yield ctx
         ctx.finalize()
+        ctx.destroy()
 
     def stream_insert_numpy(self, path, data, start = None, end = None,
                             layout = None):
@@ -133,16 +134,8 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
         contiguous interval and may be None.  'dtype' is the Numpy
         dtype for this stream.
         """
-        self.last_response = None
-
+        super(StreamInserterNumpy, self).__init__(client, path, start, end)
         self._dtype = dtype
-        self._client = client
-        self._path = path
-
-        # Start and end for the overall contiguous interval we're
-        # filling
-        self._interval_start = start
-        self._interval_end = end
 
         # Max rows to send at once
         self._max_rows = self._max_data // self._dtype.itemsize
@@ -162,9 +155,12 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
         elif array.ndim == 2:
             # Convert to structured array
             sarray = numpy.zeros(array.shape[0], dtype=self._dtype)
-            sarray['timestamp'] = array[:,0]
-            # Need the squeeze in case sarray['data'] is 1 dimensional
-            sarray['data'] = numpy.squeeze(array[:,1:])
+            try:
+                sarray['timestamp'] = array[:,0]
+                # Need the squeeze in case sarray['data'] is 1 dimensional
+                sarray['data'] = numpy.squeeze(array[:,1:])
+            except (IndexError, ValueError):
+                raise ValueError("wrong number of fields for this data type")
             array = sarray
         else:
             raise ValueError("wrong number of dimensions in array")
@@ -247,9 +243,12 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
             # Next block continues where this one ended
             self._interval_start = end_ts
 
-        # If we have no endpoints, it's because we had no data to send.
-        if start_ts is None or end_ts is None:
-            return
+        # If we have no endpoints, or equal endpoints, it's OK as long
+        # as there's no data to send
+        if (start_ts is None or end_ts is None) or (start_ts == end_ts):
+            if len(array) == 0:
+                return
+            raise ClientError("have data to send, but invalid start/end times")
 
         # Send it
         data = array.tostring()

nilmdb/cmdline/cmdline.py

@@ -10,6 +10,7 @@ import sys
 import os
 import argparse
 from argparse import ArgumentDefaultsHelpFormatter as def_form
+import signal
 
 try: # pragma: no cover
     import argcomplete
@@ -80,6 +81,12 @@ class Cmdline(object):
 
     def __init__(self, argv = None):
         self.argv = argv or sys.argv[1:]
+        try:
+            # Assume command line arguments are encoded with stdin's encoding,
+            # and reverse it.  Won't be needed in Python 3, but for now..
+            self.argv = [ x.decode(sys.stdin.encoding) for x in self.argv ]
+        except Exception: # pragma: no cover
+            pass
         self.client = None
         self.def_url = os.environ.get("NILMDB_URL", "http://localhost/nilmdb/")
         self.subcmd = {}
@@ -126,6 +133,13 @@ class Cmdline(object):
         sys.exit(-1)
 
     def run(self):
+        # Set SIGPIPE to its default handler -- we don't need Python
+        # to catch it for us.
+        try:
+            signal.signal(signal.SIGPIPE, signal.SIG_DFL)
+        except ValueError: # pragma: no cover
+            pass
+
         # Clear cached timezone, so that we can pick up timezone changes
         # while running this from the test suite.
         datetime_tz._localtz = None

nilmdb/cmdline/destroy.py

@@ -1,5 +1,6 @@
 from nilmdb.utils.printf import *
 import nilmdb.client
+import fnmatch
 
 from argparse import ArgumentDefaultsHelpFormatter as def_form
 
@@ -10,25 +11,39 @@ def setup(self, sub):
                          Destroy the stream at the specified path.
                          The stream must be empty.  All metadata
                          related to the stream is permanently deleted.
+
+                         Wildcards and multiple paths are supported.
                          """)
     cmd.set_defaults(handler = cmd_destroy)
     group = cmd.add_argument_group("Options")
     group.add_argument("-R", "--remove", action="store_true",
                        help="Remove all data before destroying stream")
+    group.add_argument("-q", "--quiet", action="store_true",
+                       help="Don't display names when destroying "
+                       "multiple paths")
     group = cmd.add_argument_group("Required arguments")
-    group.add_argument("path",
-                       help="Path of the stream to delete, e.g. /foo/bar",
+    group.add_argument("path", nargs='+',
+                       help="Path of the stream to delete, e.g. /foo/bar/*",
                        ).completer = self.complete.path
     return cmd
 
 def cmd_destroy(self):
     """Destroy stream"""
-    if self.args.remove:
+    streams = [ s[0] for s in self.client.stream_list() ]
+    paths = []
+    for path in self.args.path:
+        new = fnmatch.filter(streams, path)
+        if not new:
+            self.die("error: no stream matched path: %s", path)
+        paths.extend(new)
+
+    for path in paths:
+        if not self.args.quiet and len(paths) > 1:
+            printf("Destroying %s\n", path)
+
         try:
-            count = self.client.stream_remove(self.args.path)
+            if self.args.remove:
+                count = self.client.stream_remove(path)
+            self.client.stream_destroy(path)
         except nilmdb.client.ClientError as e:
-            self.die("error removing data: %s", str(e))
-    try:
-        self.client.stream_destroy(self.args.path)
-    except nilmdb.client.ClientError as e:
-        self.die("error destroying stream: %s", str(e))
+            self.die("error destroying stream: %s", str(e))

nilmdb/cmdline/extract.py

@@ -1,6 +1,7 @@
 from __future__ import print_function
 from nilmdb.utils.printf import *
 import nilmdb.client
+import sys
 
 def setup(self, sub):
     cmd = sub.add_parser("extract", help="Extract data",
@@ -24,6 +25,8 @@ def setup(self, sub):
                        ).completer = self.complete.time
 
     group = cmd.add_argument_group("Output format")
+    group.add_argument("-B", "--binary", action="store_true",
+                       help="Raw binary output")
     group.add_argument("-b", "--bare", action="store_true",
                        help="Exclude timestamps from output lines")
     group.add_argument("-a", "--annotate", action="store_true",
@@ -42,6 +45,11 @@ def cmd_extract_verify(self):
         if self.args.start > self.args.end:
             self.parser.error("start is after end")
 
+    if self.args.binary:
+        if (self.args.bare or self.args.annotate or self.args.markup or
+            self.args.timestamp_raw or self.args.count):
+            self.parser.error("--binary cannot be combined with other options")
+
 def cmd_extract(self):
     streams = self.client.stream_list(self.args.path)
     if len(streams) != 1:
@@ -60,16 +68,23 @@ def cmd_extract(self):
         printf("# end: %s\n", time_string(self.args.end))
 
     printed = False
+    if self.args.binary:
+        printer = sys.stdout.write
+    else:
+        printer = print
+    bare = self.args.bare
+    count = self.args.count
     for dataline in self.client.stream_extract(self.args.path,
                                                self.args.start,
                                                self.args.end,
                                                self.args.count,
-                                               self.args.markup):
-        if self.args.bare and not self.args.count:
+                                               self.args.markup,
+                                               self.args.binary):
+        if bare and not count:
             # Strip timestamp (first element).  Doesn't make sense
             # if we are only returning a count.
             dataline = ' '.join(dataline.split(' ')[1:])
-        print(dataline)
+        printer(dataline)
         printed = True
     if not printed:
         if self.args.annotate:

nilmdb/cmdline/list.py

@@ -10,22 +10,16 @@ def setup(self, sub):
                          formatter_class = def_form,
                          description="""
                          List streams available in the database,
-                         optionally filtering by layout or path.  Wildcards
-                         are accepted.
+                         optionally filtering by path.  Wildcards
+                         are accepted; non-matching paths or wildcards
+                         are ignored.
                          """)
     cmd.set_defaults(verify = cmd_list_verify,
                      handler = cmd_list)
 
     group = cmd.add_argument_group("Stream filtering")
-    group.add_argument("-p", "--path", metavar="PATH", default="*",
-                       help="Match only this path (-p can be omitted)",
+    group.add_argument("path", metavar="PATH", default=["*"], nargs='*',
                        ).completer = self.complete.path
-    group.add_argument("path_positional", default="*",
-                       nargs="?", help=argparse.SUPPRESS,
-                       ).completer = self.complete.path
-    group.add_argument("-l", "--layout", default="*",
-                       help="Match only this stream layout",
-                       ).completer = self.complete.layout
 
     group = cmd.add_argument_group("Interval info")
     group.add_argument("-E", "--ext", action="store_true",
@@ -49,20 +43,12 @@ def setup(self, sub):
     group = cmd.add_argument_group("Misc options")
     group.add_argument("-T", "--timestamp-raw", action="store_true",
                        help="Show raw timestamps when printing times")
+    group.add_argument("-l", "--layout", action="store_true",
+                       help="Show layout type next to path name")
 
     return cmd
 
 def cmd_list_verify(self):
-    # A hidden "path_positional" argument lets the user leave off the
-    # "-p" when specifying the path.  Handle it here.
-    got_opt = self.args.path != "*"
-    got_pos = self.args.path_positional != "*"
-    if got_pos:
-        if got_opt:
-            self.parser.error("too many paths specified")
-        else:
-            self.args.path = self.args.path_positional
-
     if self.args.start is not None and self.args.end is not None:
         if self.args.start >= self.args.end:
             self.parser.error("start must precede end")
@@ -80,29 +66,33 @@ def cmd_list(self):
     else:
         time_string = nilmdb.utils.time.timestamp_to_human
 
-    for stream in streams:
-        (path, layout, int_min, int_max, rows, time) = stream[:6]
-        if not (fnmatch.fnmatch(path, self.args.path) and
-                fnmatch.fnmatch(layout, self.args.layout)):
-            continue
+    for argpath in self.args.path:
+        for stream in streams:
+            (path, layout, int_min, int_max, rows, time) = stream[:6]
+            if not fnmatch.fnmatch(path, argpath):
+                continue
 
-        printf("%s %s\n", path, layout)
-
-        if self.args.ext:
-            if int_min is None or int_max is None:
-                printf("  interval extents: (no data)\n")
+            if self.args.layout:
+                printf("%s %s\n", path, layout)
             else:
-                printf("  interval extents: %s -> %s\n",
-                       time_string(int_min), time_string(int_max))
-            printf("        total data: %d rows, %.6f seconds\n",
-                   rows or 0,
-                   nilmdb.utils.time.timestamp_to_seconds(time or 0))
+                printf("%s\n", path)
 
-        if self.args.detail:
-            printed = False
-            for (start, end) in self.client.stream_intervals(
-                path, self.args.start, self.args.end):
-                printf("  [ %s -> %s ]\n", time_string(start), time_string(end))
-                printed = True
-            if not printed:
-                printf("  (no intervals)\n")
+            if self.args.ext:
+                if int_min is None or int_max is None:
+                    printf("  interval extents: (no data)\n")
+                else:
+                    printf("  interval extents: %s -> %s\n",
+                           time_string(int_min), time_string(int_max))
+                printf("        total data: %d rows, %.6f seconds\n",
+                       rows or 0,
+                       nilmdb.utils.time.timestamp_to_seconds(time or 0))
+
+            if self.args.detail:
+                printed = False
+                for (start, end) in self.client.stream_intervals(
+                    path, self.args.start, self.args.end):
+                    printf("  [ %s -> %s ]\n",
+                           time_string(start), time_string(end))
+                    printed = True
+                if not printed:
+                    printf("  (no intervals)\n")

nilmdb/cmdline/remove.py

@@ -1,17 +1,19 @@
 from nilmdb.utils.printf import *
 import nilmdb.client
+import fnmatch
 
 def setup(self, sub):
     cmd = sub.add_parser("remove", help="Remove data",
                          description="""
                          Remove all data from a specified time range within a
-                         stream.
+                         stream.  If multiple streams or wildcards are provided,
+                         the same time range is removed from all streams.
                          """)
     cmd.set_defaults(handler = cmd_remove)
 
     group = cmd.add_argument_group("Data selection")
-    group.add_argument("path",
-                       help="Path of stream, e.g. /foo/bar",
+    group.add_argument("path", nargs='+',
+                       help="Path of stream, e.g. /foo/bar/*",
                        ).completer = self.complete.path
     group.add_argument("-s", "--start", required=True,
                        metavar="TIME", type=self.arg_time,
@@ -23,18 +25,31 @@ def setup(self, sub):
                        ).completer = self.complete.time
 
     group = cmd.add_argument_group("Output format")
+    group.add_argument("-q", "--quiet", action="store_true",
+                       help="Don't display names when removing "
+                       "from multiple paths")
     group.add_argument("-c", "--count", action="store_true",
                        help="Output number of data points removed")
     return cmd
 
 def cmd_remove(self):
+    streams = [ s[0] for s in self.client.stream_list() ]
+    paths = []
+    for path in self.args.path:
+        new = fnmatch.filter(streams, path)
+        if not new:
+            self.die("error: no stream matched path: %s", path)
+        paths.extend(new)
+
     try:
-        count = self.client.stream_remove(self.args.path,
-                                          self.args.start, self.args.end)
+        for path in paths:
+            if not self.args.quiet and len(paths) > 1:
+                printf("Removing from %s\n", path)
+            count = self.client.stream_remove(path,
+                                              self.args.start, self.args.end)
+            if self.args.count:
+                printf("%d\n", count);
     except nilmdb.client.ClientError as e:
         self.die("error removing data: %s", str(e))
 
-    if self.args.count:
-        printf("%d\n", count)
-
     return 0

nilmdb/server/nilmdb.py

@@ -675,6 +675,7 @@ class NilmDB(object):
 
             # Count how many were removed
             removed += row_end - row_start
+            remaining -= row_end - row_start
 
             if restart is not None:
                 break

nilmdb/server/rocket.c

@@ -468,7 +468,7 @@ static PyObject *Rocket_append_binary(Rocket *self, PyObject *args)
         }
 
         /* Write binary data */
-        if (fwrite(data, data_len, 1, self->file) != 1) {
+        if (fwrite(data, self->binary_size, rows, self->file) != rows) {
                 PyErr_SetFromErrno(PyExc_OSError);
                 return NULL;
         }

nilmdb/server/server.py

@@ -347,24 +347,34 @@ class Stream(NilmApp):
     # /stream/remove?path=/newton/prep&start=1234567890.0&end=1234567899.0
     @cherrypy.expose
     @cherrypy.tools.json_in()
-    @cherrypy.tools.json_out()
-    @exception_to_httperror(NilmDBError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
+    @chunked_response
+    @response_type("application/x-json-stream")
     def remove(self, path, start = None, end = None):
         """
         Remove data from the backend database.  Removes all data in
-        the interval [start, end).  Returns the number of data points
-        removed.
+        the interval [start, end).
+
+        Returns the number of data points removed.  Since this is a potentially
+        long-running operation, multiple numbers may be returned as the
+        data gets removed from the backend database.  The total number of
+        points removed is the sum of all of these numbers.
         """
         (start, end) = self._get_times(start, end)
-        total_removed = 0
-        while True:
-            (removed, restart) = self.db.stream_remove(path, start, end)
-            total_removed += removed
-            if restart is None:
-                break
-            start = restart
-        return total_removed
+
+        if len(self.db.stream_list(path = path)) != 1:
+            raise cherrypy.HTTPError("404", "No such stream: " + path)
+
+        @workaround_cp_bug_1200
+        def content(start, end):
+            # Note: disable chunked responses to see tracebacks from here.
+            while True:
+                (removed, restart) = self.db.stream_remove(path, start, end)
+                yield json.dumps(removed) + "\r\n"
+                if restart is None:
+                    break
+                start = restart
+        return content(start, end)
 
     # /stream/intervals?path=/newton/prep
     # /stream/intervals?path=/newton/prep&start=1234567890.0&end=1234567899.0

nilmdb/utils/__init__.py

@@ -13,3 +13,4 @@ import nilmdb.utils.time
 import nilmdb.utils.iterator
 import nilmdb.utils.interval
 import nilmdb.utils.lock
+import nilmdb.utils.sort

nilmdb/utils/diskusage.py

@@ -1,4 +1,5 @@
 import os
+import errno
 from math import log
 
 def human_size(num):
@@ -16,10 +17,17 @@ def human_size(num):
         return '1 byte'
 
 def du(path):
-    """Like du -sb, returns total size of path in bytes."""
-    size = os.path.getsize(path)
-    if os.path.isdir(path):
-        for thisfile in os.listdir(path):
-            filepath = os.path.join(path, thisfile)
-            size += du(filepath)
-    return size
+    """Like du -sb, returns total size of path in bytes.  Ignore
+    errors that might occur if we encounter broken symlinks or
+    files in the process of being removed."""
+    try:
+        size = os.path.getsize(path)
+        if os.path.isdir(path):
+            for thisfile in os.listdir(path):
+                filepath = os.path.join(path, thisfile)
+                size += du(filepath)
+        return size
+    except OSError as e: # pragma: no cover
+        if e.errno != errno.ENOENT:
+            raise
+        return 0

nilmdb/utils/sort.py

@@ -0,0 +1,18 @@
+import re
+
+def sort_human(items, key = None):
+    """Human-friendly sort (/stream/2 before /stream/10)"""
+    def to_num(val):
+        try:
+            return int(val)
+        except Exception:
+            return val
+
+    def human_key(text):
+        if key:
+            text = key(text)
+        # Break into character and numeric chunks.
+        chunks = re.split(r'([0-9]+)', text)
+        return [ to_num(c) for c in chunks ]
+
+    return sorted(items, key = human_key)

nilmdb/utils/time.py

@@ -6,7 +6,7 @@ import time
 
 # Range
 min_timestamp = (-2**63)
-max_timestamp = (2**62 - 1)
+max_timestamp = (2**63 - 1)
 
 # Smallest representable step
 epsilon = 1
@@ -32,6 +32,10 @@ def timestamp_to_human(timestamp):
     """Convert a timestamp (integer microseconds since epoch) to a
     human-readable string, using the local timezone for display
     (e.g. from the TZ env var)."""
+    if timestamp == min_timestamp:
+        return "(minimum)"
+    if timestamp == max_timestamp:
+        return "(maximum)"
     dt = datetime_tz.datetime_tz.fromtimestamp(timestamp_to_unix(timestamp))
     return dt.strftime("%a, %d %b %Y %H:%M:%S.%f %z")
 

tests/test_client.py

@@ -105,16 +105,19 @@ class TestClient(object):
                 client.http.post("/stream/list")
         client = nilmdb.client.Client(url = testurl)
 
-        # Create three streams
+        # Create four streams
         client.stream_create("/newton/prep", "float32_8")
         client.stream_create("/newton/raw", "uint16_6")
-        client.stream_create("/newton/zzz/rawnotch", "uint16_9")
+        client.stream_create("/newton/zzz/rawnotch2", "uint16_9")
+        client.stream_create("/newton/zzz/rawnotch11", "uint16_9")
 
-        # Verify we got 3 streams
+        # Verify we got 4 streams in the right order
         eq_(client.stream_list(), [ ["/newton/prep", "float32_8"],
                                     ["/newton/raw", "uint16_6"],
-                                    ["/newton/zzz/rawnotch", "uint16_9"]
+                                    ["/newton/zzz/rawnotch2", "uint16_9"],
+                                    ["/newton/zzz/rawnotch11", "uint16_9"]
                                     ])
+
         # Match just one type or one path
         eq_(client.stream_list(layout="uint16_6"),
             [ ["/newton/raw", "uint16_6"] ])
@@ -327,6 +330,10 @@ class TestClient(object):
                      2525.169921875, 8350.83984375, 3724.699951171875,
                      1355.3399658203125, 2039.0))
 
+        # Just get some coverage
+        with assert_raises(ClientError) as e:
+            client.http.post("/stream/remove", { "path": "/none" })
+
         client.close()
 
     def test_client_06_generators(self):
@@ -613,8 +620,12 @@ class TestClient(object):
         with client.stream_insert_context("/empty/test", end = 950):
             pass
 
+        # Equal start and end is OK as long as there's no data
+        with client.stream_insert_context("/empty/test", start=9, end=9):
+            pass
+
         # Try various things that might cause problems
-        with client.stream_insert_context("/empty/test", 1000, 1050):
+        with client.stream_insert_context("/empty/test", 1000, 1050) as ctx:
             ctx.finalize() # inserts [1000, 1050]
             ctx.finalize() # nothing
             ctx.finalize() # nothing

tests/test_cmdline.py

@@ -300,38 +300,19 @@ class TestCmdline(object):
 
         # Verify we got those 3 streams and they're returned in
         # alphabetical order.
-        self.ok("list")
+        self.ok("list -l")
         self.match("/newton/prep float32_8\n"
                    "/newton/raw uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
 
         # Match just one type or one path.  Also check
         # that --path is optional
-        self.ok("list --path /newton/raw")
-        self.match("/newton/raw uint16_6\n")
-
-        self.ok("list /newton/raw")
-        self.match("/newton/raw uint16_6\n")
-
-        self.fail("list -p /newton/raw /newton/raw")
-        self.contain("too many paths")
-
-        self.ok("list --layout uint16_6")
+        self.ok("list --layout /newton/raw")
         self.match("/newton/raw uint16_6\n")
 
         # Wildcard matches
-        self.ok("list --layout uint16*")
-        self.match("/newton/raw uint16_6\n"
-                   "/newton/zzz/rawnotch uint16_9\n")
-
-        self.ok("list --path *zzz* --layout uint16*")
-        self.match("/newton/zzz/rawnotch uint16_9\n")
-
-        self.ok("list *zzz* --layout uint16*")
-        self.match("/newton/zzz/rawnotch uint16_9\n")
-
-        self.ok("list --path *zzz* --layout float32*")
-        self.match("")
+        self.ok("list *zzz*")
+        self.match("/newton/zzz/rawnotch\n")
 
         # reversed range
         self.fail("list /newton/prep --start 2020-01-01 --end 2000-01-01")
@@ -497,28 +478,28 @@ class TestCmdline(object):
         self.ok("list --detail")
         lines_(self.captured, 8)
 
-        self.ok("list --detail --path *prep")
+        self.ok("list --detail *prep")
         lines_(self.captured, 4)
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:02'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:02'")
         lines_(self.captured, 3)
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05'")
         lines_(self.captured, 2)
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05:15'")
         lines_(self.captured, 2)
         self.contain("10:05:15.000")
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15.50'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05:15.50'")
         lines_(self.captured, 2)
         self.contain("10:05:15.500")
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 19:05:15.50'")
+        self.ok("list --detail *prep --start='23 Mar 2012 19:05:15.50'")
         lines_(self.captured, 2)
         self.contain("no intervals")
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15.50'"
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05:15.50'"
                 + " --end='23 Mar 2012 10:05:15.51'")
         lines_(self.captured, 2)
         self.contain("10:05:15.500")
@@ -527,15 +508,15 @@ class TestCmdline(object):
         lines_(self.captured, 8)
 
         # Verify the "raw timestamp" output
-        self.ok("list --detail --path *prep --timestamp-raw "
+        self.ok("list --detail *prep --timestamp-raw "
                 "--start='23 Mar 2012 10:05:15.50'")
         lines_(self.captured, 2)
         self.contain("[ 1332497115500000 -> 1332497160000000 ]")
 
         # bad time
-        self.fail("list --detail --path *prep -T --start='9332497115.612'")
+        self.fail("list --detail *prep -T --start='9332497115.612'")
         # good time
-        self.ok("list --detail --path *prep -T --start='1332497115.612'")
+        self.ok("list --detail *prep -T --start='1332497115.612'")
         lines_(self.captured, 2)
         self.contain("[ 1332497115612000 -> 1332497160000000 ]")
 
@@ -615,11 +596,19 @@ class TestCmdline(object):
         test(8, "10:01:59.9", "10:02:00.1", extra="-m")
 
         # all data put in by tests
-        self.ok("extract -a /newton/prep --start 2000-01-01 --end 2020-01-01")
+        self.ok("extract -a /newton/prep --start min --end max")
         lines_(self.captured, 43204)
         self.ok("extract -c /newton/prep --start 2000-01-01 --end 2020-01-01")
         self.match("43200\n")
 
+        # test binary mode
+        self.fail("extract -c -B /newton/prep -s min -e max")
+        self.contain("binary cannot be combined")
+        self.fail("extract -m -B /newton/prep -s min -e max")
+        self.contain("binary cannot be combined")
+        self.ok("extract -B /newton/prep -s min -e max")
+        eq_(len(self.captured), 43200 * (8 + 8*4))
+
         # markup for 3 intervals, plus extra markup lines whenever we had
         # a "restart" from the nilmdb.stream_extract function
         self.ok("extract -m /newton/prep --start 2000-01-01 --end 2020-01-01")
@@ -639,7 +628,7 @@ class TestCmdline(object):
 
         # Try nonexistent stream
         self.fail("remove /no/such/foo --start 2000-01-01 --end 2020-01-01")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         # empty or backward ranges return errors
         self.fail("remove /newton/prep --start 2020-01-01 --end 2000-01-01")
@@ -667,9 +656,14 @@ class TestCmdline(object):
                 "--start '23 Mar 2022 20:00:30' " +
                 "--end '23 Mar 2022 20:00:31'")
         self.match("0\n")
+        self.ok("remove -c /newton/prep /newton/pre* " +
+                "--start '23 Mar 2022 20:00:30' " +
+                "--end '23 Mar 2022 20:00:31'")
+        self.match("Removing from /newton/prep\n0\n" +
+                   "Removing from /newton/prep\n0\n")
 
         # Make sure we have the data we expect
-        self.ok("list --detail /newton/prep")
+        self.ok("list -l --detail /newton/prep")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n"
@@ -704,7 +698,7 @@ class TestCmdline(object):
         self.match("24000\n")
 
         # See the missing chunks in list output
-        self.ok("list --detail /newton/prep")
+        self.ok("list --layout --detail /newton/prep")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n"
@@ -718,7 +712,7 @@ class TestCmdline(object):
         # Remove all data, verify it's missing
         self.ok("remove /newton/prep --start 2000-01-01 --end 2020-01-01")
         self.match("")  # no count requested this time
-        self.ok("list --detail /newton/prep")
+        self.ok("list -l --detail /newton/prep")
         self.match("/newton/prep float32_8\n" +
                    "  (no intervals)\n")
 
@@ -736,16 +730,16 @@ class TestCmdline(object):
         self.contain("too few arguments")
 
         self.fail("destroy /no/such/stream")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         self.fail("destroy -R /no/such/stream")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         self.fail("destroy asdfasdf")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         # From previous tests, we have:
-        self.ok("list")
+        self.ok("list -l")
         self.match("/newton/prep float32_8\n"
                    "/newton/raw uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
@@ -761,13 +755,13 @@ class TestCmdline(object):
         lines_(self.captured, 7)
 
         # Destroy for real
-        self.ok("destroy -R /newton/prep")
-        self.ok("list")
+        self.ok("destroy -R /n*/prep")
+        self.ok("list -l")
         self.match("/newton/raw uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
 
         self.ok("destroy /newton/zzz/rawnotch")
-        self.ok("list")
+        self.ok("list -l")
         self.match("/newton/raw uint16_6\n")
 
         self.ok("destroy /newton/raw")
@@ -786,18 +780,17 @@ class TestCmdline(object):
             self.ok("list")
             self.contain(path)
             # Make sure it was created empty
-            self.ok("list --detail --path " + path)
+            self.ok("list --detail " + path)
             self.contain("(no intervals)")
 
     def test_12_unicode(self):
         # Unicode paths.
         self.ok("destroy /newton/asdf/qwer")
-        self.ok("destroy /newton/prep")
-        self.ok("destroy /newton/raw")
+        self.ok("destroy /newton/prep /newton/raw")
         self.ok("destroy /newton/zzz")
 
         self.ok(u"create /düsseldorf/raw uint16_6")
-        self.ok("list --detail")
+        self.ok("list -l --detail")
         self.contain(u"/düsseldorf/raw uint16_6")
         self.contain("(no intervals)")
 
@@ -883,7 +876,7 @@ class TestCmdline(object):
         du_before = nilmdb.utils.diskusage.du(testdb)
 
         # Make sure we have the data we expect
-        self.ok("list --detail")
+        self.ok("list -l --detail")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n"
@@ -919,7 +912,7 @@ class TestCmdline(object):
         self.match("3600\n")
 
         # See the missing chunks in list output
-        self.ok("list --detail")
+        self.ok("list -l --detail")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n"
@@ -1043,7 +1036,7 @@ class TestCmdline(object):
             else:
                 raise AssertionError("data not found at " + seek)
             # Verify "list" output
-            self.ok("list")
+            self.ok("list -l")
             self.match("/" + "/".join(components) + " float32_8\n")
 
         # Lots of renames

tests/test_numpyclient.py

@@ -28,7 +28,10 @@ def setup_module():
     recursive_unlink(testdb)
 
     # Start web app on a custom port
-    test_db = nilmdb.utils.serializer_proxy(nilmdb.server.NilmDB)(testdb)
+    test_db = nilmdb.utils.serializer_proxy(nilmdb.server.NilmDB)(
+        testdb, bulkdata_args = { "file_size" : 16384,
+                                  "files_per_dir" : 3 } )
+
     test_server = nilmdb.server.Server(test_db, host = "127.0.0.1",
                                        port = 32180, stoppable = False,
                                        fast_shutdown = True,
@@ -130,6 +133,15 @@ class TestNumpyClient(object):
                                                   [4, 5]]]))
         in_("wrong number of dimensions", str(e.exception))
 
+        # Wrong number of fields
+        with assert_raises(ValueError) as e:
+            client.stream_insert_numpy("/test/1",
+                                       np.array([[0, 1, 2],
+                                                 [1, 2, 3],
+                                                 [3, 4, 5],
+                                                 [4, 5, 6]]))
+        in_("wrong number of fields", str(e.exception))
+
         # Unstructured
         client.stream_create("/test/2", "float32_8")
         client.stream_insert_numpy(
@@ -170,6 +182,17 @@ class TestNumpyClient(object):
         assert(np.array_equal(a,b))
         assert(np.array_equal(a,c))
 
+        # Make sure none of the files are greater than 16384 bytes as
+        # we configured with the bulkdata_args above.
+        datapath = os.path.join(testdb, "data")
+        for (dirpath, dirnames, filenames) in os.walk(datapath):
+            for f in filenames:
+                fn = os.path.join(dirpath, f)
+                size = os.path.getsize(fn)
+                if size > 16384:
+                    raise AssertionError(sprintf("%s is too big: %d > %d\n",
+                                                 fn, size, 16384))
+
         nilmdb.client.numpyclient.StreamInserterNumpy._max_data = old_max_data
         client.close()
 
@@ -286,8 +309,25 @@ class TestNumpyClient(object):
         with client.stream_insert_numpy_context("/empty/test", end = 950):
             pass
 
+        # Equal start and end is OK as long as there's no data
+        with assert_raises(ClientError) as e:
+            with client.stream_insert_numpy_context("/empty/test",
+                                                    start=9, end=9) as ctx:
+                ctx.insert([[9, 9]])
+                ctx.finalize()
+        in_("have data to send, but invalid start/end times", str(e.exception))
+
+        with client.stream_insert_numpy_context("/empty/test",
+                                                start=9, end=9) as ctx:
+            pass
+
+        # reusing a context object is bad
+        with assert_raises(Exception) as e:
+            ctx.insert([[9, 9]])
+
         # Try various things that might cause problems
-        with client.stream_insert_numpy_context("/empty/test", 1000, 1050):
+        with client.stream_insert_numpy_context("/empty/test",
+                                                1000, 1050) as ctx:
             ctx.finalize() # inserts [1000, 1050]
             ctx.finalize() # nothing
             ctx.finalize() # nothing