mirror of
https://github.com/laurivosandi/certidude
synced 2024-12-22 16:25:17 +00:00
Add test for fetching logs
This commit is contained in:
parent
0168b03f6b
commit
5ddbf87ed2
@ -179,7 +179,7 @@ class NormalizeMiddleware(object):
|
|||||||
if isinstance(resp.location, unicode):
|
if isinstance(resp.location, unicode):
|
||||||
resp.location = resp.location.encode("ascii")
|
resp.location = resp.location.encode("ascii")
|
||||||
|
|
||||||
def certidude_app():
|
def certidude_app(log_handlers=[]):
|
||||||
from certidude import config
|
from certidude import config
|
||||||
from .revoked import RevocationListResource
|
from .revoked import RevocationListResource
|
||||||
from .signed import SignedCertificateDetailResource
|
from .signed import SignedCertificateDetailResource
|
||||||
@ -225,4 +225,18 @@ def certidude_app():
|
|||||||
# Add sink for serving static files
|
# Add sink for serving static files
|
||||||
app.add_sink(StaticResource(os.path.join(__file__, "..", "..", "static")))
|
app.add_sink(StaticResource(os.path.join(__file__, "..", "..", "static")))
|
||||||
|
|
||||||
|
# Set up log handlers
|
||||||
|
if config.LOGGING_BACKEND == "sql":
|
||||||
|
from certidude.mysqllog import LogHandler
|
||||||
|
from certidude.api.log import LogResource
|
||||||
|
uri = config.cp.get("logging", "database")
|
||||||
|
log_handlers.append(LogHandler(uri))
|
||||||
|
app.add_route("/api/log/", LogResource(uri))
|
||||||
|
elif config.LOGGING_BACKEND == "syslog":
|
||||||
|
from logging.handlers import SyslogHandler
|
||||||
|
log_handlers.append(SysLogHandler())
|
||||||
|
# Browsing syslog via HTTP is obviously not possible out of the box
|
||||||
|
elif config.LOGGING_BACKEND:
|
||||||
|
raise ValueError("Invalid logging.backend = %s" % config.LOGGING_BACKEND)
|
||||||
|
|
||||||
return app
|
return app
|
||||||
|
@ -1204,7 +1204,7 @@ def certidude_serve(port, listen, fork):
|
|||||||
|
|
||||||
click.echo("Listening on %s:%d" % (listen, port))
|
click.echo("Listening on %s:%d" % (listen, port))
|
||||||
|
|
||||||
app = certidude_app()
|
app = certidude_app(log_handlers)
|
||||||
|
|
||||||
httpd = make_server(listen, port, app, ThreadingWSGIServer)
|
httpd = make_server(listen, port, app, ThreadingWSGIServer)
|
||||||
|
|
||||||
@ -1236,21 +1236,6 @@ def certidude_serve(port, listen, fork):
|
|||||||
|
|
||||||
os.umask(0o007)
|
os.umask(0o007)
|
||||||
|
|
||||||
|
|
||||||
# Set up log handlers
|
|
||||||
if config.LOGGING_BACKEND == "sql":
|
|
||||||
from certidude.mysqllog import LogHandler
|
|
||||||
from certidude.api.log import LogResource
|
|
||||||
uri = config.cp.get("logging", "database")
|
|
||||||
log_handlers.append(LogHandler(uri))
|
|
||||||
app.add_route("/api/log/", LogResource(uri))
|
|
||||||
elif config.LOGGING_BACKEND == "syslog":
|
|
||||||
from logging.handlers import SyslogHandler
|
|
||||||
log_handlers.append(SysLogHandler())
|
|
||||||
# Browsing syslog via HTTP is obviously not possible out of the box
|
|
||||||
elif config.LOGGING_BACKEND:
|
|
||||||
raise ValueError("Invalid logging.backend = %s" % config.LOGGING_BACKEND)
|
|
||||||
|
|
||||||
if config.EVENT_SOURCE_PUBLISH:
|
if config.EVENT_SOURCE_PUBLISH:
|
||||||
from certidude.push import EventSourceLogHandler
|
from certidude.push import EventSourceLogHandler
|
||||||
log_handlers.append(EventSourceLogHandler())
|
log_handlers.append(EventSourceLogHandler())
|
||||||
|
@ -52,11 +52,6 @@ def test_cli_setup_authority():
|
|||||||
result = runner.invoke(cli, ['users'])
|
result = runner.invoke(cli, ['users'])
|
||||||
assert not result.exception
|
assert not result.exception
|
||||||
|
|
||||||
|
|
||||||
# Try starting up forked server
|
|
||||||
result = runner.invoke(cli, ['serve', '-f', '-p', '8080'])
|
|
||||||
assert not result.exception
|
|
||||||
|
|
||||||
# Check that we can retrieve empty CRL
|
# Check that we can retrieve empty CRL
|
||||||
r = client().simulate_get("/api/revoked/")
|
r = client().simulate_get("/api/revoked/")
|
||||||
assert r.status_code == 200
|
assert r.status_code == 200
|
||||||
@ -264,4 +259,18 @@ def test_cli_setup_authority():
|
|||||||
assert r.status_code == 200
|
assert r.status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
# Log can be read only by admin
|
||||||
|
r = client().simulate_get("/api/log/")
|
||||||
|
assert r.status_code == 401
|
||||||
|
r = client().simulate_get("/api/log/",
|
||||||
|
headers={"Authorization":usertoken})
|
||||||
|
assert r.status_code == 403
|
||||||
|
r = client().simulate_get("/api/log/",
|
||||||
|
headers={"Authorization":admintoken})
|
||||||
|
assert r.status_code == 200
|
||||||
|
assert r.headers.get('content-type') == "application/json; charset=UTF-8"
|
||||||
|
|
||||||
|
# Try starting up forked server
|
||||||
|
result = runner.invoke(cli, ['serve', '-f', '-p', '8080'])
|
||||||
|
assert not result.exception
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user