FastHTML tests

import time

from IPython import display
from enum import Enum
from pprint import pprint

from fastcore.test import *
from starlette.testclient import TestClient
from starlette.requests import Headers

---

source

is_typeddict

 is_typeddict (cls:type)

Check if cls is a TypedDict

class MyDict(TypedDict): name:str

assert is_typeddict(MyDict)
assert not is_typeddict({'a':1})

---

source

is_namedtuple

 is_namedtuple (cls)

True is cls is a namedtuple type

assert is_namedtuple(namedtuple('tst', ['a']))
assert not is_namedtuple(tuple)

---

source

date

 date (s:str)

Convert s to a datetime

date('2pm')

datetime.datetime(2024, 6, 22, 14, 0)

---

source

snake2hyphens

 snake2hyphens (s:str)

Convert s from snake case to hyphenated and capitalised

snake2hyphens("snake_case")

'Snake-Case'

---

source

HtmxHeaders

 HtmxHeaders (boosted:str|None=None, current_url:str|None=None,
              history_restore_request:str|None=None, prompt:str|None=None,
              request:str|None=None, target:str|None=None,
              trigger_name:str|None=None, trigger:str|None=None)

def test_request(url: str='/', headers: dict={}, method: str='get') -> Request:
    scope = {
        'type': 'http',
        'method': method,
        'path': url,
        'headers': Headers(headers).raw,
        'query_string': b'',
        'scheme': 'http',
        'client': ('127.0.0.1', 8000),
        'server': ('127.0.0.1', 8000),
    }
    receive = lambda: {"body": b"", "more_body": False}
    return Request(scope, receive)

h = test_request(headers=Headers({'HX-Request':'1'}))
_get_htmx(h)

HtmxHeaders(boosted=None, current_url=None, history_restore_request=None, prompt=None, request='1', target=None, trigger_name=None, trigger=None)

---

source

str2int

 str2int (s)

Convert s to an int

str2int('1'),str2int('none')

(1, 0)

def _mk_list(t, v): return [t(o) for o in v]

_fix_anno(Union[str,None]),_fix_anno(float)

(str, float)

_fix_anno(int)('1')

1

_fix_anno(list[int])(['1','2'])

[1, 2]

_fix_anno(list[int])('1')

[1]

d = dict(k=int, l=List[int])
_form_arg('k', "1", d)

1

_form_arg('l', "1", d)

[1]

_form_arg('l', ["1","2"], d)

[1, 2]

---

source

HttpHeader

 HttpHeader (k:str, v:str)

---

source

form2dict

 form2dict (form:starlette.datastructures.FormData)

Convert starlette form data to a dict

d = [('a',1),('a',2),('b',0)]
fd = FormData(d)
res = form2dict(fd)
test_eq(res['a'], [1,2])
test_eq(res['b'], 0)

async def f(req):
    def _f(p:HttpHeader): ...
    p = first(signature(_f).parameters.values())
    result = await _from_body(req, p)
    return JSONResponse(result.__dict__)

app = Starlette(routes=[Route('/', f, methods=['POST'])])
client = TestClient(app)

d = dict(k='value1',v=['value2','value3'])
response = client.post('/', data=d)
print(response.json())

{'k': 'value1', 'v': "['value2', 'value3']"}

def g(req, this:Starlette, a:str, b:HttpHeader): ...

async def f(req):
    a = await _wrap_req(req, signature(g).parameters)
    return Response(str(a))

app = Starlette(routes=[Route('/', f, methods=['POST'])])
client = TestClient(app)

response = client.post('/?a=1', data=d)
print(response.text)

[<starlette.requests.Request object>, <starlette.applications.Starlette object>, '1', HttpHeader(k='value1', v="['value2', 'value3']")]

---

source

flat_xt

 flat_xt (lst)

Flatten lists, except for XTs

x = XT('a',1)
flat_xt([x, x, [x,x]])

[['a', 1, {}], ['a', 1, {}], ['a', 1, {}], ['a', 1, {}]]

---

source

Beforeware

 Beforeware (f, skip=None)

Initialize self. See help(type(self)) for accurate signature.

---

source

WS_RouteX

 WS_RouteX (path:str, endpoint, name=None, middleware=None, hdrs=None,
            before=None, **bodykw)

Initialize self. See help(type(self)) for accurate signature.

---

source

RouteX

 RouteX (path:str, endpoint, methods=None, name=None,
         include_in_schema=True, middleware=None, hdrs=None, before=None,
         **bodykw)

Initialize self. See help(type(self)) for accurate signature.

---

source

RouterX

 RouterX (routes=None, redirect_slashes=True, default=None,
          on_startup=None, on_shutdown=None, lifespan=None,
          middleware=None, hdrs=None, before=None, **bodykw)

Initialize self. See help(type(self)) for accurate signature.

---

source

get_key

 get_key (key=None, fname='.sesskey')

get_key()

'08b63b51-be3a-4f54-8d26-4cd27eb17c0d'

def _list(o): return [] if not o else o if isinstance(o, (tuple,list)) else [o]

---

source

FastHTML

 FastHTML (debug=False, routes=None, middleware=None,
           exception_handlers=None, on_startup=None, on_shutdown=None,
           lifespan=None, hdrs=None, before=None, default_hdrs=True,
           secret_key=None, session_cookie='session_', max_age=31536000,
           sess_path='/', same_site='lax', sess_https_only=False,
           sess_domain=None, key_fname='.sesskey', **bodykw)

*Creates an application instance.

Parameters:

• debug - Boolean indicating if debug tracebacks should be returned on errors.
• routes - A list of routes to serve incoming HTTP and WebSocket requests.
• middleware - A list of middleware to run for every request. A starlette application will always automatically include two middleware classes. ServerErrorMiddleware is added as the very outermost middleware, to handle any uncaught errors occurring anywhere in the entire stack. ExceptionMiddleware is added as the very innermost middleware, to deal with handled exception cases occurring in the routing or endpoints.
• exception_handlers - A mapping of either integer status codes, or exception class types onto callables which handle the exceptions. Exception handler callables should be of the form handler(request, exc) -> response and may be either standard functions, or async functions.
• on_startup - A list of callables to run on application startup. Startup handler callables do not take any arguments, and may be either standard functions, or async functions.
• on_shutdown - A list of callables to run on application shutdown. Shutdown handler callables do not take any arguments, and may be either standard functions, or async functions.
• lifespan - A lifespan context function, which can be used to perform startup and shutdown tasks. This is a newer style that replaces the on_startup and on_shutdown handlers. Use one or the other, not both.*

---

source

reg_re_param

 reg_re_param (m, s)

---

source

MiddlewareBase

 MiddlewareBase ()

Initialize self. See help(type(self)) for accurate signature.

def get_cli(app): return app,TestClient(app),app.route

app,cli,rt = get_cli(FastHTML(secret_key='soopersecret'))

@rt("/hi")
def get(): return 'Hi there'

r = cli.get('/hi')
r.text

'Hi there'

@rt("/hi")
def post(): return 'Postal'

cli.post('/hi').text

'Postal'

@app.get("/")
def show_host(req): return req.headers['host']

cli.get('/').text

'testserver'

@rt('/user/{nm}', name='gday')
def get(nm:str): return f"Good day to you, {nm}!"

cli.get('/user/Alexis').text

'Good day to you, Alexis!'

test_eq(app.router.url_path_for('gday', nm='Jeremy'), '/user/Jeremy')

hxhdr = {'headers':{'hx-request':"1"}}

@rt('/xt')
def get(): return Title('Foo'),H1('bar')

txt = cli.get('/xt').text
assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt

@rt('/xt2')
def get(): return H1('bar')

txt = cli.get('/xt2').text
assert '<title>FastHTML page</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt

assert cli.get('/xt2', **hxhdr).text.strip() == '<h1>bar</h1>'

@rt('/xt3')
def get(): return Html(Head(Title('hi')), Body(P('there')))

txt = cli.get('/xt3').text
assert '<title>FastHTML page</title>' not in txt and '<title>hi</title>' in txt and '<p>there</p>' in txt

def test_r(cli, path, exp, meth='get', hx=False, **kwargs):
    if hx: kwargs['headers'] = {'hx-request':"1"}
    test_eq(getattr(cli, meth)(path, **kwargs).text, exp)

app.chk = 'foo'
ModelName = str_enum('ModelName', "alexnet", "resnet", "lenet")
fake_db = [{"name": "Foo"}, {"name": "Bar"}]

@rt('/html/{idx}')
async def get(idx:int): return Body(H4(f'Next is {idx+1}.'))

reg_re_param("imgext", "ico|gif|jpg|jpeg|webm")

@rt(r'/static/{path:path}{fn}.{ext:imgext}')
def get(fn:str, path:str, ext:str): return f"Getting {fn}.{ext} from /{path}"

@rt("/models/{nm}")
def get(nm:ModelName): return nm

@rt("/files/{path}")
async def get(path: Path): return path.with_suffix('.txt')

@rt("/items/")
def get(idx:int|None = 0): return fake_db[idx]

test_r(cli, '/html/1', '<body>\n  <h4>Next is 2.</h4>\n</body>\n', hx=True)
test_r(cli, '/static/foo/jph.ico', 'Getting jph.ico from /foo/')
test_r(cli, '/models/alexnet', 'alexnet')
test_r(cli, '/files/foo', 'foo.txt')
test_r(cli, '/items/?idx=1', '{"name":"Bar"}')
test_r(cli, '/items/', '{"name":"Foo"}')

@app.get("/booly/")
def _(coming:bool=True): return 'Coming' if coming else 'Not coming'

@app.get("/datie/")
def _(d:date): return d

@app.get("/ua")
async def _(user_agent:str): return user_agent

@app.get("/hxtest")
def _(htmx): return htmx.request

@app.get("/hxtest2")
def _(foo:HtmxHeaders, req): return foo.request

@app.get("/app")
def _(app): return app.chk

@app.get("/app2")
def _(foo:FastHTML): return foo.chk,HttpHeader("mykey", "myval")

test_r(cli, '/booly/?coming=true', 'Coming')
test_r(cli, '/booly/?coming=no', 'Not coming')
date_str = "17th of May, 2024, 2p"
test_r(cli, f'/datie/?d={date_str}', '2024-05-17 14:00:00')
test_r(cli, '/ua', 'FastHTML', headers={'User-Agent':'FastHTML'})
test_r(cli, '/hxtest' , '1', headers={'HX-Request':'1'})
test_r(cli, '/hxtest2', '1', headers={'HX-Request':'1'})
test_r(cli, '/app' , 'foo')

r = cli.get('/app2', **hxhdr)
test_eq(r.text, 'foo\n')
test_eq(r.headers['mykey'], 'myval')

@dataclass
class Bodie: a:int;b:str

@rt("/bodie/{nm}/")
async def post(nm:str, data:Bodie):
    res = asdict(data)
    res['nm'] = nm
    return res

@app.post("/bodied/")
async def bodied(nm:str, data:dict): return data

nt = namedtuple('Bodient', ['a','b'])

@app.post("/bodient/")
async def bodient(data:nt): return data._asdict()

class BodieTD(TypedDict): a:int;b:str='foo'

@app.post("/bodietd/")
async def bodient(data:BodieTD): return data

class Bodie2:
    a:int|None; b:str
    def __init__(self, a, b='foo'): store_attr()

@app.post("/bodie2/")
async def bodie(d:Bodie2): return f"a: {d.a}; b: {d.b}"

d = dict(a=1, b='foo')

test_r(cli, '/bodie/me', '{"a":1,"b":"foo","nm":"me"}', 'post', data=d)
test_r(cli, '/bodied/', '{"a":"1","b":"foo"}', 'post', data=d)
test_r(cli, '/bodie2/', 'a: 1; b: foo', 'post', data={'a':1})
test_r(cli, '/bodient/', '{"a":"1","b":"foo"}', 'post', data=d)
test_r(cli, '/bodietd/', '{"a":1,"b":"foo"}', 'post', data=d)

@rt("/setcookie")
async def get(req):
    now = datetime.now()
    res = Response(f'Set to {now}')
    res.set_cookie('now', str(now))
    return res

@rt("/getcookie")
async def get(now:date): return f'Cookie was set at time {now.time()}'

print(cli.get('/setcookie').text)
time.sleep(0.01)
cli.get('/getcookie').text

Set to 2024-06-15 09:33:00.521810

'Cookie was set at time 09:33:00.521810'

@rt("/setsess")
async def get(sess):
    now = datetime.now()
    sess['noo'] = str(now)
    return f'Set to {now}'

@rt("/getsess")
async def get(noo:date): return f'Session time: {noo.time()}'

print(cli.get('/setsess').text)
time.sleep(0.01)
cli.get('/getsess').text

Set to 2024-06-15 09:33:01.830672

'Session time: 09:33:01.830672'

@rt("/upload")
async def post(uploadfile:str): return (await uploadfile.read()).decode()

fn = '../CHANGELOG.md'
data = {'message': 'Hello, world!'}
with open(fn, 'rb') as f:
    print(cli.post('/upload', files={'uploadfile': f}, data=data).text[:80])

# Release notes

<!-- do not remove -->

## 0.0.13

### New Features

- Add `htm

auth = user_pwd_auth(testuser='spycraft')
app,cli,rt = get_cli(FastHTML(middleware=[auth]))

@rt("/locked")
def get(auth): return 'Hello, ' + auth

test_eq(cli.get('/locked').text, 'not authenticated')
test_eq(cli.get('/locked', auth=("testuser","spycraft")).text, 'Hello, testuser')

hdrs, routes = app.router.hdrs, app.routes

app,cli,rt = get_cli(FastHTMLWithLiveReload())

@rt("/hi")
def get(): return 'Hi there'

test_eq(cli.get('/hi').text, "Hi there")

lr_hdrs, lr_routes = app.router.hdrs, app.routes
test_eq(len(lr_hdrs), len(hdrs)+1)
assert app.LIVE_RELOAD_HEADER in lr_hdrs
test_eq(len(lr_routes), len(routes)+1)
assert app.LIVE_RELOAD_ROUTE in lr_routes