Commits

Cmed Technology committed 34d2878

initial import

Comments (0)

Files changed (4)

+This code is made available under the BSD 2-Clause License
+==========================================================
+
+Copyright (c) 2012, Timothy Corbett-Clark <timothy@corbettclark.com>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+Redistributions of source code must retain the above copyright notice, this list
+of conditions and the following disclaimer.
+
+Redistributions in binary form must reproduce the above copyright notice, this
+list of conditions and the following disclaimer in the documentation and/or
+other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+# Mock HTTP server
+
+Test a web service which makes callbacks or
+[webhooks](https://en.wikipedia.org/wiki/Webhook) by using a mock server which
+logs all incoming HTTP traffic and makes them available for subsequent
+introspection. Responses may be optionally queued-up. Inspired by Python's
+[unittest.mock](http://docs.python.org/py3k/library/unittest.mock.html#module-
+unittest.mock).
+
+# API
+
+One special URL ("_control") is used to control the mock object. It has 3 REST
+methods as follows.
+
+## Resetting the mock server
+
+To reset all state (removing all logs and queued responses):
+
+    DELETE _control
+
+## Introspecting the log of inbound HTTP since last reset
+
+To return list of requests and corresponding responses:
+
+    GET _control
+    {
+        log: [
+            {
+                request: {
+                    url: ...
+                    method: ...
+                    headers: ...
+                    data: ...
+                }
+                response: {
+                    status: ...
+                    headers: ...
+                    data: ...
+                }
+            }
+        ]
+        status: "ok"
+    }
+
+## Queuing a response
+
+To queue a response to be used exactly once on next incoming HTTP:
+
+    POST _control
+    {
+        status: ...
+        headers: ...
+        data: ...
+    }
+
+(The queued responses don't need to include all the attributes, and on creation
+of a response the queued headers are merged with sensible default headers.)
+
+## All other inbound HTTP
+
+All GET/POST/etc to any other URL are logged and responded to with the next
+queued response. If there are no responses then a generic 200 is used.
+
+# Requirements
+
+Use pip to install the requirements from ``requirements.txt``:
+
+    pip install -r requirements.txt
+
+# Example session
+
+Start the server
+
+    $ python mockserver.py
+     * Running on http://127.0.0.1:5000/
+     * Restarting with reloader
+
+Then in another console, tell the mock server to respond to the next incoming HTTP with particular data and status:
+
+    $ http -v POST localhost:5000/_control data="hi tim" status=201
+    POST /_control HTTP/1.1
+    Accept: application/json
+    Accept-Encoding: identity, deflate, compress, gzip
+    Content-Type: application/json; charset=utf-8
+    Host: localhost:5000
+    User-Agent: HTTPie/0.3.0
+
+    {
+        "data": "hi tim",
+        "status": "201"
+    }
+
+    HTTP/1.0 200 OK
+    Content-Length: 20
+    Content-Type: application/json
+    Date: Mon, 08 Oct 2012 13:04:21 GMT
+    Server: Werkzeug/0.8.3 Python/2.7.3
+
+    {
+        "status": "ok"
+    }
+
+Make the subsequent incoming HTTP and observe the specified data and status.
+
+    $ http -v localhost:5000/some/new/url
+    GET /some/new/url HTTP/1.1
+    Accept: */*
+    Accept-Encoding: identity, deflate, compress, gzip
+    Host: localhost:5000
+    User-Agent: HTTPie/0.3.0
+
+
+
+    HTTP/1.0 201 CREATED
+    Content-Length: 6
+    Content-Type: text/html; charset=utf-8
+    Date: Mon, 08 Oct 2012 13:04:26 GMT
+    Server: Werkzeug/0.8.3 Python/2.7.3
+
+    hi tim
+
+Ask for a log of calls so far:
+
+    $ http -v localhost:5000/_control
+    GET /_control HTTP/1.1
+    Accept: */*
+    Accept-Encoding: identity, deflate, compress, gzip
+    Host: localhost:5000
+    User-Agent: HTTPie/0.3.0
+
+
+
+    HTTP/1.0 200 OK
+    Content-Length: 629
+    Content-Type: application/json
+    Date: Mon, 08 Oct 2012 13:06:42 GMT
+    Server: Werkzeug/0.8.3 Python/2.7.3
+
+    {
+        "log": [
+            {
+                "request": {
+                    "data": "",
+                    "headers": {
+                        "Accept": "*/*",
+                        "Accept-Encoding": "identity, deflate, compress, gzip",
+                        "Content-Length": "",
+                        "Content-Type": "",
+                        "Host": "localhost:5000",
+                        "User-Agent": "HTTPie/0.3.0"
+                    },
+                    "method": "GET",
+                    "url": "http://localhost:5000/some/new/url"
+                },
+                "response": {
+                    "data": "hi tim",
+                    "headers": {
+                        "Content-Length": "6",
+                        "Content-Type": "text/html; charset=utf-8"
+                    },
+                    "status": 201
+                }
+            }
+        ],
+        "status": "ok"
+    }
+"""HTTP mock server for testing webhooks (and any other client invoked HTTP).
+
+Mock server to test web clients, particularly servers which use webhooks. Logs
+all incoming HTTP traffic and makes it available for introspection. Responses
+can be queued-up (optional). Inspired by Python's unittest.mock.
+
+One special URL ("_control") is used to control the mock object. It has the
+following REST methods:
+
+DELETE _control # Reset all state (removing all logs and queued responses)
+
+GET _control    # Return list of requests and corresponding responses
+{
+    log: [
+        {
+            request: {
+                url: ...
+                method: ...
+                headers: ...
+                data: ...
+            }
+            response: {
+                status: ...
+                headers: ...
+                data: ...
+            }
+        }
+    ]
+    status: "ok"
+}
+
+POST _control   # Queue a response to be used exactly once on next incoming HTTP
+{
+    status: ...
+    headers: ...
+    data: ...
+}
+
+(The queued responses don't need to include all the attributes, and on creation
+of a response the queued headers are merged with sensible default headers.)
+
+All GET/POST/etc to any other URL are logged and responded to with the next
+queued response. If there are no responses then a generic 200 is used.
+
+For easy consumption, we assume that the HTTP headers are a dictionary even
+though they can validly contain duplicate keys.
+
+"""
+from flask import Flask, request, json, jsonify, make_response
+
+app = Flask(__name__)
+
+# Log of HTTP invocations since last reset.
+log = []
+
+# Queue of responses to be used (FIFO) on next non-control HTTP invocation. The
+# responses are data templates and not real flask responses because they must be
+# made at the time they are needed.
+responses = []
+
+
+@app.route('/_control', methods=['GET', 'POST', 'DELETE'])
+def _control():
+    if request.method == 'DELETE':
+        del log[:]
+        del responses[:]
+        return jsonify(status='ok')
+    elif request.method == 'POST':
+        responses.append(request.json)
+        return jsonify(status='ok')
+    elif request.method == 'GET':
+        return jsonify(status='ok', log=log)
+
+
+@app.errorhandler(404)
+def catch_all(e):
+    try:
+        response_template = responses.pop(0)
+    except IndexError:
+        response_template = dict()
+    data = response_template.get('data', '')
+    if not isinstance(data, basestring):
+        data = json.dumps(data)
+    status = response_template.get('status', 200)
+    if not isinstance(status, int):
+        status = int(status)
+    response = make_response(data, status)
+    for k, v in response_template.get('headers', {}).items():
+        response.headers[k] = v
+    log.append(
+        dict(
+            request=dict(
+                url=request.url,
+                method=request.method,
+                headers=dict(request.headers.items()),
+                data=request.data
+            ),
+            response=dict(
+                status=response.status_code,
+                headers=dict(response.headers.items()),
+                data=response.data
+            )
+        )
+    )
+    return response
+
+
+if __name__ == "__main__":
+    app.run(debug=True)
+argparse==1.2.1
+distribute==0.6.24
+Flask==0.9
+httpie==0.3.0
+Jinja2==2.6
+Pygments==1.5
+requests==0.14.1
+Werkzeug==0.8.3
+wsgiref==0.1.2