forked from s3lph/matemat
Merge branch 'jinja2-template' into 'webserver-impl'
Jinja2 Template support, more explicit pagelet return API See merge request s3lph/matemat!9
This commit is contained in:
commit
652520dc85
16 changed files with 275 additions and 147 deletions
|
@ -19,6 +19,7 @@ This project intends to provide a well-tested and maintainable alternative to
|
||||||
- Python 3 (>=3.6)
|
- Python 3 (>=3.6)
|
||||||
- Python dependencies:
|
- Python dependencies:
|
||||||
- apsw
|
- apsw
|
||||||
|
- jinja2
|
||||||
|
|
||||||
## Usage
|
## Usage
|
||||||
|
|
||||||
|
|
2
doc
2
doc
|
@ -1 +1 @@
|
||||||
Subproject commit 51e940460ddbaebb7f2ffc48d00d9ef19cf8d33f
|
Subproject commit 9634785e5621b324f72598b3cee83bbc45c25d7b
|
|
@ -7,4 +7,5 @@ server will attempt to serve the request with a static resource in a previously
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from .requestargs import RequestArgument, RequestArguments
|
from .requestargs import RequestArgument, RequestArguments
|
||||||
|
from .responses import PageletResponse, RedirectResponse, TemplateResponse
|
||||||
from .httpd import MatematWebserver, HttpHandler, pagelet
|
from .httpd import MatematWebserver, HttpHandler, pagelet
|
||||||
|
|
|
@ -12,9 +12,11 @@ from http.cookies import SimpleCookie
|
||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
import jinja2
|
||||||
|
|
||||||
from matemat import __version__ as matemat_version
|
from matemat import __version__ as matemat_version
|
||||||
from matemat.exceptions import HttpException
|
from matemat.exceptions import HttpException
|
||||||
from matemat.webserver import RequestArguments
|
from matemat.webserver import RequestArguments, PageletResponse, RedirectResponse, TemplateResponse
|
||||||
from matemat.webserver.util import parse_args
|
from matemat.webserver.util import parse_args
|
||||||
|
|
||||||
#
|
#
|
||||||
|
@ -35,7 +37,7 @@ _PAGELET_PATHS: Dict[str, Callable[[str, # HTTP method (GET, POST, ...)
|
||||||
Dict[str, str]], # Response headers
|
Dict[str, str]], # Response headers
|
||||||
Union[ # Return type: either a response body, or a redirect
|
Union[ # Return type: either a response body, or a redirect
|
||||||
bytes, str, # Response body: will assign HTTP/1.0 200 OK
|
bytes, str, # Response body: will assign HTTP/1.0 200 OK
|
||||||
Tuple[int, str] # Redirect: First element must be 301, second the redirect path
|
PageletResponse, # A generic response
|
||||||
]]] = dict()
|
]]] = dict()
|
||||||
|
|
||||||
# Inactivity timeout for client sessions
|
# Inactivity timeout for client sessions
|
||||||
|
@ -64,7 +66,8 @@ def pagelet(path: str):
|
||||||
headers: The dictionary of HTTP response headers. Add headers you wish to send with the response.
|
headers: The dictionary of HTTP response headers. Add headers you wish to send with the response.
|
||||||
returns: One of the following:
|
returns: One of the following:
|
||||||
- A HTTP Response body as str or bytes
|
- A HTTP Response body as str or bytes
|
||||||
- A HTTP redirect: A tuple of 301 (an int) and the path to redirect to (a str)
|
- A PageletResponse class instance: An instance of (a subclass of)
|
||||||
|
matemat.webserver.PageletResponse, e.g. encapsulating a redirect or a Jinja2 template.
|
||||||
raises: HttpException: If a non-200 HTTP status code should be returned
|
raises: HttpException: If a non-200 HTTP status code should be returned
|
||||||
|
|
||||||
:param path: The path to register the function for.
|
:param path: The path to register the function for.
|
||||||
|
@ -77,7 +80,7 @@ def pagelet(path: str):
|
||||||
Dict[str, str]],
|
Dict[str, str]],
|
||||||
Union[
|
Union[
|
||||||
bytes, str,
|
bytes, str,
|
||||||
Tuple[int, str]
|
PageletResponse
|
||||||
]]):
|
]]):
|
||||||
# Add the function to the dict of pagelets
|
# Add the function to the dict of pagelets
|
||||||
_PAGELET_PATHS[path] = fun
|
_PAGELET_PATHS[path] = fun
|
||||||
|
@ -97,13 +100,18 @@ class MatematHTTPServer(HTTPServer):
|
||||||
def __init__(self,
|
def __init__(self,
|
||||||
server_address: Any,
|
server_address: Any,
|
||||||
handler: Type[BaseHTTPRequestHandler],
|
handler: Type[BaseHTTPRequestHandler],
|
||||||
webroot: str,
|
staticroot: str,
|
||||||
|
templateroot: str,
|
||||||
bind_and_activate: bool = True) -> None:
|
bind_and_activate: bool = True) -> None:
|
||||||
super().__init__(server_address, handler, bind_and_activate)
|
super().__init__(server_address, handler, bind_and_activate)
|
||||||
# Resolve webroot directory
|
# Resolve webroot directory
|
||||||
self.webroot = os.path.abspath(webroot)
|
self.webroot = os.path.abspath(staticroot)
|
||||||
# Set up session vars dict
|
# Set up session vars dict
|
||||||
self.session_vars: Dict[str, Tuple[datetime, Dict[str, Any]]] = dict()
|
self.session_vars: Dict[str, Tuple[datetime, Dict[str, Any]]] = dict()
|
||||||
|
# Set up the Jinja2 environment
|
||||||
|
self.jinja_env: jinja2.Environment = jinja2.Environment(
|
||||||
|
loader=jinja2.FileSystemLoader(os.path.abspath(templateroot))
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class MatematWebserver(object):
|
class MatematWebserver(object):
|
||||||
|
@ -121,13 +129,18 @@ class MatematWebserver(object):
|
||||||
server.start()
|
server.start()
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, listen: str = '::', port: int = 80, webroot: str = './webroot') -> None:
|
def __init__(self,
|
||||||
|
listen: str = '::',
|
||||||
|
port: int = 80,
|
||||||
|
staticroot: str = './static',
|
||||||
|
templateroot: str = './templates') -> None:
|
||||||
"""
|
"""
|
||||||
Instantiate a MatematWebserver.
|
Instantiate a MatematWebserver.
|
||||||
|
|
||||||
:param listen: The IPv4 or IPv6 address to listen on
|
:param listen: The IPv4 or IPv6 address to listen on.
|
||||||
:param port: The TCP port to listen on
|
:param port: The TCP port to listen on.
|
||||||
:param webroot: Path to the webroot directory
|
:param staticroot: Path to the static webroot directory.
|
||||||
|
:param templateroot: Path to the Jinja2 templates root directory.
|
||||||
"""
|
"""
|
||||||
if len(listen) == 0:
|
if len(listen) == 0:
|
||||||
# Empty string should be interpreted as all addresses
|
# Empty string should be interpreted as all addresses
|
||||||
|
@ -137,7 +150,7 @@ class MatematWebserver(object):
|
||||||
# Rewrite IPv4 address to IPv6-mapped form
|
# Rewrite IPv4 address to IPv6-mapped form
|
||||||
listen = f'::ffff:{listen}'
|
listen = f'::ffff:{listen}'
|
||||||
# Create the http server
|
# Create the http server
|
||||||
self._httpd = MatematHTTPServer((listen, port), HttpHandler, webroot)
|
self._httpd = MatematHTTPServer((listen, port), HttpHandler, staticroot, templateroot)
|
||||||
|
|
||||||
def start(self) -> None:
|
def start(self) -> None:
|
||||||
"""
|
"""
|
||||||
|
@ -202,8 +215,11 @@ class HttpHandler(BaseHTTPRequestHandler):
|
||||||
if session_id in self.server.session_vars:
|
if session_id in self.server.session_vars:
|
||||||
del self.server.session_vars[session_id]
|
del self.server.session_vars[session_id]
|
||||||
|
|
||||||
@staticmethod
|
def _parse_pagelet_result(self,
|
||||||
def _parse_pagelet_result(pagelet_res: Union[bytes, str, Tuple[int, str]], headers: Dict[str, str]) \
|
pagelet_res: Union[bytes, # Response body as bytes
|
||||||
|
str, # Response body as str
|
||||||
|
PageletResponse], # Encapsulated or unresolved response body
|
||||||
|
headers: Dict[str, str]) \
|
||||||
-> Tuple[int, bytes]:
|
-> Tuple[int, bytes]:
|
||||||
"""
|
"""
|
||||||
Process the return value of a pagelet function call.
|
Process the return value of a pagelet function call.
|
||||||
|
@ -215,29 +231,34 @@ class HttpHandler(BaseHTTPRequestHandler):
|
||||||
"""
|
"""
|
||||||
# The HTTP Response Status Code, defaults to 200 OK
|
# The HTTP Response Status Code, defaults to 200 OK
|
||||||
hsc: int = 200
|
hsc: int = 200
|
||||||
# The HTTP Response body, defaults to None
|
# The HTTP Response body, defaults to empty
|
||||||
data: Union[bytes, str] = None
|
data: bytes = bytes()
|
||||||
if isinstance(pagelet_res, tuple):
|
|
||||||
# If the return type is a tuple, the first element must be 301 (the HTTP Redirect status code)
|
# If the response is a bytes object, it is used without further modification
|
||||||
head, tail = pagelet_res
|
if isinstance(pagelet_res, bytes):
|
||||||
if head == 301:
|
|
||||||
# Set the HTTP Response Status Code, and the redirect header
|
|
||||||
hsc = 301
|
|
||||||
headers['Location'] = tail
|
|
||||||
else:
|
|
||||||
raise TypeError(f'Return value of pagelet not understood: {pagelet_res}')
|
|
||||||
elif isinstance(pagelet_res, str) or isinstance(pagelet_res, bytes):
|
|
||||||
# Return value is a response body
|
|
||||||
data = pagelet_res
|
data = pagelet_res
|
||||||
|
|
||||||
|
# If the response is a str object, it is encoded into a bytes object
|
||||||
|
elif isinstance(pagelet_res, str):
|
||||||
|
data = pagelet_res.encode('utf-8')
|
||||||
|
|
||||||
|
# If the response is a PageletResponse object, the status code is extracted. Generation of the body depends
|
||||||
|
# on the subtype
|
||||||
|
elif isinstance(pagelet_res, PageletResponse):
|
||||||
|
hsc = pagelet_res.status
|
||||||
|
|
||||||
|
# If the object is a RedirectResponse instance, no body is needed
|
||||||
|
if isinstance(pagelet_res, RedirectResponse):
|
||||||
|
headers['Location'] = pagelet_res.location
|
||||||
|
# If the object is a TemplateRespinse instance, pass the Jinja2 environment instance for rendering
|
||||||
|
elif isinstance(pagelet_res, TemplateResponse):
|
||||||
|
# noinspection PyProtectedMember
|
||||||
|
data = pagelet_res._render(self.server.jinja_env)
|
||||||
|
# else: Empty body
|
||||||
|
|
||||||
else:
|
else:
|
||||||
# Return value is not a response body or a redirect
|
|
||||||
raise TypeError(f'Return value of pagelet not understood: {pagelet_res}')
|
raise TypeError(f'Return value of pagelet not understood: {pagelet_res}')
|
||||||
# The pagelet may return None as data as a shorthand for an empty response
|
|
||||||
if data is None:
|
|
||||||
data = bytes()
|
|
||||||
# If the pagelet returns a Python str, convert it to an UTF-8 encoded bytes object
|
|
||||||
if isinstance(data, str):
|
|
||||||
data = data.encode('utf-8')
|
|
||||||
# Return the resulting status code and body
|
# Return the resulting status code and body
|
||||||
return hsc, data
|
return hsc, data
|
||||||
|
|
||||||
|
@ -271,7 +292,7 @@ class HttpHandler(BaseHTTPRequestHandler):
|
||||||
# Call the pagelet function
|
# Call the pagelet function
|
||||||
pagelet_res = _PAGELET_PATHS[path](method, path, args, self.session_vars, headers)
|
pagelet_res = _PAGELET_PATHS[path](method, path, args, self.session_vars, headers)
|
||||||
# Parse the pagelet's return value, vielding a HTTP status code and a response body
|
# Parse the pagelet's return value, vielding a HTTP status code and a response body
|
||||||
hsc, data = HttpHandler._parse_pagelet_result(pagelet_res, headers)
|
hsc, data = self._parse_pagelet_result(pagelet_res, headers)
|
||||||
# Send the HTTP status code
|
# Send the HTTP status code
|
||||||
self.send_response(hsc)
|
self.send_response(hsc)
|
||||||
# Format the session cookie timeout string and send the session cookie header
|
# Format the session cookie timeout string and send the session cookie header
|
||||||
|
|
|
@ -1,8 +1,8 @@
|
||||||
|
|
||||||
from typing import Any, Dict, Tuple, Union
|
from typing import Any, Dict, Union
|
||||||
|
|
||||||
from matemat.exceptions import AuthenticationError, HttpException
|
from matemat.exceptions import AuthenticationError, HttpException
|
||||||
from matemat.webserver import pagelet, RequestArguments
|
from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse
|
||||||
from matemat.primitives import User
|
from matemat.primitives import User
|
||||||
from matemat.db import MatematDatabase
|
from matemat.db import MatematDatabase
|
||||||
|
|
||||||
|
@ -13,40 +13,17 @@ def login_page(method: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str])\
|
headers: Dict[str, str])\
|
||||||
-> Union[bytes, str, Tuple[int, str]]:
|
-> Union[bytes, str, PageletResponse]:
|
||||||
if 'user' in session_vars:
|
if 'user' in session_vars:
|
||||||
return 301, '/'
|
return RedirectResponse('/')
|
||||||
if method == 'GET':
|
if method == 'GET':
|
||||||
data = '''
|
return TemplateResponse('login.html')
|
||||||
<DOCTYPE html>
|
|
||||||
<html>
|
|
||||||
<head>
|
|
||||||
<title>Matemat</title>
|
|
||||||
<style>
|
|
||||||
body {{
|
|
||||||
color: #f0f0f0;
|
|
||||||
background: #000000;
|
|
||||||
}};
|
|
||||||
</style>
|
|
||||||
</head>
|
|
||||||
<body>
|
|
||||||
<h1>Matemat</h1>
|
|
||||||
{msg}
|
|
||||||
<form action="/login" method="post">
|
|
||||||
Username: <input type="text" name="username"/><br/>
|
|
||||||
Password: <input type="password" name="password" /><br/>
|
|
||||||
<input type="submit" value="Login"/>
|
|
||||||
</form>
|
|
||||||
</body>
|
|
||||||
</html>
|
|
||||||
'''
|
|
||||||
return data.format(msg=str(args.msg) if 'msg' in args else '')
|
|
||||||
elif method == 'POST':
|
elif method == 'POST':
|
||||||
with MatematDatabase('test.db') as db:
|
with MatematDatabase('test.db') as db:
|
||||||
try:
|
try:
|
||||||
user: User = db.login(str(args.username), str(args.password))
|
user: User = db.login(str(args.username), str(args.password))
|
||||||
except AuthenticationError:
|
except AuthenticationError:
|
||||||
return 301, '/login?msg=Username%20or%20password%20wrong.%20Please%20try%20again.'
|
return RedirectResponse('/login')
|
||||||
session_vars['user'] = user
|
session_vars['user'] = user
|
||||||
return 301, '/'
|
return RedirectResponse('/')
|
||||||
raise HttpException(405)
|
raise HttpException(405)
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
|
|
||||||
from typing import Any, Dict, List, Optional, Tuple, Union
|
from typing import Any, Dict, Union
|
||||||
|
|
||||||
from matemat.webserver import pagelet, RequestArguments
|
from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse
|
||||||
|
|
||||||
|
|
||||||
@pagelet('/logout')
|
@pagelet('/logout')
|
||||||
|
@ -10,7 +10,7 @@ def logout(method: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str])\
|
headers: Dict[str, str])\
|
||||||
-> Union[bytes, str, Tuple[int, str]]:
|
-> Union[bytes, str, PageletResponse]:
|
||||||
if 'user' in session_vars:
|
if 'user' in session_vars:
|
||||||
del session_vars['user']
|
del session_vars['user']
|
||||||
return 301, '/'
|
return RedirectResponse('/')
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
|
|
||||||
from typing import Any, Dict, Optional, Tuple, Union
|
from typing import Any, Dict, Union
|
||||||
|
|
||||||
from matemat.webserver import pagelet, RequestArguments
|
from matemat.webserver import pagelet, RequestArguments, PageletResponse, TemplateResponse
|
||||||
from matemat.primitives import User
|
from matemat.primitives import User
|
||||||
from matemat.db import MatematDatabase
|
from matemat.db import MatematDatabase
|
||||||
|
|
||||||
|
@ -12,41 +12,12 @@ def main_page(method: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str])\
|
headers: Dict[str, str])\
|
||||||
-> Union[bytes, str, Tuple[int, str]]:
|
-> Union[bytes, str, PageletResponse]:
|
||||||
data = '''
|
|
||||||
<DOCTYPE html>
|
|
||||||
<html>
|
|
||||||
<head>
|
|
||||||
<title>Matemat</title>
|
|
||||||
<style>
|
|
||||||
body {{
|
|
||||||
color: #f0f0f0;
|
|
||||||
background: #000000;
|
|
||||||
}};
|
|
||||||
</style>
|
|
||||||
</head>
|
|
||||||
<body>
|
|
||||||
<h1>Matemat</h1>
|
|
||||||
{user}
|
|
||||||
<ul>
|
|
||||||
{list}
|
|
||||||
</ul>
|
|
||||||
</body>
|
|
||||||
</html>
|
|
||||||
'''
|
|
||||||
with MatematDatabase('test.db') as db:
|
with MatematDatabase('test.db') as db:
|
||||||
if 'user' in session_vars:
|
if 'user' in session_vars:
|
||||||
user: User = session_vars['user']
|
user: User = session_vars['user']
|
||||||
products = db.list_products()
|
products = db.list_products()
|
||||||
plist = '\n'.join([f'<li/> <b>{p.name}</b> ' +
|
return TemplateResponse('main.html', user=user, list=products)
|
||||||
f'{p.price_member//100 if user.is_member else p.price_non_member//100}' +
|
|
||||||
f'.{p.price_member%100 if user.is_member else p.price_non_member%100}'
|
|
||||||
for p in products])
|
|
||||||
uname = f'<b>{user.name}</b> <a href="/logout">(Logout)</a>'
|
|
||||||
data = data.format(user=uname, list=plist)
|
|
||||||
else:
|
else:
|
||||||
users = db.list_users()
|
users = db.list_users()
|
||||||
ulist = '\n'.join([f'<li/> <b><a href=/touchkey?username={u.name}>{u.name}</a></b>' for u in users])
|
return TemplateResponse('main.html', list=users)
|
||||||
ulist = ulist + '<li/> <a href=/login>Password login</a>'
|
|
||||||
data = data.format(user='', list=ulist)
|
|
||||||
return data
|
|
||||||
|
|
|
@ -1,8 +1,10 @@
|
||||||
|
|
||||||
from typing import Any, Dict, Tuple, Union
|
from typing import Any, Dict, Tuple, Union
|
||||||
|
|
||||||
|
import urllib.parse
|
||||||
|
|
||||||
from matemat.exceptions import AuthenticationError, HttpException
|
from matemat.exceptions import AuthenticationError, HttpException
|
||||||
from matemat.webserver import pagelet, RequestArguments
|
from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse
|
||||||
from matemat.primitives import User
|
from matemat.primitives import User
|
||||||
from matemat.db import MatematDatabase
|
from matemat.db import MatematDatabase
|
||||||
|
|
||||||
|
@ -13,39 +15,18 @@ def touchkey_page(method: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str])\
|
headers: Dict[str, str])\
|
||||||
-> Union[bytes, str, Tuple[int, str]]:
|
-> Union[bytes, str, PageletResponse]:
|
||||||
if 'user' in session_vars:
|
if 'user' in session_vars:
|
||||||
return 301, '/'
|
return RedirectResponse('/')
|
||||||
if method == 'GET':
|
if method == 'GET':
|
||||||
data = '''
|
return TemplateResponse('touchkey.html', username=str(args.username) if 'username' in args else None)
|
||||||
<DOCTYPE html>
|
|
||||||
<html>
|
|
||||||
<head>
|
|
||||||
<title>Matemat</title>
|
|
||||||
<style>
|
|
||||||
body {{
|
|
||||||
color: #f0f0f0;
|
|
||||||
background: #000000;
|
|
||||||
}};
|
|
||||||
</style>
|
|
||||||
</head>
|
|
||||||
<body>
|
|
||||||
<h1>Matemat</h1>
|
|
||||||
<form action="/touchkey" method="post">
|
|
||||||
<input type="hidden" name="username" value="{username}"/><br/>
|
|
||||||
Touchkey: <input type="password" name="touchkey" /><br/>
|
|
||||||
<input type="submit" value="Login"/>
|
|
||||||
</form>
|
|
||||||
</body>
|
|
||||||
</html>
|
|
||||||
'''
|
|
||||||
return data.format(username=str(args.username) if 'username' in args else '')
|
|
||||||
elif method == 'POST':
|
elif method == 'POST':
|
||||||
with MatematDatabase('test.db') as db:
|
with MatematDatabase('test.db') as db:
|
||||||
try:
|
try:
|
||||||
user: User = db.login(str(args.username), touchkey=str(args.touchkey))
|
user: User = db.login(str(args.username), touchkey=str(args.touchkey))
|
||||||
except AuthenticationError:
|
except AuthenticationError:
|
||||||
return 301, f'/touchkey?username={args["username"]}&msg=Please%20try%20again.'
|
quoted = urllib.parse.quote_plus(bytes(args.username))
|
||||||
|
return RedirectResponse(f'/touchkey?username={quoted}')
|
||||||
session_vars['user'] = user
|
session_vars['user'] = user
|
||||||
return 301, '/'
|
return RedirectResponse('/')
|
||||||
raise HttpException(405)
|
raise HttpException(405)
|
||||||
|
|
65
matemat/webserver/responses.py
Normal file
65
matemat/webserver/responses.py
Normal file
|
@ -0,0 +1,65 @@
|
||||||
|
|
||||||
|
from jinja2 import Environment, Template
|
||||||
|
|
||||||
|
from matemat import __version__
|
||||||
|
|
||||||
|
|
||||||
|
class PageletResponse:
|
||||||
|
"""
|
||||||
|
Base class for pagelet return values that require more action than simply sending plain data.
|
||||||
|
|
||||||
|
An instance of this base class will result in an empty 200 OK response.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, status: int = 200):
|
||||||
|
"""
|
||||||
|
Create an empty response.
|
||||||
|
|
||||||
|
:param status: The HTTP status code, defaults to 200 (OK).
|
||||||
|
"""
|
||||||
|
self.status: int = status
|
||||||
|
|
||||||
|
|
||||||
|
class RedirectResponse(PageletResponse):
|
||||||
|
"""
|
||||||
|
A pagelet response that causes the server to redirect to another location, using a 301 Permanently Moved (uncached)
|
||||||
|
response status, and a Location header.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, location: str):
|
||||||
|
"""
|
||||||
|
Create a redirection response with the given redirection location.
|
||||||
|
|
||||||
|
:param location: The location to redirect to.
|
||||||
|
"""
|
||||||
|
super().__init__(status=301)
|
||||||
|
self.location: str = location
|
||||||
|
|
||||||
|
|
||||||
|
class TemplateResponse(PageletResponse):
|
||||||
|
"""
|
||||||
|
A pagelet response that causes the server to load a Jinja2 template and render it with the provided arguments, then
|
||||||
|
sending the result as response body, with a 200 OK response status.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, name: str, **kwargs):
|
||||||
|
"""
|
||||||
|
Create a template response with the given template name and arguments.
|
||||||
|
|
||||||
|
:param name: Name of the template to load.
|
||||||
|
:param kwargs: Arguments for rendering the template, will be passed to jinja2.Template.render as is.
|
||||||
|
"""
|
||||||
|
super().__init__()
|
||||||
|
self.name: str = name
|
||||||
|
self.kwargs = kwargs
|
||||||
|
|
||||||
|
def _render(self, jinja_env: Environment) -> bytes:
|
||||||
|
"""
|
||||||
|
Load and render the template using the Jinja2 environment managed by the web server instance. This method
|
||||||
|
should not be called by a pagelet.
|
||||||
|
|
||||||
|
:param jinja_env: The Jinja2 environment.
|
||||||
|
:return: An UTF-8 encoded bytes object containing the template rendering result.
|
||||||
|
"""
|
||||||
|
template: Template = jinja_env.get_template(self.name)
|
||||||
|
return template.render(**self.kwargs, __version__=__version__).encode('utf-8')
|
|
@ -9,6 +9,8 @@ from abc import ABC
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from http.server import HTTPServer
|
from http.server import HTTPServer
|
||||||
|
|
||||||
|
import jinja2
|
||||||
|
|
||||||
from matemat.webserver import pagelet, RequestArguments
|
from matemat.webserver import pagelet, RequestArguments
|
||||||
|
|
||||||
|
|
||||||
|
@ -107,6 +109,10 @@ class MockServer:
|
||||||
self.session_vars: Dict[str, Tuple[datetime, Dict[str, Any]]] = dict()
|
self.session_vars: Dict[str, Tuple[datetime, Dict[str, Any]]] = dict()
|
||||||
# Webroot for statically served content
|
# Webroot for statically served content
|
||||||
self.webroot: str = webroot
|
self.webroot: str = webroot
|
||||||
|
# Jinja environment with a single, static template
|
||||||
|
self.jinja_env = jinja2.Environment(
|
||||||
|
loader=jinja2.DictLoader({'test.txt': 'Hello, {{ what }}!'})
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class MockSocket(bytes):
|
class MockSocket(bytes):
|
||||||
|
|
|
@ -1,21 +1,31 @@
|
||||||
|
|
||||||
from typing import Any, Dict
|
from typing import Any, Dict, Union
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import os.path
|
import os.path
|
||||||
from matemat.exceptions import HttpException
|
from matemat.exceptions import HttpException
|
||||||
from matemat.webserver import HttpHandler, RequestArguments
|
from matemat.webserver import HttpHandler, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse
|
||||||
from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet
|
from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet
|
||||||
|
|
||||||
|
|
||||||
@test_pagelet('/just/testing/serve_pagelet_ok')
|
@test_pagelet('/just/testing/serve_pagelet_str')
|
||||||
def serve_test_pagelet_ok(method: str,
|
def serve_test_pagelet_str(method: str,
|
||||||
path: str,
|
path: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str]):
|
headers: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
|
||||||
headers['Content-Type'] = 'text/plain'
|
headers['Content-Type'] = 'text/plain'
|
||||||
return 'serve test pagelet ok'
|
return 'serve test pagelet str'
|
||||||
|
|
||||||
|
|
||||||
|
@test_pagelet('/just/testing/serve_pagelet_bytes')
|
||||||
|
def serve_test_pagelet_bytes(method: str,
|
||||||
|
path: str,
|
||||||
|
args: RequestArguments,
|
||||||
|
session_vars: Dict[str, Any],
|
||||||
|
headers: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
|
||||||
|
headers['Content-Type'] = 'application/octet-stream'
|
||||||
|
return b'serve\x80test\xffpagelet\xfebytes'
|
||||||
|
|
||||||
|
|
||||||
@test_pagelet('/just/testing/serve_pagelet_redirect')
|
@test_pagelet('/just/testing/serve_pagelet_redirect')
|
||||||
|
@ -23,16 +33,27 @@ def serve_test_pagelet_redirect(method: str,
|
||||||
path: str,
|
path: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str]):
|
headers: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
|
||||||
return 301, '/foo/bar'
|
return RedirectResponse('/foo/bar')
|
||||||
|
|
||||||
|
|
||||||
|
@test_pagelet('/just/testing/serve_pagelet_template')
|
||||||
|
def serve_test_pagelet_template(method: str,
|
||||||
|
path: str,
|
||||||
|
args: RequestArguments,
|
||||||
|
session_vars: Dict[str, Any],
|
||||||
|
headers: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
|
||||||
|
headers['Content-Type'] = 'text/plain'
|
||||||
|
return TemplateResponse('test.txt', what='World')
|
||||||
|
|
||||||
|
|
||||||
|
# noinspection PyTypeChecker
|
||||||
@test_pagelet('/just/testing/serve_pagelet_fail')
|
@test_pagelet('/just/testing/serve_pagelet_fail')
|
||||||
def serve_test_pagelet_fail(method: str,
|
def serve_test_pagelet_fail(method: str,
|
||||||
path: str,
|
path: str,
|
||||||
args: RequestArguments,
|
args: RequestArguments,
|
||||||
session_vars: Dict[str, Any],
|
session_vars: Dict[str, Any],
|
||||||
headers: Dict[str, str]):
|
headers: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
|
||||||
session_vars['test'] = 'hello, world!'
|
session_vars['test'] = 'hello, world!'
|
||||||
headers['Content-Type'] = 'text/plain'
|
headers['Content-Type'] = 'text/plain'
|
||||||
raise HttpException()
|
raise HttpException()
|
||||||
|
@ -54,17 +75,29 @@ class TestServe(AbstractHttpdTest):
|
||||||
f.write('This should not be readable')
|
f.write('This should not be readable')
|
||||||
os.chmod(forbidden, 0)
|
os.chmod(forbidden, 0)
|
||||||
|
|
||||||
def test_serve_pagelet_ok(self):
|
def test_serve_pagelet_str(self):
|
||||||
# Call the test pagelet that produces a 200 OK result
|
# Call the test pagelet that produces a 200 OK result
|
||||||
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_ok HTTP/1.1\r\n\r\n')
|
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_str HTTP/1.1\r\n\r\n')
|
||||||
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
||||||
packet = self.client_sock.get_response()
|
packet = self.client_sock.get_response()
|
||||||
|
|
||||||
# Make sure the correct pagelet was called
|
# Make sure the correct pagelet was called
|
||||||
self.assertEqual('serve_test_pagelet_ok', packet.pagelet)
|
self.assertEqual('serve_test_pagelet_str', packet.pagelet)
|
||||||
# Make sure the expected content is served
|
# Make sure the expected content is served
|
||||||
self.assertEqual(200, packet.statuscode)
|
self.assertEqual(200, packet.statuscode)
|
||||||
self.assertEqual(b'serve test pagelet ok', packet.body)
|
self.assertEqual(b'serve test pagelet str', packet.body)
|
||||||
|
|
||||||
|
def test_serve_pagelet_bytes(self):
|
||||||
|
# Call the test pagelet that produces a 200 OK result
|
||||||
|
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_bytes HTTP/1.1\r\n\r\n')
|
||||||
|
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
||||||
|
packet = self.client_sock.get_response()
|
||||||
|
|
||||||
|
# Make sure the correct pagelet was called
|
||||||
|
self.assertEqual('serve_test_pagelet_bytes', packet.pagelet)
|
||||||
|
# Make sure the expected content is served
|
||||||
|
self.assertEqual(200, packet.statuscode)
|
||||||
|
self.assertEqual(b'serve\x80test\xffpagelet\xfebytes', packet.body)
|
||||||
|
|
||||||
def test_serve_pagelet_fail(self):
|
def test_serve_pagelet_fail(self):
|
||||||
# Call the test pagelet that produces a 500 Internal Server Error result
|
# Call the test pagelet that produces a 500 Internal Server Error result
|
||||||
|
@ -81,12 +114,26 @@ class TestServe(AbstractHttpdTest):
|
||||||
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
||||||
packet = self.client_sock.get_response()
|
packet = self.client_sock.get_response()
|
||||||
|
|
||||||
|
# Make sure the correct pagelet was called
|
||||||
|
self.assertEqual('serve_test_pagelet_redirect', packet.pagelet)
|
||||||
# Make sure the correct redirect is issued
|
# Make sure the correct redirect is issued
|
||||||
self.assertEqual(301, packet.statuscode)
|
self.assertEqual(301, packet.statuscode)
|
||||||
self.assertEqual('/foo/bar', packet.headers['Location'])
|
self.assertEqual('/foo/bar', packet.headers['Location'])
|
||||||
# Make sure the response body is empty
|
# Make sure the response body is empty
|
||||||
self.assertEqual(0, len(packet.body))
|
self.assertEqual(0, len(packet.body))
|
||||||
|
|
||||||
|
def test_serve_pagelet_template(self):
|
||||||
|
# Call the test pagelet that redirects to another path
|
||||||
|
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_template HTTP/1.1\r\n\r\n')
|
||||||
|
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
||||||
|
packet = self.client_sock.get_response()
|
||||||
|
|
||||||
|
# Make sure the correct pagelet was called
|
||||||
|
self.assertEqual('serve_test_pagelet_template', packet.pagelet)
|
||||||
|
self.assertEqual(200, packet.statuscode)
|
||||||
|
# Make sure the response body was rendered correctly by the templating engine
|
||||||
|
self.assertEqual(b'Hello, World!', packet.body)
|
||||||
|
|
||||||
def test_serve_static_ok(self):
|
def test_serve_static_ok(self):
|
||||||
# Request a static resource
|
# Request a static resource
|
||||||
self.client_sock.set_request(b'GET /static_resource.txt HTTP/1.1\r\n\r\n')
|
self.client_sock.set_request(b'GET /static_resource.txt HTTP/1.1\r\n\r\n')
|
||||||
|
|
|
@ -1 +1,2 @@
|
||||||
apsw
|
apsw
|
||||||
|
jinja2
|
||||||
|
|
17
templates/base.html
Normal file
17
templates/base.html
Normal file
|
@ -0,0 +1,17 @@
|
||||||
|
<!DOCTYPE html>
|
||||||
|
<html>
|
||||||
|
<head>
|
||||||
|
<title>Matemat</title>
|
||||||
|
<style>
|
||||||
|
body {
|
||||||
|
color: #f0f0f0;
|
||||||
|
background: #000000;
|
||||||
|
}
|
||||||
|
</style>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
<h1>Matemat {{__version__}}</h1>
|
||||||
|
{% block main %}
|
||||||
|
{% endblock %}
|
||||||
|
</body>
|
||||||
|
</html>
|
9
templates/login.html
Normal file
9
templates/login.html
Normal file
|
@ -0,0 +1,9 @@
|
||||||
|
{% extends "base.html" %}
|
||||||
|
|
||||||
|
{% block main %}
|
||||||
|
<form action="/login" method="post">
|
||||||
|
Username: <input type="text" name="username"/><br/>
|
||||||
|
Password: <input type="password" name="password" /><br/>
|
||||||
|
<input type="submit" value="Login"/>
|
||||||
|
</form>
|
||||||
|
{% endblock%}
|
22
templates/main.html
Normal file
22
templates/main.html
Normal file
|
@ -0,0 +1,22 @@
|
||||||
|
{% extends "base.html" %}
|
||||||
|
|
||||||
|
{% block main %}
|
||||||
|
{{ user|default("") }}
|
||||||
|
<ul>
|
||||||
|
{% if user is defined %}
|
||||||
|
{% for l in list %}
|
||||||
|
<li/> <b>{{ l.name }}</b>
|
||||||
|
{% if user.is_member %}
|
||||||
|
{{ l.price_member }}
|
||||||
|
{% else %}
|
||||||
|
{{ l.price_non_member }}
|
||||||
|
{% endif %}
|
||||||
|
{% endfor %}
|
||||||
|
{% else %}
|
||||||
|
{% for l in list %}
|
||||||
|
<li/> <b><a href="/touchkey?username={{ l.name }}">{{ l.name }}</a></b>
|
||||||
|
{% endfor %}
|
||||||
|
<li/> <a href="/login">Password login</a>
|
||||||
|
{% endif %}
|
||||||
|
</ul>
|
||||||
|
{% endblock %}
|
9
templates/touchkey.html
Normal file
9
templates/touchkey.html
Normal file
|
@ -0,0 +1,9 @@
|
||||||
|
{% extends "base.html" %}
|
||||||
|
|
||||||
|
{% block main %}
|
||||||
|
<form action="/touchkey" method="post">
|
||||||
|
<input type="hidden" name="username" value="{{ username }}"/><br/>
|
||||||
|
Touchkey: <input type="password" name="touchkey" /><br/>
|
||||||
|
<input type="submit" value="Login"/>
|
||||||
|
</form>
|
||||||
|
{% endblock %}
|
Loading…
Reference in a new issue