Demonstrate py.test integration with python-bottle API

listener.py

#!/usr/bin/env python

# parse API get requests from CFME automate
# requires:
# * python 'bottle'
# * port 8080 open
# example: curl -X PUT http://localhost:8080/events/redhat/vm?event=power_on_vm

import os
import sys
import json
import sqlite3
from socket import gethostname
try:
    from bottle import run, route, request, response, install
except ImportError, e:
    print "Unable to import bottle.  Is python-bottle installed?"
    sys.exit(1)

try:
    from bottle_sqlite import SQLitePlugin
except ImportError, e:
    print "Unable to import bottle_sqlite.  Is python-bottle_sqlite installed?"
    sys.exit(1)

# If required, remove existing db
# FIXME, check perms too
db_file = "/tmp/eventdb.sqlite"
if os.path.isfile(db_file):
    os.unlink(db_file)

# Initialize database
conn = sqlite3.connect(db_file) # or use :memory: to put it in RAM
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE event_log (
    system TEXT,
    msg_type TEXT,
    event TEXT,
    event_time TIMESTAMP DEFAULT (datetime('now','localtime'))
)
""")

# Install sqlite bottle plugin
install(SQLitePlugin(dbfile=db_file))

if __name__ == '__main__':
    @route('/events', method='GET')
    @route('/events/', method='GET')
    @route('/events/<system>', method='GET')
    @route('/events/<system>/<msg_type>', method='GET')
    def events_list(db, system=None, msg_type=None):
        response.content_type = 'application/json'

        # Build SQL
        sql = 'SELECT * FROM event_log'

        # Build WHERE clause(s)
        bindings = ()
        where_clause = list()
        if system is not None:
            where_clause.append('system = ?')
            bindings += (system,)
        if msg_type is not None:
            where_clause.append('msg_type = ?')
            bindings += (msg_type,)
        if request.query.event:
            where_clause.append('event = ?')
            bindings += (request.query.event,)

        if where_clause:
            sql += ' WHERE %s' % " AND ".join(where_clause)

        # execute query
        c = db.execute(sql, bindings)

        rows = c.fetchall()
        return json.dumps([dict(r) for r in rows])

    @route('/events/<system>/<msg_type>', method='PUT')
    def event_put(db, system, msg_type):
        response.content_type = 'application/json'
        event = request.query.event
        c = db.executemany("INSERT INTO event_log VALUES (?, ?, ?, CURRENT_TIMESTAMP)", \
                [(system, msg_type, event)])
        db.commit()
        return dict(result='success')

    @route('/events/<system>/<msg_type>', method='DELETE' )
    def event_delete(db, system, msg_type):
        response.content_type = 'application/json'
        event = request.query.event
        c = db.execute('DELETE FROM event_log WHERE event = ?', (event,))
        db.commit()
        return dict(result='success')

    # If binding to a specific name is need ... use socket.gethostname()
    # run(host=gethostname(), port=8080, debug=True)
    run(host='localhost', port=8080, debug=True)

test_listener.py

#!/usr/bin/env python

import os
import sys
import time
import json
import pytest
import subprocess
import shlex
import requests

# Global variable (yuck!) used to record listener process info
listener = None

# Command(s) used to start the listener
listener_script = '''
source %s/.python/bin/activate
/usr/bin/env python listener.py
''' % (os.environ.get('HOME',''),)

# Listener API URL
url = 'http://localhost:8080'

def setup_module(module):
    global listener
    sys.stdout.write("Starting listener ... ")
    listener = subprocess.Popen(listener_script,
            stderr=subprocess.PIPE,
            shell=True)
    sys.stdout.write("(%s)\n" % listener.pid)
    # Wait for listener to start ...
    time.sleep(1)

def teardown_module(module):
    global listener
    print "\nKilling listener (%s)..." % (listener.pid)
    listener.kill()
    (stdout, stderr) = listener.communicate()
    print "%s\n%s" % (stdout, stderr)

def is_successful(r):
    # reqeusts built-in exception handler. Is None if okay
    r.raise_for_status()
    # additional response validation:
    try:
        assert r.headers['content-type'] == "application/json", \
            "Reponse is not JSON format (%s)" % r.headers.get('content-type','UNKNOWN')
    except AssertionError as e:
        print e
        return False
    else:
        return True

def json_pprint(j):
    print json.dumps(j , sort_keys=True, indent=4, separators=(',', ': '))

def test_request_list_1():
    r = requests.get(url+'/events')
    assert is_successful(r)
    data = r.json()
    assert len(data) == 0

def test_request_put_1():
    r = requests.put(url+'/events/system-1/msgtype-1?event=etype-1')
    assert is_successful(r)
    data = r.json()
    assert data.get('result') == 'success'

def test_request_put_2():
    r = requests.put(url+'/events/system-1/msgtype-1?event=etype-2')
    assert is_successful(r)
    data = r.json()
    assert data.get('result') == 'success'

def test_request_list_2():
    r = requests.get(url+'/events')
    assert is_successful(r)
    data = r.json()
    assert len(data) == 2

def test_request_get():
    # FIXME - pytest parameterize me
    for etype in ['etype-1', 'etype-2']:
        r = requests.get(url+'/events/system-1/msgtype-1?event=%s' % etype)
        assert is_successful(r)
        data = r.json()
        assert len(data) == 1
        assert data[0].get('event') == etype

def test_request_delete_1():
    r = requests.delete(url+'/events/system-1/msgtype-1?event=etype-1')
    assert is_successful(r)
    data = r.json()
    assert data.get('result') == 'success'

def test_request_list_3():
    r = requests.get(url+'/events')
    data = r.json()
    assert len(data) == 1

def test_request_delete_2():
    r = requests.delete(url+'/events/system-1/msgtype-1?event=etype-2')
    assert is_successful(r)
    data = r.json()
    assert data.get('result') == 'success'

def test_request_list_4():
    r = requests.get(url+'/events')
    data = r.json()
    assert len(data) == 0