commit c0a11ab37b52e3bf0a8f1e5981500e09abb3c232
parent 21dae15c5c12dd030feab58df81a228a646272a6
Author: Silas Brack <silasbrack@gmail.com>
Date: Wed, 15 Oct 2025 17:12:26 +0200
test: random fixes
Diffstat:
4 files changed, 173 insertions(+), 92 deletions(-)
diff --git a/logging.local.json b/logging.local.json
@@ -7,21 +7,9 @@
}
},
"handlers": {
- "root": {
- "class": "logging.StreamHandler",
- "level": "WARN",
- "formatter": "default",
- "stream": "ext://sys.stdout"
- },
- "simple_web_app": {
- "class": "logging.StreamHandler",
- "level": "INFO",
- "formatter": "default",
- "stream": "ext://sys.stdout"
- },
- "uvicorn": {
+ "default": {
"class": "logging.StreamHandler",
- "level": "INFO",
+ "level": "DEBUG",
"formatter": "default",
"stream": "ext://sys.stdout"
}
@@ -29,15 +17,19 @@
"loggers": {
"simple_web_app": {
"level": "INFO",
- "handlers": ["simple_web_app"]
+ "handlers": ["default"]
},
"uvicorn": {
"level": "INFO",
- "handlers": ["uvicorn"]
+ "handlers": ["default"]
+ },
+ "aiosqlite": {
+ "level": "DEBUG",
+ "handlers": ["default"]
}
},
"root": {
"level": "WARN",
- "handlers": ["root"]
+ "handlers": ["default"]
}
}
diff --git a/src/simple_web_app/app.py b/src/simple_web_app/app.py
@@ -2,10 +2,6 @@ import asyncio
import contextlib
import importlib.resources
import logging
-import time
-import os
-import random
-import sqlite3
import datetime
from pathlib import Path
@@ -15,13 +11,17 @@ from starlette.applications import Starlette
from starlette.config import Config
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
+
# from starlette.middleware.sessions import SessionMiddleware
from starlette.requests import Request
from starlette.routing import Mount, Route
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
-from simple_web_app.migration import apply_migrations, create_migrations_table_if_not_exists
+from simple_web_app.migration import (
+ apply_migrations,
+ create_migrations_table_if_not_exists,
+)
logger = logging.getLogger(__name__)
@@ -93,7 +93,9 @@ def format_timedelta(tdelta, fmt):
async def show_home_page(request: Request):
page = request.query_params.get("page", default=0)
page = int(page) if page is not None else page
- current_time = request.query_params.get("current_time", default=datetime.datetime.now(tz=datetime.UTC))
+ current_time = request.query_params.get(
+ "current_time", default=datetime.datetime.now(tz=datetime.UTC)
+ )
category_id = request.query_params.get("category_id")
category_id = int(category_id) if category_id is not None else None
@@ -103,23 +105,56 @@ async def show_home_page(request: Request):
all_categories = await queries_basic.get_categories(request.state.conn, limit=20)
rows = (
- await queries_basic.get_news_by_category(request.state.conn, category_id=category_id, limit=limit + 1, offset=offset, max_published_time=current_time)
+ await queries_basic.get_news_by_category(
+ request.state.conn,
+ category_id=category_id,
+ limit=limit + 1,
+ offset=offset,
+ max_published_time=current_time,
+ )
if category_id
- else
- await queries_basic.get_news(request.state.conn, limit=limit + 1, offset=offset, max_published_time=current_time)
+ else await queries_basic.get_news(
+ request.state.conn,
+ limit=limit + 1,
+ offset=offset,
+ max_published_time=current_time,
+ )
)
reached_end = len(rows) <= limit
rows = rows[:limit]
- categories = await asyncio.gather(*[queries_basic.get_categories_for_news(request.state.conn, news_item_id=row["id"]) for row in rows])
+ categories = await asyncio.gather(
+ *[
+ queries_basic.get_categories_for_news(
+ request.state.conn, news_item_id=row["id"]
+ )
+ for row in rows
+ ]
+ )
current_time_utc = datetime.datetime.now(tz=datetime.UTC)
+
def _get_time_since_published(published: str, current_time: datetime):
published = datetime.datetime.strptime(published, "%Y-%m-%dT%H:%M:%S.%f%z")
- return format_timedelta(current_time - published, "{days} days, {hours} hours ago")
- times_since_published = [_get_time_since_published(row["published"], current_time_utc) for row in rows]
+ return format_timedelta(
+ current_time - published, "{days} days, {hours} hours ago"
+ )
- news = [{**row, "categories": c, "time_since_published": tsp} for row, c, tsp in zip(rows, categories, times_since_published, strict=True)]
- context = {"news": news, "categories": all_categories, "page": page, "current_time": current_time, "category_id": category_id, "reached_end": reached_end}
+ times_since_published = [
+ _get_time_since_published(row["published"], current_time_utc) for row in rows
+ ]
+
+ news = [
+ {**row, "categories": c, "time_since_published": tsp}
+ for row, c, tsp in zip(rows, categories, times_since_published, strict=True)
+ ]
+ context = {
+ "news": news,
+ "categories": all_categories,
+ "page": page,
+ "current_time": current_time,
+ "category_id": category_id,
+ "reached_end": reached_end,
+ }
if is_htmx_request(request):
context = context | {"oob": True}
@@ -134,7 +169,11 @@ async def open_search(request: Request):
async def search(request: Request):
async with request.form() as form:
query = form["search"]
- rows = await queries_basic.search_news(request.state.conn, query=query, limit=10) if query else []
+ rows = (
+ await queries_basic.search_news(request.state.conn, query=query, limit=10)
+ if query
+ else []
+ )
return render(request, "search_results.html", context={"news": rows})
@@ -145,14 +184,14 @@ async def open_settings(request: Request):
async def open_settings_tab(request: Request):
tab = request.query_params.get("tab", default="general")
- return render(request, "settings_tab.html" , context={"tab": tab})
+ return render(request, "settings_tab.html", context={"tab": tab})
class CacheControlMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response = await call_next(request)
if request.headers.get("HX-Preloaded") == "true":
- response.headers['cache-control'] = 'private, max-age=60'
+ response.headers["cache-control"] = "private, max-age=60"
return response
@@ -179,7 +218,12 @@ routes = [
Route("/search", methods=["GET"], endpoint=open_search, name="search_page"),
Route("/search", methods=["POST"], endpoint=search, name="search"),
Route("/settings", methods=["GET"], endpoint=open_settings, name="settings_page"),
- Route("/settings/tab", methods=["GET"], endpoint=open_settings_tab, name="settings_tab"),
+ Route(
+ "/settings/tab",
+ methods=["GET"],
+ endpoint=open_settings_tab,
+ name="settings_tab",
+ ),
Mount("/static", StaticFiles(directory=STATIC_DIR), name="static"),
]
middleware = [
diff --git a/tests/test_simple_web_app/conftest.py b/tests/test_simple_web_app/conftest.py
@@ -1,6 +1,5 @@
import os
import unittest.mock
-from collections.abc import AsyncGenerator
import pytest
@@ -41,7 +40,9 @@ class AsyncWebSocketTestSession:
task_group = await stack.enter_async_context(anyio.create_task_group())
self.done = anyio.Event()
- async def run(*, task_status: anyio.abc.TaskStatus[anyio.CancelScope]) -> None:
+ async def run(
+ *, task_status: anyio.abc.TaskStatus[anyio.CancelScope]
+ ) -> None:
await self._run(task_status=task_status)
self.done.set()
@@ -60,10 +61,16 @@ class AsyncWebSocketTestSession:
async def __aexit__(self, *args: typing.Any) -> bool | None:
return await self.exit_stack.__aexit__(*args)
- async def _run(self, *, task_status: anyio.abc.TaskStatus[anyio.CancelScope]) -> None:
- send: anyio.create_memory_object_stream[Message] = anyio.create_memory_object_stream(math.inf)
+ async def _run(
+ self, *, task_status: anyio.abc.TaskStatus[anyio.CancelScope]
+ ) -> None:
+ send: anyio.create_memory_object_stream[Message] = (
+ anyio.create_memory_object_stream(math.inf)
+ )
send_tx, send_rx = send
- receive: anyio.create_memory_object_stream[Message] = anyio.create_memory_object_stream(math.inf)
+ receive: anyio.create_memory_object_stream[Message] = (
+ anyio.create_memory_object_stream(math.inf)
+ )
receive_tx, receive_rx = receive
with send_tx, send_rx, receive_tx, receive_rx, anyio.CancelScope() as cs:
self._receive_tx = receive_tx
@@ -76,7 +83,9 @@ class AsyncWebSocketTestSession:
async def _raise_on_close(self, message: Message) -> None:
if message["type"] == "websocket.close":
- raise WebSocketDisconnect(code=message.get("code", 1000), reason=message.get("reason", ""))
+ raise WebSocketDisconnect(
+ code=message.get("code", 1000), reason=message.get("reason", "")
+ )
elif message["type"] == "websocket.http.response.start":
status_code: int = message["status"]
headers: list[tuple[bytes, bytes]] = message["headers"]
@@ -87,7 +96,9 @@ class AsyncWebSocketTestSession:
body.append(message["body"])
if not message.get("more_body", False):
break
- raise WebSocketDenialResponse(status_code=status_code, headers=headers, content=b"".join(body))
+ raise WebSocketDenialResponse(
+ status_code=status_code, headers=headers, content=b"".join(body)
+ )
async def send(self, message: Message) -> None:
await self._receive_tx.send(message)
@@ -98,15 +109,21 @@ class AsyncWebSocketTestSession:
async def send_bytes(self, data: bytes) -> None:
await self.send({"type": "websocket.receive", "bytes": data})
- async def send_json(self, data: typing.Any, mode: typing.Literal["text", "binary"] = "text") -> None:
+ async def send_json(
+ self, data: typing.Any, mode: typing.Literal["text", "binary"] = "text"
+ ) -> None:
text = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
if mode == "text":
await self.send({"type": "websocket.receive", "text": text})
else:
- await self.send({"type": "websocket.receive", "bytes": text.encode("utf-8")})
+ await self.send(
+ {"type": "websocket.receive", "bytes": text.encode("utf-8")}
+ )
async def aclose(self, code: int = 1000, reason: str | None = None) -> None:
- await self.send({"type": "websocket.disconnect", "code": code, "reason": reason})
+ await self.send(
+ {"type": "websocket.disconnect", "code": code, "reason": reason}
+ )
async def receive(self) -> Message:
return await self._send_rx.receive()
@@ -121,7 +138,9 @@ class AsyncWebSocketTestSession:
await self._raise_on_close(message)
return typing.cast(bytes, message["bytes"])
- async def receive_json(self, mode: typing.Literal["text", "binary"] = "text") -> typing.Any:
+ async def receive_json(
+ self, mode: typing.Literal["text", "binary"] = "text"
+ ) -> typing.Any:
message = await self.receive()
await self._raise_on_close(message)
if mode == "text":
@@ -172,7 +191,10 @@ class _AsyncTestClientTransport(httpx.AsyncBaseTransport):
headers = [(b"host", (f"{host}:{port}").encode())]
# Include other request headers.
- headers += [(key.lower().encode(), value.encode()) for key, value in request.headers.multi_items()]
+ headers += [
+ (key.lower().encode(), value.encode())
+ for key, value in request.headers.multi_items()
+ ]
scope: dict[str, typing.Any]
@@ -254,13 +276,22 @@ class _AsyncTestClientTransport(httpx.AsyncBaseTransport):
nonlocal raw_kwargs, response_started, template, context
if message["type"] == "http.response.start":
- assert not response_started, 'Received multiple "http.response.start" messages.'
+ assert not response_started, (
+ 'Received multiple "http.response.start" messages.'
+ )
raw_kwargs["status_code"] = message["status"]
- raw_kwargs["headers"] = [(key.decode(), value.decode()) for key, value in message.get("headers", [])]
+ raw_kwargs["headers"] = [
+ (key.decode(), value.decode())
+ for key, value in message.get("headers", [])
+ ]
response_started = True
elif message["type"] == "http.response.body":
- assert response_started, 'Received "http.response.body" without "http.response.start".'
- assert not response_complete.is_set(), 'Received "http.response.body" after response completed.'
+ assert response_started, (
+ 'Received "http.response.body" without "http.response.start".'
+ )
+ assert not response_complete.is_set(), (
+ 'Received "http.response.body" after response completed.'
+ )
body = message.get("body", b"")
more_body = message.get("more_body", False)
if request.method != "HEAD":
@@ -353,9 +384,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
if timeout is not httpx.USE_CLIENT_DEFAULT:
@@ -388,9 +422,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().get(
@@ -411,9 +448,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().options(
@@ -434,9 +474,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().head(
@@ -461,9 +504,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().post(
@@ -492,9 +538,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().put(
@@ -523,9 +572,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().patch(
@@ -550,9 +602,12 @@ class AsyncTestClient(httpx.AsyncClient):
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
- auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- follow_redirects: bool | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
- timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ auth: httpx._types.AuthTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ follow_redirects: bool
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
+ timeout: httpx._types.TimeoutTypes
+ | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT,
extensions: dict[str, typing.Any] | None = None,
) -> httpx.Response:
return await super().delete(
@@ -592,12 +647,12 @@ class AsyncTestClient(httpx.AsyncClient):
async def __aenter__(self) -> "AsyncTestClient":
async with contextlib.AsyncExitStack() as stack:
task_group = await stack.enter_async_context(anyio.create_task_group())
- send: anyio.create_memory_object_stream[typing.MutableMapping[str, typing.Any] | None] = (
- anyio.create_memory_object_stream(math.inf)
- )
- receive: anyio.create_memory_object_stream[typing.MutableMapping[str, typing.Any]] = (
- anyio.create_memory_object_stream(math.inf)
- )
+ send: anyio.create_memory_object_stream[
+ typing.MutableMapping[str, typing.Any] | None
+ ] = anyio.create_memory_object_stream(math.inf)
+ receive: anyio.create_memory_object_stream[
+ typing.MutableMapping[str, typing.Any]
+ ] = anyio.create_memory_object_stream(math.inf)
for channel in (*send, *receive):
stack.push_async_callback(channel.aclose)
self.stream_send = StapledObjectStream(*send)
@@ -670,5 +725,5 @@ class AsyncTestClient(httpx.AsyncClient):
async def async_test_client(test_env):
with unittest.mock.patch.dict(os.environ, test_env, clear=True):
from simple_web_app.app import app
- return AsyncTestClient(app)
+ return AsyncTestClient(app)
diff --git a/tests/test_simple_web_app/test_app.py b/tests/test_simple_web_app/test_app.py
@@ -1,14 +1,9 @@
-import os
-import unittest
import datetime
import collections
-from pprint import pprint
import aiosqlite
import pytest
from bs4 import BeautifulSoup
-from httpx import AsyncClient
-from starlette.testclient import TestClient
DummyData = collections.namedtuple("DummyData", ["table", "cols", "rows"])
@@ -173,8 +168,3 @@ async def test_settings(async_test_client, test_data):
assert response.template.name == "settings_tab.html"
soup = BeautifulSoup(response.text, features="html.parser")
assert soup.find("h4").text == "Sync Preferences"
-
-
-async def test_preload():
- pass
-
