Merge lp:~pedronis/u1db/http-app into lp:u1db

[Samuele Pedroni]

Start of a WSGI application to expose access to a collection of u1dbs, sync and variants of put docs are implemented.

To centralize handling of deserialisation and matching argument introduce and use a http_method decorator. The goal is that http method implementation shouldn't need to deal with finding arguments or deserializing json on their own, in order to ensure proper BadRequest to be generated as needed in a uniform way.

Still to do on future branches:
- pick a proper approach to lookup *Resource classes
- more operations
- (more) error handling

Integration testing and a start of a HTTP client side are started in http-app-sync-testing.

[John A Meinel (jameinel)]

Does it matter that we don't check the number of bytes in 'self.kept' during read_chunk?

http_method seems a bit too magical, and I'm having some trouble understanding what it is trying to do just from its docstring.

I think you're trying to convert a call that passes args and content dict into actually calling arguments of a function.

I wonder if it would be less magical if you didn't have it also have options itself. like 'content_as_args'. Would it be better to just have "http_method" and "http_content_method" or something like that.

I like that the decorator makes writing the other functions cleaner, it is just hard to understand the decorator itself.

Maybe a separate type conversion decorator? So you would have

@arg_conversion(b=int)
@http_method()
def f(self, a, b):
 ...

It may not be worthwhile, I'm just brainstorming since I still find http_method a bit hard to follow. It does seem to have good test coverage, though.

test_args_no_query seems like the exception could be raised because f() doesn't take 'a' as a parameter, rather than because we explicitly requested no_query=True. Maybe use a default argument a=N, and assert that we still pass when we think we should, and fail appropriately?

content_unserialized_as_args I would probably rename to just content_as_args. At least, I found the 'unserialized' bit more confusing than helpful. I'm certainly open to discussion on it, though.

If content-length isn't sent, it seems to set the total length to 0 for FencedReader, which looks like the read() will get called and always return the empty string. Is that intentional? Would it be cleaner to just fail as soon as we notice it is not set?

I wonder what we'll want the routing logic to look like. Since I think we'll want ~user, and maybe even nesting in the api. (path/to/database, vs only supporting top-level databases). Maybe not, though. Short term this certainly works fine.

class TestFencedReader(testtools.TestCase):

I usually use

from u1db import tests

class TestFencedReader(tests.TestCase):

that way we can change our base TestCase class, without updating all the tests themselves. (I've already used that once to switch from unittest.TestCase to testtools.TestCase.)

We want to be using "simplejson" and not just "json". At least from my benchmarks:

$ TIMEIT "[json.dumps(o) for o in objs]"
10 loops, best of 3: 5.33 sec per loop

vs

$ TIMEIT -s "import json" "[simplejson.dumps(o) for o in objs]"
10 loops, best of 3: 1.15 sec per loop

It appears that the stdlib's "json" module is pure-python, and about 5x slower than simplejson's C version.

If we want to make it a soft dependency we can do something like:

try:
  import simplejson as json
except ImportError:
  import json

But the performance impact is pretty significant.

review: Needs Fixing

[lp:~pedronis/u1db/http-app]

116. By Samuele Pedroni on 2011-11-16

switch to simplejson which is faster

117. By Samuele Pedroni on 2011-11-16

use our own TestCase

118. By Samuele Pedroni on 2011-11-16

content_unserialized_as_args -> content_as_args, fix doc

119. By Samuele Pedroni on 2011-11-16

an empty body is not valid json

120. By Samuele Pedroni on 2011-11-16

clarification, make _FencedReader and ._kept protected

121. By Samuele Pedroni on 2011-11-16

clarify test

[Samuele Pedroni (pedronis)]

> Does it matter that we don't check the number of bytes in 'self.kept' during
> read_chunk?

it's going and is meant to be a subchunk from a previous call result, so ignoring atmost is ok,
and it's not worth trying to read a bit more at that point.

I renamed things though, because they are really internal helpers, and added a comment about this.

>
> http_method seems a bit too magical, and I'm having some trouble understanding
> what it is trying to do just from its docstring.
>
> I think you're trying to convert a call that passes args and content dict into
> actually calling arguments of a function.
>
> I wonder if it would be less magical if you didn't have it also have options
> itself. like 'content_as_args'. Would it be better to just have "http_method"
> and "http_content_method" or something like that.
>
> I like that the decorator makes writing the other functions cleaner, it is
> just hard to understand the decorator itself.
>
> Maybe a separate type conversion decorator? So you would have
>
> @arg_conversion(b=int)
> @http_method()
> def f(self, a, b):
> ...

that's hard to do without either having order matter, or adding further complexity

>
> It may not be worthwhile, I'm just brainstorming since I still find
> http_method a bit hard to follow. It does seem to have good test coverage,
> though.
>

I understand the concern, I want to use it a bit more before refactoring. I think the idea of having different decorators for the special parameters would be nice, not sure how to do that without duplication or more complexity though.

>
>
> test_args_no_query seems like the exception could be raised because f()
> doesn't take 'a' as a parameter, rather than because we explicitly requested
> no_query=True. Maybe use a default argument a=N, and assert that we still pass
> when we think we should, and fail appropriately?

changed the test along those lines

>
>
> content_unserialized_as_args I would probably rename to just content_as_args.
> At least, I found the 'unserialized' bit more confusing than helpful. I'm
> certainly open to discussion on it, though.

renamed

>
>
> If content-length isn't sent, it seems to set the total length to 0 for
> FencedReader, which looks like the read() will get called and always return
> the empty string. Is that intentional? Would it be cleaner to just fail as
> soon as we notice it is not set?
>

as we discussed an empty body is never valid for the content types we expect so changed the code to raise a BadRequest directly.

>
> I wonder what we'll want the routing logic to look like. Since I think we'll
> want ~user, and maybe even nesting in the api. (path/to/database, vs only
> supporting top-level databases). Maybe not, though. Short term this certainly
> works fine.

we don't want to support ~user I think for the ref implementation. supporting path/to/database with this kind of interface involves some level of DWIM or we need to go the couch %2F route but it adds a lot of complexity and it's error prone, especially when you throw oauth in the mix. I think I would like to not do path/to/database unless if have a strong use case for that.

>
>
> class TestFencedReader(testtoo...

[John A Meinel (jameinel)]

Looks giod to me.

review: Approve

=== added file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	1970-01-01 00:00:00 +0000
+++ u1db/remote/http_app.py	2011-11-16 15:30:08 +0000
@@ -0,0 +1,319 @@
+# Copyright 2011 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 3, as published
+# by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""HTTP Application exposing U1DB."""
+
+import functools
+import httplib
+import inspect
+import simplejson
+import sys
+import urlparse
+
+
+class _FencedReader(object):
+    """Read and get lines from a file but not past a given length."""
+
+    MAXCHUNK = 8192
+
+    # xxx do checks for maximum admissible total size
+    # and maximum admissible line size
+
+    def __init__(self, rfile, total):
+        self.rfile = rfile
+        self.remaining = total
+        self._kept = None
+
+    def read_chunk(self, atmost):
+        if self._kept is not None:
+            # ignore atmost, kept data should be a subchunk anyway
+            return self._kept
+        if self.remaining == 0:
+            return ''
+        data = self.rfile.read(min(self.remaining, atmost))
+        self.remaining -= len(data)
+        return data
+
+    def getline(self):
+        line_parts = []
+        while True:
+            chunk = self.read_chunk(self.MAXCHUNK)
+            if chunk == '':
+                break
+            nl = chunk.find("\n")
+            if nl != -1:
+                line_parts.append(chunk[:nl+1])
+                rest = chunk[nl+1:]
+                self._kept = rest or None
+                break
+            else:
+                line_parts.append(chunk)
+        return ''.join(line_parts)
+
+
+class BadRequest(Exception):
+    """Bad request."""
+
+
+def http_method(**control):
+    """Decoration for handling of query arguments and content for a HTTP method.
+
+       args and content here are the query arguments and body of the incoming
+       HTTP requests.
+
+       Match query arguments to python method arguments:
+           w = http_method()(f)
+           w(self, args, content) => args["content"]=content;
+                                     f(self, **args)
+
+       JSON deserialize content to arguments:
+           w = http_method(content_as_args=True,...)(f)
+           w(self, args, content) => args.update(simplejson.loads(content));
+                                     f(self, **args)
+
+       Support conversions (e.g int):
+           w = http_method(Arg=Conv,...)(f)
+           w(self, args, content) => args["Arg"]=Conv(args["Arg"]);
+                                     f(self, **args)
+
+       Enforce no use of query arguments:
+           w = http_method(no_query=True,...)(f)
+           w(self, args, content) raises BadRequest if args is not empty
+
+       Argument mismatches, deserialisation failures produce BadRequest.
+    """
+    content_as_args = control.pop('content_as_args', False)
+    no_query = control.pop('no_query', False)
+    conversions = control.items()
+    def wrap(f):
+        argspec = inspect.getargspec(f)
+        assert argspec.args[0] == "self"
+        nargs = len(argspec.args)
+        ndefaults = len(argspec.defaults or ())
+        required_args = set(argspec.args[1:nargs-ndefaults])
+        all_args = set(argspec.args)
+        @functools.wraps(f)
+        def wrapper(self, args, content):
+            if no_query and args:
+                raise BadRequest()
+            if content is not None:
+                if content_as_args:
+                    try:
+                        args.update(simplejson.loads(content))
+                    except ValueError:
+                        raise BadRequest()
+                else:
+                    args["content"] = content
+            if not (required_args <= set(args) <= all_args):
+                raise BadRequest()
+            for name, conv in conversions:
+                try:
+                    args[name] = conv(args[name])
+                except ValueError:
+                    raise BadRequest()
+            return f(self, **args)
+        return wrapper
+    return wrap
+
+
+class DocResource(object):
+    """Document resource."""
+
+    def __init__(self, dbname, id, state, responder):
+        self.id = id
+        self.responder = responder
+        self.db = state.open_database(dbname)
+
+    @http_method()
+    def put(self, content, old_rev=None):
+        doc_rev = self.db.put_doc(self.id, old_rev, content)
+        if old_rev is None:
+            status = 201 # created
+        else:
+            status = 200
+        self.responder.send_response(status, rev=doc_rev)
+
+
+class SyncResource(object):
+    """Sync endpoint resource."""
+
+    def __init__(self, dbname, from_replica_uid, state, responder):
+        self.from_replica_uid = from_replica_uid
+        self.responder = responder
+        self.target = state.open_database(dbname).get_sync_target()
+
+    @http_method()
+    def get(self):
+        result = self.target.get_sync_info(self.from_replica_uid)
+        self.responder.send_response(this_replica_uid=result[0],
+                                     this_replica_generation=result[1],
+                                     other_replica_uid=self.from_replica_uid,
+                                     other_replica_generation=result[2])
+
+    @http_method(generation=int,
+                 content_as_args=True, no_query=True)
+    def put(self, generation):
+        self.target.record_sync_info(self.from_replica_uid, generation)
+        self.responder.send_response(ok=True)
+
+    # Implements the same logic as LocalSyncTarget.sync_exchange
+
+    @http_method(from_replica_generation=int, last_known_generation=int,
+                 content_as_args=True)
+    def post_args(self, last_known_generation, from_replica_generation):
+        self.from_replica_generation = from_replica_generation
+        self.last_known_generation = last_known_generation
+        self.sync_exch = self.target.get_sync_exchange()
+
+    @http_method(content_as_args=True)
+    def post_stream_entry(self, id, rev, doc):
+        self.sync_exch.insert_doc_from_source(id, rev, doc)
+
+    def post_end(self):
+        def send_doc(doc_id, doc_rev, doc):
+            entry = dict(id=doc_id, rev=doc_rev, doc=doc)
+            self.responder.stream_entry(entry)
+        new_gen = self.sync_exch.find_docs_to_return(self.last_known_generation)
+        self.responder.content_type = 'application/x-u1db-multi-json'
+        self.responder.start_response(new_generation=new_gen)
+        new_gen = self.sync_exch.return_docs_and_record_sync(
+                                                  self.from_replica_uid,
+                                                  self.from_replica_generation,
+                                                  send_doc)
+        self.responder.finish_response()
+
+
+OK = 200
+
+class HTTPResponder(object):
+    """Encode responses from the server back to the client."""
+
+    # a multi document response will put args and documents
+    # each on one line of the response body
+
+    def __init__(self, start_response):
+        self._started = False
+        self.sent_response = False
+        self._start_response = start_response
+        self._write = None
+        self.content_type = 'application/json'
+
+    def start_response(self, status=OK, **kwargs):
+        """start sending response: header and args."""
+        if self._started:
+            return
+        self._started = True
+        status_text = httplib.responses[status]
+        self._write = self._start_response('%d %s' % (status, status_text),
+                                         [('content-type', self.content_type),
+                                          ('cache-control', 'no-cache')])
+        # xxx version in headers
+        if kwargs:
+            self._write(simplejson.dumps(kwargs)+"\r\n")
+
+    def finish_response(self):
+        """finish sending response."""
+        self.sent_response = True
+
+    def send_response(self, status=OK, **kwargs):
+        """send and finish response in one go."""
+        self.start_response(status, **kwargs)
+        self.finish_response()
+
+    def stream_entry(self, entry):
+        "send stream entry as part of the response."
+        assert self._started
+        self._write(simplejson.dumps(entry)+"\r\n")
+
+
+class HTTPInvocationByMethodWithBody(object):
+    """Invoke methods on a resource."""
+
+    def __init__(self, resource, environ):
+        self.resource = resource
+        self.environ = environ
+
+    def _lookup(self, method):
+        try:
+            return getattr(self.resource, method)
+        except AttributeError:
+            raise BadRequest()
+
+    def __call__(self):
+        args = urlparse.parse_qsl(self.environ['QUERY_STRING'],
+                                  strict_parsing=False)
+        try:
+            args = dict((k.decode('utf-8'), v.decode('utf-8')) for k,v in args)
+        except ValueError:
+            raise BadRequest()
+        method = self.environ['REQUEST_METHOD'].lower()
+        if method in ('get', 'delete'):
+            meth = self._lookup(method)
+            return meth(args, None)
+        else:
+            # we expect content-length > 0, reconsider if we move
+            # to support chunked enconding
+            try:
+                content_length = int(self.environ['CONTENT_LENGTH'])
+            except (ValueError, KeyError), e:
+                raise BadRequest
+            if content_length <= 0:
+                raise BadRequest
+            reader = _FencedReader(self.environ['wsgi.input'], content_length)
+            content_type = self.environ.get('CONTENT_TYPE')
+            if content_type == 'application/json':
+                meth = self._lookup(method)
+                body = reader.read_chunk(sys.maxint)
+                return meth(args, body)
+            elif content_type == 'application/x-u1db-multi-json':
+                meth_args = self._lookup('%s_args' % method)
+                meth_entry = self._lookup('%s_stream_entry' % method)
+                meth_end = self._lookup('%s_end' % method)
+                body_getline = reader.getline
+                meth_args(args, body_getline())
+                while True:
+                    line = body_getline()
+                    if not line:
+                        break
+                    entry = line.strip()
+                    meth_entry({}, entry)
+                return meth_end()
+            else:
+                raise BadRequest()
+
+
+class HTTPApp(object):
+
+    def __init__(self, state):
+        self.state = state
+
+    def _lookup_resource(self, environ, responder):
+        # xxx proper dispatch logic
+        parts = environ['PATH_INFO'].split('/')
+        if len(parts) == 4 and parts[2] == 'doc':
+            resource = DocResource(parts[1], parts[3], self.state, responder)
+        elif len(parts) == 4 and parts[2] == 'sync-from':
+            resource = SyncResource(parts[1], parts[3], self.state, responder)
+        else:
+            raise BadRequest()
+        return resource
+
+    def __call__(self, environ, start_response):
+        responder = HTTPResponder(start_response)
+        try:
+            resource = self._lookup_resource(environ, responder)
+            HTTPInvocationByMethodWithBody(resource, environ)()
+        except BadRequest:
+            responder.send_response(400)
+        return []

=== added file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	1970-01-01 00:00:00 +0000
+++ u1db/tests/test_http_app.py	2011-11-16 15:30:08 +0000
@@ -0,0 +1,466 @@
+# Copyright 2011 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 3, as published
+# by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the WSGI app."""
+
+import paste.fixture
+import simplejson
+import StringIO
+
+from u1db import (
+    tests,
+    )
+
+from u1db.remote import (
+    http_app,
+    )
+
+
+class TestFencedReader(tests.TestCase):
+
+    def test_init(self):
+        reader = http_app._FencedReader(StringIO.StringIO(""), 25)
+        self.assertEqual(25, reader.remaining)
+
+    def test_read_chunk(self):
+        inp = StringIO.StringIO("abcdef")
+        reader = http_app._FencedReader(inp, 5)
+        data = reader.read_chunk(2)
+        self.assertEqual("ab", data)
+        self.assertEqual(2, inp.tell())
+        self.assertEqual(3, reader.remaining)
+
+    def test_read_chunk_remaining(self):
+        inp = StringIO.StringIO("abcdef")
+        reader = http_app._FencedReader(inp, 4)
+        data = reader.read_chunk(9999)
+        self.assertEqual("abcd", data)
+        self.assertEqual(4, inp.tell())
+        self.assertEqual(0, reader.remaining)
+
+    def test_read_chunk_nothing_left(self):
+        inp = StringIO.StringIO("abc")
+        reader = http_app._FencedReader(inp, 2)
+        reader.read_chunk(2)
+        self.assertEqual(2, inp.tell())
+        self.assertEqual(0, reader.remaining)
+        data = reader.read_chunk(2)
+        self.assertEqual("", data)
+        self.assertEqual(2, inp.tell())
+        self.assertEqual(0, reader.remaining)
+
+    def test_read_chunk_kept(self):
+        inp = StringIO.StringIO("abcde")
+        reader = http_app._FencedReader(inp, 4)
+        reader._kept = "xyz"
+        data = reader.read_chunk(2) # atmost ignored
+        self.assertEqual("xyz", data)
+        self.assertEqual(0, inp.tell())
+        self.assertEqual(4, reader.remaining)
+
+    def test_getline(self):
+        inp = StringIO.StringIO("abc\r\nde")
+        reader = http_app._FencedReader(inp, 6)
+        reader.MAXCHUNK = 6
+        line = reader.getline()
+        self.assertEqual("abc\r\n", line)
+        self.assertEqual("d", reader._kept)
+
+    def test_getline_exact(self):
+        inp = StringIO.StringIO("abcd\r\nef")
+        reader = http_app._FencedReader(inp, 6)
+        reader.MAXCHUNK = 6
+        line = reader.getline()
+        self.assertEqual("abcd\r\n", line)
+        self.assertIs(None, reader._kept)
+
+    def test_getline_no_newline(self):
+        inp = StringIO.StringIO("abcd")
+        reader = http_app._FencedReader(inp, 4)
+        reader.MAXCHUNK = 6
+        line = reader.getline()
+        self.assertEqual("abcd", line)
+
+    def test_getline_many_chunks(self):
+        inp = StringIO.StringIO("abcde\r\nf")
+        reader = http_app._FencedReader(inp, 8)
+        reader.MAXCHUNK = 4
+        line = reader.getline()
+        self.assertEqual("abcde\r\n", line)
+        self.assertEqual("f", reader._kept)
+
+    def test_getline_empty(self):
+        inp = StringIO.StringIO("")
+        reader = http_app._FencedReader(inp, 0)
+        reader.MAXCHUNK = 4
+        line = reader.getline()
+        self.assertEqual("", line)
+        line = reader.getline()
+        self.assertEqual("", line)
+
+    def test_getline_just_newline(self):
+        inp = StringIO.StringIO("\r\n")
+        reader = http_app._FencedReader(inp, 2)
+        reader.MAXCHUNK = 4
+        line = reader.getline()
+        self.assertEqual("\r\n", line)
+        line = reader.getline()
+        self.assertEqual("", line)
+
+
+class TestHTTPMethodDecorator(tests.TestCase):
+
+    def test_args(self):
+        @http_app.http_method()
+        def f(self, a, b):
+            return self, a, b
+        res = f("self", {"a": "x", "b": "y"}, None)
+        self.assertEqual(("self", "x", "y"), res)
+
+    def test_args_missing(self):
+        @http_app.http_method()
+        def f(self, a, b):
+            return a, b
+        self.assertRaises(http_app.BadRequest, f, "self", {"a": "x"}, None)
+
+    def test_args_unexpected(self):
+        @http_app.http_method()
+        def f(self, a):
+            return a
+        self.assertRaises(http_app.BadRequest, f, "self",
+                                                  {"a": "x", "c": "z"}, None)
+
+    def test_args_default(self):
+        @http_app.http_method()
+        def f(self, a, b="z"):
+            return a, b
+        res = f("self", {"a": "x"}, None)
+        self.assertEqual(("x", "z"), res)
+
+    def test_args_conversion(self):
+        @http_app.http_method(b=int)
+        def f(self, a, b):
+            return self, a, b
+        res = f("self", {"a": "x", "b": "2"}, None)
+        self.assertEqual(("self", "x", 2), res)
+
+        self.assertRaises(http_app.BadRequest, f, "self",
+                                                  {"a": "x", "b": "foo"}, None)
+
+    def test_args_content(self):
+        @http_app.http_method()
+        def f(self, a, content):
+            return a, content
+        res = f(self, {"a": "x"}, "CONTENT")
+        self.assertEqual(("x", "CONTENT"), res)
+
+    def test_args_content_as_args(self):
+        @http_app.http_method(b=int, content_as_args=True)
+        def f(self, a, b):
+            return self, a, b
+        res = f("self", {"a": "x"}, '{"b": "2"}')
+        self.assertEqual(("self", "x", 2), res)
+
+        self.assertRaises(http_app.BadRequest, f, "self", {}, 'not-json')
+
+    def test_args_content_no_query(self):
+        @http_app.http_method(no_query=True,
+                              content_as_args=True)
+        def f(self, a='a', b='b'):
+            return a, b
+        res = f("self", {}, '{"b": "y"}')
+        self.assertEqual(('a', 'y'), res)
+
+        self.assertRaises(http_app.BadRequest, f, "self", {'a': 'x'},
+                          '{"b": "y"}')
+
+class TestResource(object):
+
+    @http_app.http_method()
+    def get(self, a, b):
+        self.args = dict(a=a, b=b)
+        return 'Get'
+
+    @http_app.http_method()
+    def put(self, a, content):
+        self.args = dict(a=a)
+        self.content = content
+        return 'Put'
+
+    @http_app.http_method(content_as_args=True)
+    def put_args(self, a, b):
+        self.args = dict(a=a, b=b)
+        self.order = ['a']
+        self.entries = []
+
+    @http_app.http_method()
+    def put_stream_entry(self, content):
+        self.entries.append(content)
+        self.order.append('s')
+
+    def put_end(self):
+        self.order.append('e')
+        return "Put/end"
+
+class TestHTTPInvocationByMethodWithBody(tests.TestCase):
+
+    def test_get(self):
+        resource = TestResource()
+        environ = {'QUERY_STRING': 'a=1&b=2', 'REQUEST_METHOD': 'GET'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        res = invoke()
+        self.assertEqual('Get', res)
+        self.assertEqual({'a': '1', 'b': '2'}, resource.args)
+
+    def test_put_json(self):
+        resource = TestResource()
+        body = '{"body": true}'
+        environ = {'QUERY_STRING': 'a=1', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO(body),
+                   'CONTENT_LENGTH': str(len(body)),
+                   'CONTENT_TYPE': 'application/json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        res = invoke()
+        self.assertEqual('Put', res)
+        self.assertEqual({'a': '1'}, resource.args)
+        self.assertEqual('{"body": true}', resource.content)
+
+    def test_put_multi_json(self):
+        resource = TestResource()
+        body = (
+            '{"b": 2}\r\n'       # args
+            '{"entry": "x"}\r\n' # stream entry
+            '{"entry": "y"}\r\n' # stream entry
+            )
+        environ = {'QUERY_STRING': 'a=1', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO(body),
+                   'CONTENT_LENGTH': str(len(body)),
+                   'CONTENT_TYPE': 'application/x-u1db-multi-json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        res = invoke()
+        self.assertEqual('Put/end', res)
+        self.assertEqual({'a': '1', 'b': 2}, resource.args)
+        self.assertEqual(['{"entry": "x"}', '{"entry": "y"}'], resource.entries)
+        self.assertEqual(['a', 's', 's', 'e'], resource.order)
+
+    def test_bad_request_decode_failure(self):
+        resource = TestResource()
+        environ = {'QUERY_STRING': 'a=\xff', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO('{}'),
+                   'CONTENT_LENGTH': '2',
+                   'CONTENT_TYPE': 'application/json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_unsupported_content_type(self):
+        resource = TestResource()
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO('{}'),
+                   'CONTENT_LENGTH': '2',
+                   'CONTENT_TYPE': 'text/plain'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_no_content_length(self):
+        resource = TestResource()
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO('a'),
+                   'CONTENT_TYPE': 'application/json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_invalid_content_length(self):
+        resource = TestResource()
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO('abc'),
+                   'CONTENT_LENGTH': '1unk',
+                   'CONTENT_TYPE': 'application/json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_empty_body(self):
+        resource = TestResource()
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO(''),
+                   'CONTENT_LENGTH': '0',
+                   'CONTENT_TYPE': 'application/json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(resource, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_unsupported_method_get_like(self):
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'DELETE'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(None, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_unsupported_method_put_like(self):
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'PUT',
+                   'wsgi.input': StringIO.StringIO('{}'),
+                   'CONTENT_LENGTH': '2',
+                   'CONTENT_TYPE': 'application/json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(None, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+    def test_bad_request_unsupported_method_put_like_multi_json(self):
+        body = '{}\r\n{}\r\n'
+        environ = {'QUERY_STRING': '', 'REQUEST_METHOD': 'POST',
+                   'wsgi.input': StringIO.StringIO(body),
+                   'CONTENT_LENGTH': str(len(body)),
+                   'CONTENT_TYPE': 'application/x-u1db-multi-json'}
+        invoke = http_app.HTTPInvocationByMethodWithBody(None, environ)
+        self.assertRaises(http_app.BadRequest, invoke)
+
+
+class TestHTTPResponder(tests.TestCase):
+
+    def start_response(self, status, headers):
+        self.status = status
+        self.headers = dict(headers)
+        self.response_body = []
+        def write(data):
+            self.response_body.append(data)
+        return write
+
+    def test_send_response(self):
+        responder = http_app.HTTPResponder(self.start_response)
+        responder.send_response(value='success')
+        self.assertEqual('200 OK', self.status)
+        self.assertEqual({'content-type': 'application/json',
+                          'cache-control': 'no-cache'}, self.headers)
+        self.assertEqual(['{"value": "success"}\r\n'], self.response_body)
+
+    def test_send_response_status_fail(self):
+        responder = http_app.HTTPResponder(self.start_response)
+        responder.send_response(400)
+        self.assertEqual('400 Bad Request', self.status)
+        self.assertEqual({'content-type': 'application/json',
+                          'cache-control': 'no-cache'}, self.headers)
+        self.assertEqual([], self.response_body)
+
+    def test_start_finish_response_status_fail(self):
+        responder = http_app.HTTPResponder(self.start_response)
+        responder.start_response(404, error='not found')
+        responder.finish_response()
+        self.assertEqual('404 Not Found', self.status)
+        self.assertEqual({'content-type': 'application/json',
+                          'cache-control': 'no-cache'}, self.headers)
+        self.assertEqual(['{"error": "not found"}\r\n'], self.response_body)
+
+    def test_send_stream_entry(self):
+        responder = http_app.HTTPResponder(self.start_response)
+        responder.content_type = "application/x-u1db-multi-json"
+        responder.start_response(one=1)
+        responder.stream_entry({'entry': True})
+        responder.finish_response()
+        self.assertEqual('200 OK', self.status)
+        self.assertEqual({'content-type': 'application/x-u1db-multi-json',
+                          'cache-control': 'no-cache'}, self.headers)
+        self.assertEqual(['{"one": 1}\r\n',
+                          '{"entry": true}\r\n'], self.response_body)
+
+
+class TestHTTPApp(tests.TestCase):
+
+    def setUp(self):
+        super(TestHTTPApp, self).setUp()
+        self.state = tests.ServerStateForTests()
+        application = http_app.HTTPApp(self.state)
+        self.app = paste.fixture.TestApp(application)
+        self.db0 = self.state._create_database('db0')
+
+    def test_bad_request_broken(self):
+        resp = self.app.put('/db0/doc/doc1', params='{"x": 1}',
+                            headers={'content-type': 'application/foo'},
+                            expect_errors=True)
+        self.assertEqual(400, resp.status)
+
+    def test_bad_request_dispatch(self):
+        resp = self.app.put('/db0/foo/doc1', params='{"x": 1}',
+                            headers={'content-type': 'application/json'},
+                            expect_errors=True)
+        self.assertEqual(400, resp.status)
+
+    def test_put_doc_create(self):
+        resp = self.app.put('/db0/doc/doc1', params='{"x": 1}',
+                            headers={'content-type': 'application/json'})
+        doc_rev, doc, _ = self.db0.get_doc('doc1')
+        self.assertEqual(201, resp.status) # created
+        self.assertEqual('{"x": 1}', doc)
+        self.assertEqual('application/json', resp.header('content-type'))
+        self.assertEqual({'rev': doc_rev}, simplejson.loads(resp.body))
+
+    def test_put_doc(self):
+        doc_id, orig_rev = self.db0.create_doc('doc1', '{"x": 1}')
+        resp = self.app.put('/db0/doc/doc1?old_rev=%s' % orig_rev,
+                            params='{"x": 2}',
+                            headers={'content-type': 'application/json'})
+        doc_rev, doc, _ = self.db0.get_doc('doc1')
+        self.assertEqual(200, resp.status)
+        self.assertEqual('{"x": 2}', doc)
+        self.assertEqual('application/json', resp.header('content-type'))
+        self.assertEqual({'rev': doc_rev}, simplejson.loads(resp.body))
+
+    def test_get_sync_info(self):
+        self.db0.set_sync_generation('other-id', 1)
+        resp = self.app.get('/db0/sync-from/other-id')
+        self.assertEqual(200, resp.status)
+        self.assertEqual('application/json', resp.header('content-type'))
+        self.assertEqual(dict(this_replica_uid='db0',
+                              this_replica_generation=0,
+                              other_replica_uid='other-id',
+                              other_replica_generation=1),
+                              simplejson.loads(resp.body))
+
+    def test_record_sync_info(self):
+        resp = self.app.put('/db0/sync-from/other-id',
+                            params='{"generation": 2}',
+                            headers={'content-type': 'application/json'})
+        self.assertEqual(200, resp.status)
+        self.assertEqual('application/json', resp.header('content-type'))
+        self.assertEqual({'ok': True}, simplejson.loads(resp.body))
+        self.assertEqual(self.db0.get_sync_generation('other-id'), 2)
+
+    def test_sync_exchange_send(self):
+        entry = {'id': 'doc-here', 'rev': 'replica:1', 'doc':
+                 '{"value": "here"}'}
+        args = dict(from_replica_generation=10, last_known_generation=0)
+        body = ("%s\r\n" % simplejson.dumps(args) +
+                "%s\r\n" % simplejson.dumps(entry))
+        resp = self.app.post('/db0/sync-from/replica',
+                            params=body,
+                            headers={'content-type':
+                                     'application/x-u1db-multi-json'})
+        self.assertEqual(200, resp.status)
+        self.assertEqual('application/x-u1db-multi-json',
+                         resp.header('content-type'))
+        self.assertEqual({'new_generation': 1}, simplejson.loads(resp.body))
+        self.assertEqual(('replica:1', '{"value": "here"}', False),
+                         self.db0.get_doc('doc-here'))
+
+    def test_sync_exchange_receive(self):
+        doc_id, doc_rev = self.db0.create_doc('{"value": "there"}')
+        args = dict(from_replica_generation=10, last_known_generation=0)
+        body = "%s\r\n" % simplejson.dumps(args)
+        resp = self.app.post('/db0/sync-from/replica',
+                            params=body,
+                            headers={'content-type':
+                                     'application/x-u1db-multi-json'})
+        self.assertEqual(200, resp.status)
+        self.assertEqual('application/x-u1db-multi-json',
+                         resp.header('content-type'))
+        parts = resp.body.splitlines()
+        self.assertEqual(2, len(parts))
+        self.assertEqual({'new_generation': 1}, simplejson.loads(parts[0]))
+        self.assertEqual({'doc': '{"value": "there"}',
+                          'rev': doc_rev, 'id': doc_id},
+                         simplejson.loads(parts[1]))