Unit tests for most of app.py, return 404 for unknown log entry.
This commit is contained in:
parent
3027aae142
commit
773ada882d
|
@ -1,5 +1,5 @@
|
|||
import logging, pprint, cStringIO
|
||||
from flask import Flask, jsonify, render_template, request
|
||||
from flask import Flask, jsonify, render_template, request, abort
|
||||
import version, rparse
|
||||
|
||||
logging.basicConfig(level="DEBUG")
|
||||
|
@ -58,7 +58,10 @@ def log():
|
|||
|
||||
@app.route('/log/<int:lid>')
|
||||
def onelog(lid):
|
||||
l = pprint.pformat(app.config["pathod"].log_by_id(int(lid)))
|
||||
item = app.config["pathod"].log_by_id(int(lid))
|
||||
if not item:
|
||||
abort(404)
|
||||
l = pprint.pformat(item)
|
||||
return render_template("onelog.html", section="log", alog=l, lid=lid)
|
||||
|
||||
|
||||
|
|
|
@ -1,7 +1,6 @@
|
|||
import json, threading, Queue
|
||||
import requests
|
||||
import pathod, utils
|
||||
import tutils
|
||||
|
||||
IFACE = "127.0.0.1"
|
||||
|
||||
|
|
|
@ -1,5 +1,4 @@
|
|||
import os, re
|
||||
import rparse
|
||||
|
||||
SIZE_UNITS = dict(
|
||||
b = 1024**0,
|
||||
|
|
|
@ -0,0 +1,23 @@
|
|||
import tutils
|
||||
|
||||
class TestApp(tutils.DaemonTests):
|
||||
SSL = False
|
||||
def test_index(self):
|
||||
r = self.getpath("/")
|
||||
assert r.status_code == 200
|
||||
assert r.content
|
||||
|
||||
def test_docs(self):
|
||||
assert self.getpath("/docs/pathod").status_code == 200
|
||||
assert self.getpath("/docs/pathoc").status_code == 200
|
||||
assert self.getpath("/docs/language").status_code == 200
|
||||
assert self.getpath("/docs/test").status_code == 200
|
||||
|
||||
def test_log(self):
|
||||
assert self.getpath("/log").status_code == 200
|
||||
assert self.get("200").status_code == 200
|
||||
id = self.d.log()[0]["id"]
|
||||
assert self.getpath("/log").status_code == 200
|
||||
assert self.getpath("/log/%s"%id).status_code == 200
|
||||
assert self.getpath("/log/9999999").status_code == 404
|
||||
|
|
@ -1,21 +1,7 @@
|
|||
import requests
|
||||
from libpathod import pathod, test, version, pathoc
|
||||
from libpathod import pathod, version
|
||||
from netlib import tcp, http
|
||||
import tutils
|
||||
|
||||
class _TestApplication:
|
||||
def test_anchors(self):
|
||||
a = pathod.PathodApp(staticdir=None)
|
||||
a.add_anchor("/foo", "200")
|
||||
assert a.get_anchors() == [("/foo", "200")]
|
||||
a.add_anchor("/bar", "400")
|
||||
assert a.get_anchors() == [("/bar", "400"), ("/foo", "200")]
|
||||
a.remove_anchor("/bar", "400")
|
||||
assert a.get_anchors() == [("/foo", "200")]
|
||||
a.remove_anchor("/oink", "400")
|
||||
assert a.get_anchors() == [("/foo", "200")]
|
||||
|
||||
|
||||
class TestPathod:
|
||||
def test_instantiation(self):
|
||||
p = pathod.Pathod(
|
||||
|
@ -40,40 +26,7 @@ class TestPathod:
|
|||
assert len(p.get_log()) <= p.LOGBUF
|
||||
|
||||
|
||||
class _DaemonTests:
|
||||
@classmethod
|
||||
def setUpAll(self):
|
||||
self.d = test.Daemon(
|
||||
staticdir=tutils.test_data.path("data"),
|
||||
anchors=[("/anchor/.*", "202")],
|
||||
ssl = self.SSL,
|
||||
sizelimit=1*1024*1024
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def tearDownAll(self):
|
||||
self.d.shutdown()
|
||||
|
||||
def setUp(self):
|
||||
self.d.clear_log()
|
||||
|
||||
def getpath(self, path):
|
||||
scheme = "https" if self.SSL else "http"
|
||||
return requests.get("%s://localhost:%s/%s"%(scheme, self.d.port, path), verify=False)
|
||||
|
||||
def get(self, spec):
|
||||
scheme = "https" if self.SSL else "http"
|
||||
return requests.get("%s://localhost:%s/p/%s"%(scheme, self.d.port, spec), verify=False)
|
||||
|
||||
def pathoc(self, spec, timeout=None):
|
||||
c = pathoc.Pathoc("localhost", self.d.port)
|
||||
c.connect()
|
||||
if self.SSL:
|
||||
c.convert_to_ssl()
|
||||
if timeout:
|
||||
c.settimeout(timeout)
|
||||
return c.request(spec)
|
||||
|
||||
class CommonTests(tutils.DaemonTests):
|
||||
def test_sizelimit(self):
|
||||
r = self.get("200:b@1g")
|
||||
assert r.status_code == 800
|
||||
|
@ -133,11 +86,11 @@ class _DaemonTests:
|
|||
assert rsp.status_code == 800
|
||||
|
||||
|
||||
class TestDaemon(_DaemonTests):
|
||||
class TestDaemon(CommonTests):
|
||||
SSL = False
|
||||
|
||||
|
||||
class TestDaemonSSL(_DaemonTests):
|
||||
class TestDaemonSSL(CommonTests):
|
||||
SSL = True
|
||||
def test_ssl_conn_failure(self):
|
||||
c = tcp.TCPClient("localhost", self.d.port)
|
||||
|
|
|
@ -1,6 +1,41 @@
|
|||
import tempfile, os, shutil
|
||||
from contextlib import contextmanager
|
||||
from libpathod import utils
|
||||
from libpathod import utils, test, pathoc
|
||||
import requests
|
||||
|
||||
class DaemonTests:
|
||||
@classmethod
|
||||
def setUpAll(self):
|
||||
self.d = test.Daemon(
|
||||
staticdir=test_data.path("data"),
|
||||
anchors=[("/anchor/.*", "202")],
|
||||
ssl = self.SSL,
|
||||
sizelimit=1*1024*1024
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def tearDownAll(self):
|
||||
self.d.shutdown()
|
||||
|
||||
def setUp(self):
|
||||
self.d.clear_log()
|
||||
|
||||
def getpath(self, path):
|
||||
scheme = "https" if self.SSL else "http"
|
||||
return requests.get("%s://localhost:%s/%s"%(scheme, self.d.port, path), verify=False)
|
||||
|
||||
def get(self, spec):
|
||||
scheme = "https" if self.SSL else "http"
|
||||
return requests.get("%s://localhost:%s/p/%s"%(scheme, self.d.port, spec), verify=False)
|
||||
|
||||
def pathoc(self, spec, timeout=None):
|
||||
c = pathoc.Pathoc("localhost", self.d.port)
|
||||
c.connect()
|
||||
if self.SSL:
|
||||
c.convert_to_ssl()
|
||||
if timeout:
|
||||
c.settimeout(timeout)
|
||||
return c.request(spec)
|
||||
|
||||
|
||||
|
||||
|
|
Loading…
Reference in New Issue