Browse Source

Work on a client implementation that can survive testing

git-svn-id: https://bucket.mit.edu/svn/nilm/nilmdb@10604 ddd99763-3ecb-0310-9145-efcb8ce7c51f
tags/bxinterval-last
Jim Paris 10 years ago
parent
commit
0099a41fd8
6 changed files with 135 additions and 6 deletions
  1. +5
    -0
      client.py
  2. +0
    -1
      nilmdb/__init__.py
  3. +35
    -0
      nilmdb/client.py
  4. +5
    -2
      nilmdb/nilmdb.py
  5. +7
    -3
      setup.cfg
  6. +83
    -0
      tests/test_client.py

+ 5
- 0
client.py View File

@@ -0,0 +1,5 @@
#!/usr/bin/python

import nilmdb

nilmdb.client.run(args = sys.argv[1:])

+ 0
- 1
nilmdb/__init__.py View File

@@ -1,4 +1,3 @@
# empty
from nilmdb import NilmDB, StreamException
from server import Server
from layout import *


+ 35
- 0
nilmdb/client.py View File

@@ -0,0 +1,35 @@
"""Command line client functionality, broken into a separate file
so it can be more easily tested."""

from __future__ import absolute_import
from nilmdb.printf import *

import time
import sys
import re
import os
import urlparse
from optparse import OptionParser, OptionGroup, OptionValueError

version = "0.1"

def parse_opts(args):
parser = OptionParser(usage="usage: %prog [options]",
version="nilmdb client script %prog " + version)
parser.add_option("-q", "--quiet", action="store_true", default=False,
dest="quiet", help="suppress unnecessary console output")
group = OptionGroup(parser, "Server")
group.add_option("-u", "--url", action="store", dest="url",
default="http://localhost:12380/",
help="NilmDB server URL (default: %default)")
parser.add_option_group(group)

(opt, rest) = parser.parse_args(args)

return (opt, rest)

def run(args):
(opt, rest) = parse_opts(args)

if not opt.quiet:
printf("Server URL: %s\n", opt.url)

+ 5
- 2
nilmdb/nilmdb.py View File

@@ -103,6 +103,7 @@ class NilmDB(object):
def sql_schema_update(self):
cur = self.con.cursor()
version = cur.execute("PRAGMA user_version").fetchone()[0]
oldversion = version

while version in sql_schema_updates:
cur.executescript(sql_schema_updates[version])
@@ -110,8 +111,10 @@ class NilmDB(object):
if self.verbose: # pragma: no cover
printf("Schema updated to %d\n", version)

with self.con:
cur.execute("PRAGMA user_version = {v:d}".format(v=version))
if version != oldversion:
# this takes .1 seconds, so only do it if necessary
with self.con:
cur.execute("PRAGMA user_version = {v:d}".format(v=version))

def stream_list(self, path = None, layout = None):
"""Return list of (path, layout) tuples of all streams


+ 7
- 3
setup.cfg View File

@@ -4,8 +4,12 @@ with-coverage=1
cover-inclusive=1
cover-package=nilmdb
cover-erase=1
# cover-html=1 # this works, puts html output in cover/ dir
# cover-branches=1 # need nose 1.1.3 for this
#cover-html=1 # this works, puts html output in cover/ dir
#cover-branches=1 # need nose 1.1.3 for this
stop=1
verbosity=2

#tests=tests/test_nilmdb.py
#tests=tests/test_client.py
#with-profile=1
profile-sort=time
profile-restrict=10

+ 83
- 0
tests/test_client.py View File

@@ -0,0 +1,83 @@
import nilmdb
from nilmdb.printf import *

from nose.tools import *
from nose.tools import assert_raises
import json
import itertools
import os
import shutil
import sys
import threading
import urllib2
from urllib2 import urlopen, HTTPError
import Queue
import StringIO
import shlex

testdb = "tests/client-testdb"

def eq_(a, b):
if not a == b:
raise AssertionError("%r != %r" % (a, b))

def ne_(a, b):
if not a != b:
raise AssertionError("unexpected %r == %r" % (a, b))

class TestClient(object):

def setUp(self):
# Clear out DB
try:
shutil.rmtree(testdb)
except:
pass
try:
os.unlink(testdb)
except:
pass

# Start web app on a custom port
self.db = nilmdb.NilmDB(testdb)
self.server = nilmdb.Server(self.db, host = "127.0.0.1",
port = 12380, stoppable = False)
self.server.start(blocking = False)

def tearDown(self):
# Close web app
self.server.stop()
self.db.close()

def run(self, arg_string, input_string = None):
"""Run a client with the specified argument string, passing
the given input. Returns a tuple with the output and exit code"""
class stdio_wrapper:
def __init__(self, stdin, stdout, stderr):
self.io = (stdin, stdout, stderr)
def __enter__(self):
self.saved = ( sys.stdin, sys.stdout, sys.stderr )
( sys.stdin, sys.stdout, sys.stderr ) = self.io
def __exit__(self, type, value, traceback):
( sys.stdin, sys.stdout, sys.stderr ) = self.saved
infile = StringIO.StringIO(input_string)
outfile = StringIO.StringIO()
with stdio_wrapper(infile, outfile, outfile) as s:
try:
nilmdb.client.run(args = shlex.split(arg_string))
sys.exit(0)
except SystemExit as e:
exitcode = e.code
output = outfile.getvalue()
return (output, exitcode)

def test_client_basic(self):
(output, exitcode) = self.run("--help")
assert("Usage:" in output)
eq_(exitcode, 0)

(output, exitcode) = self.run("--nosuchoption")
ne_(exitcode, 0)

(output, exitcode) = self.run("--url http://localhost:12380/")
eq_(exitcode, 0)

Loading…
Cancel
Save