Browse Source

Flesh out remaining tests and Python 3 fixes

tags/nilmtools-2.0.0
Jim Paris 11 months ago
parent
commit
9d49d39ff9
2 changed files with 79 additions and 5 deletions
  1. +7
    -4
      nilmtools/filter.py
  2. +72
    -1
      tests/test.py

+ 7
- 4
nilmtools/filter.py View File

@@ -173,21 +173,21 @@ class Filter(object):
self.end = None
self._interhost = False
self._force_metadata = False
self.def_url = os.environ.get("NILMDB_URL", "http://localhost/nilmdb/")
if parser_description is not None:
self.setup_parser(parser_description)
self.parse_args()
self.def_url = os.environ.get("NILMDB_URL", "http://localhost/nilmdb/")

@property
def client_src(self):
if self._using_client:
raise Exception("Filter client is in use; make another")
raise Exception("Filter src client is in use; make another")
return self._client_src

@property
def client_dest(self):
if self._using_client:
raise Exception("Filter client is in use; make another")
raise Exception("Filter dest client is in use; make another")
return self._client_dest

def setup_parser(self, description = "Filter data", skip_paths = False):
@@ -205,6 +205,9 @@ class Filter(object):
default = False,
help="Just print intervals that would be "
"processed")
group.add_argument("-q", "--quiet", action="store_true",
default = False,
help="Don't print source and dest stream info")
group.add_argument("-F", "--force-metadata", action="store_true",
default = False,
help="Force metadata changes if the dest "
@@ -269,7 +272,7 @@ class Filter(object):
args = self._parser.parse_args(argv)

self.set_args(args.url, args.dest_url, args.srcpath, args.destpath,
args.start, args.end, quiet = False, parsed_args = args)
args.start, args.end, quiet=args.quiet, parsed_args=args)

self._force_metadata = args.force_metadata
if args.dry_run:


+ 72
- 1
tests/test.py View File

@@ -28,6 +28,7 @@ import traceback
import os
import atexit
import signal
import functools

from urllib.request import urlopen
from nilmtools.filter import ArgumentError
@@ -834,7 +835,77 @@ class TestAllCommands(CommandTester):
self.ok(f"--estimate tests/data/cleanup-nodecim.cfg")

def test_12_misc(self):
# Fill in test cases that were missed by earlier code
# Fill in test cases that were missed by earlier code:

# math.py
with assert_raises(ValueError):
nilmtools.math.sfit4([1], 5)
nilmtools.math.sfit4([1,2], 5)

# filter.py
client = nilmdb.client.numpyclient.NumpyClient(self.url)
client.stream_create("/misc/a", "uint8_1")
client.stream_create("/misc/b", "uint8_1")
with client.stream_insert_context("/misc/a") as ctx:
for n in range(10000):
ctx.insert(b"%d 0\n" % n)
pni = nilmtools.filter.process_numpy_interval
src = nilmtools.filter.get_stream_info(client, "/misc/a")
extractor = functools.partial(
client.stream_extract_numpy, "/misc/a",
layout=src.layout, maxrows=1000)
inserter = functools.partial(
client.stream_insert_numpy_context, "/misc/b")
def func1(*args):
return 0
def func2(*args):
return -1
def func3(array, interval, args, insert_func, last):
if last:
return array.shape[0]
return 0
saved = (sys.stdout, sys.stderr)
try:
with open(os.devnull, 'w') as sys.stdout:
with open(os.devnull, 'w') as sys.stderr:
pni(Interval(0, 10000), extractor, inserter, 100, func1)
with assert_raises(SystemExit):
f = nilmtools.filter.Filter("hello world")
finally:
(sys.stdout, sys.stderr) = saved

with assert_raises(Exception):
pni(Interval(0, 10000), extractor, inserter, 100, func2)
pni(Interval(0, 10000), extractor, inserter, 100000, func3)

with assert_raises(NotImplementedError):
pni(Interval(0, 10000), extractor, inserter, 100000,
nilmtools.filter.example_callback_function)

self.main = nilmtools.filter.main
self.fail(f"")
self.ok(f"--help")

self.fail(f"/misc/a /misc/a")
self.contain("must be different")

self.fail(f"--start HELLOWORLD /misc/a /misc/a")
self.contain("not enough digits for a timestamp")

client.stream_create("/misc/c", "uint8_1")
self.ok(f"--quiet /misc/a /misc/c")
self.contain("Source: /misc/a", False)
self.contain("Generic filter: need to handle")

f = nilmtools.filter.Filter()
parser = f.setup_parser()
args = f.parse_args(["--quiet", "/misc/a", "/misc/c"])
x = f.client_src
x = f.client_dest
for i in f.intervals():
with assert_raises(Exception) as e:
x = f.client_src
in_("client is in use", str(e.exception))
with assert_raises(Exception) as e:
x = f.client_dest
in_("client is in use", str(e.exception))

Loading…
Cancel
Save