Force /stream/interval and /stream/extract responses to be text/plain

nilmdb/server.py

@@ -37,6 +37,14 @@ def chunked_response(func):
     func._cp_config = { 'response.stream': True }
     return func
 
+def response_type(content_type):
+    """Return a decorator-generating function that sets the
+    response type to the specified string."""
+    def wrapper(func, *args, **kwargs):
+        cherrypy.response.headers['Content-Type'] = content_type
+        return func(*args, **kwargs)
+    return decorator.decorator(wrapper)
+
 @decorator.decorator
 def workaround_cp_bug_1200(func, *args, **kwargs): # pragma: no cover
     """Decorator to work around CherryPy bug #1200 in a response
@@ -272,6 +280,7 @@ class Stream(NilmApp):
     # /stream/intervals?path=/newton/prep&start=1234567890.0&end=1234567899.0
     @cherrypy.expose
     @chunked_response
+    @response_type("text/plain")
     def intervals(self, path, start = None, end = None):
         """
         Get intervals from backend database.  Streams the resulting
@@ -308,6 +317,7 @@ class Stream(NilmApp):
     # /stream/extract?path=/newton/prep&start=1234567890.0&end=1234567899.0
     @cherrypy.expose
     @chunked_response
+    @response_type("text/plain")
     def extract(self, path, start = None, end = None, count = False):
         """
         Extract data from backend database.  Streams the resulting

tests/test_client.py

@@ -227,7 +227,7 @@ class TestClient(object):
         client = nilmdb.Client(url = "http://localhost:12380/")
 
         for x in client.stream_extract("/newton/prep", 123, 123):
-            raise Exception("shouldn't be any data for this request")
+            raise AssertionError("shouldn't be any data for this request")
 
         with assert_raises(ClientError) as e:
             client.stream_remove("/newton/prep", 123, 120)
@@ -281,28 +281,38 @@ class TestClient(object):
             in_("404 Not Found", str(e.exception))
             in_("No such stream", str(e.exception))
 
-    def test_client_7_chunked(self):
+    def test_client_7_headers(self):
         # Make sure that /stream/intervals and /stream/extract
-        # properly return streaming, chunked response.  Pokes around
-        # in client.http internals a bit to look at the response
-        # headers.
+        # properly return streaming, chunked, text/plain response.
+        # Pokes around in client.http internals a bit to look at the
+        # response headers.
 
         client = nilmdb.Client(url = "http://localhost:12380/")
+        http = client.http
 
         # Use a warning rather than returning a test failure, so that we can
         # still disable chunked responses for debugging.
-        x = client.http.get("stream/intervals", { "path": "/newton/prep" },
+
+        # Intervals
+        x = http.get("stream/intervals", { "path": "/newton/prep" },
                             retjson=False)
         lines_(x, 1)
-        if "transfer-encoding: chunked" not in client.http._headers.lower():
+        if "Transfer-Encoding: chunked" not in http._headers:
             warnings.warn("Non-chunked HTTP response for /stream/intervals")
+        if "Content-Type: text/plain;charset=utf-8" not in http._headers:
+            raise AssertionError("/stream/intervals is not text/plain:\n" +
+                                 http._headers)
 
-        x = client.http.get("stream/extract",
+        # Extract
+        x = http.get("stream/extract",
                             { "path": "/newton/prep",
                               "start": "123",
                               "end": "123" }, retjson=False)
-        if "transfer-encoding: chunked" not in client.http._headers.lower():
+        if "Transfer-Encoding: chunked" not in http._headers:
             warnings.warn("Non-chunked HTTP response for /stream/extract")
+        if "Content-Type: text/plain;charset=utf-8" not in http._headers:
+            raise AssertionError("/stream/extract is not text/plain:\n" +
+                                 http._headers)
 
     def test_client_8_unicode(self):
         # Basic Unicode tests