test_couch.py 文件源码

def test_cancel_replication():
    server = Server("http://test.test:5984")
    httpretty.register_uri(
        httpretty.HEAD, "http://test.test:5984/_replicator",
        status=200
    )
    global doc_exists
    doc_exists = True
    def head_test_src(request, uri, headers):
        if doc_exists:
            return 200, headers, ""
        else:
            return 404, headers, ""
    httpretty.register_uri(
        httpretty.HEAD, "http://test.test:5984/_replicator/test_src",
        etag="a", body=head_test_src
    )
    def delete_test_src(request, uri, headers):
        global doc_exists
        doc_exists = False
        return 200, headers, ""
    httpretty.register_uri(
        httpretty.DELETE, "http://test.test:5984/_replicator/test_src",
        etag="a", body=delete_test_src
    )
    assert "test_src" in server["_replicator"]
    server.cancel_replication("test_src")
    assert "test_src" not in server["_replicator"]