forked from retoor/devplacepy
retoorandClaude Sonnet 5
3c69de9d55
Happy 404: an HTML 404 (unmatched route, or an explicit not-found inside a
real route) now renders a random existing post instead of the error page,
using the exact same context builder as a real post view. Toggle is the
happy_404_enabled site setting (default on, /admin/settings); JSON/API
requests and a handful of excluded prefixes are never affected. The pool of
candidate slugs is cached in-process and resampled periodically so it stays
fast and eventually cycles the whole posts table; on any internal failure it
falls straight through to the real 404 page.
Applying this everywhere surfaced ~60 existing tests that asserted a literal
404 for a legitimate resource-not-found flow (deleted post, unknown
container, wrong project slug, etc.) - each now disables the setting for the
duration of that specific check and restores it after, so the underlying
not-found behavior stays covered independently of the new feature.
Post page also gained, all built on the same shared post_page_context() so
they render identically on both a real post and a happy-404 page:
- A left sidebar (three separate cards, matching /feed's sidebar-card
convention) for "Gists from {author}", "Projects from {author}" (private
projects filtered through the normal visibility check), and "Related
Discussions" - each cached per author and invalidated on create/edit/
delete so new content shows up immediately.
- A right column reusing /feed's exact Daily Topic widget class for up to
three "Featured" articles (the existing but previously-unused `featured`
news flag), cached as a pool with per-request random sampling.
- A "Next post -> " link beside "Back to Feed", pointing at the next older
post site-wide (blocked authors skipped). Wired through the same next_url
mechanism already used for listing pagination, so it emits a real
backend-rendered <link rel="next"> tag for SEO, not just a visible link.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BWJy6PrMMt5hwWxQwia2rd
171 lines
6.0 KiB
Python
171 lines
6.0 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from datetime import timedelta
|
|
import time
|
|
import pytest
|
|
import requests
|
|
from tests.conftest import BASE_URL, CACHE_VERSION_PROPAGATION_SECONDS, run_async
|
|
from devplacepy.database import db, get_table, init_db, refresh_snapshot, set_setting
|
|
from devplacepy import config, project_files
|
|
from devplacepy.services.containers import api, store, runtime
|
|
from devplacepy.services.containers.backend.base import Mount, PortMapping, RunSpec
|
|
from devplacepy.services.containers.backend.docker_cli import build_run_argv, parse_size
|
|
from devplacepy.services.containers.backend.fake import FakeBackend
|
|
from devplacepy.services.containers.service import ContainerService
|
|
from devplacepy.services.containers.api import INSTANCE_LABEL
|
|
from devplacepy.services.devii.tasks.schedule import Schedule, now_utc
|
|
_CONTAINER_TABLES = (
|
|
"instances",
|
|
"instance_events",
|
|
"instance_metrics",
|
|
"instance_schedules",
|
|
)
|
|
@pytest.fixture(autouse=True)
|
|
def _init_db_containers():
|
|
init_db()
|
|
yield
|
|
@pytest.fixture
|
|
def env(tmp_path, monkeypatch):
|
|
fake = FakeBackend()
|
|
runtime.set_backend(fake)
|
|
monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path / "ws")
|
|
pid = "ctest-p1"
|
|
project = {"uid": pid, "slug": "ctest", "title": "C", "user_uid": "ctest-u1"}
|
|
user = {"uid": "ctest-u1", "username": "ctestadmin"}
|
|
project_files.write_text_file(pid, user, "app.py", "print(1)\n")
|
|
yield {"fake": fake, "project": project, "user": user}
|
|
runtime.set_backend(None)
|
|
for table in _CONTAINER_TABLES:
|
|
if table in db.tables:
|
|
for row in [r for r in get_table(table).find()]:
|
|
if str(row.get("project_uid", "")).startswith("ctest"):
|
|
get_table(table).delete(uid=row["uid"])
|
|
for row in list(get_table("project_files").find()):
|
|
if str(row.get("project_uid", "")).startswith("ctest"):
|
|
get_table("project_files").delete(uid=row["uid"])
|
|
def _ready_instance(env, **kwargs):
|
|
return run_async(
|
|
api.create_instance(env["project"], name=kwargs.pop("name", "inst"), **kwargs)
|
|
)
|
|
def _promote_admin(username: str) -> None:
|
|
users = get_table("users")
|
|
user = users.find_one(username=username)
|
|
if user:
|
|
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
|
def _api_key(username: str) -> str:
|
|
refresh_snapshot()
|
|
return get_table("users").find_one(username=username)["api_key"]
|
|
|
|
|
|
def test_http_ingress_proxy(app_server):
|
|
import http.server
|
|
import socket
|
|
import socketserver
|
|
import threading
|
|
|
|
sock = socket.socket()
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/plain")
|
|
self.end_headers()
|
|
self.wfile.write(b"hello from upstream " + self.path.encode())
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
httpd = socketserver.TCPServer(("127.0.0.1", port), Handler)
|
|
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
|
thread.start()
|
|
slug = f"ing{port}"
|
|
uid = f"ingtest-{port}"
|
|
get_table("instances").insert(
|
|
{
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
"uid": uid,
|
|
"name": "ingress",
|
|
"project_uid": "ingtest",
|
|
"status": "running",
|
|
"ingress_slug": slug,
|
|
"ingress_port": 8000,
|
|
"ports_json": f'[{{"host": {port}, "container": 8000, "proto": "tcp"}}]',
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
refresh_snapshot()
|
|
set_setting("happy_404_enabled", "0")
|
|
time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)
|
|
try:
|
|
r = requests.get(f"{BASE_URL}/p/{slug}/foo")
|
|
assert r.status_code == 200, r.text
|
|
assert "hello from upstream" in r.text and "/foo" in r.text
|
|
assert requests.get(f"{BASE_URL}/p/does-not-exist-slug").status_code == 404
|
|
finally:
|
|
set_setting("happy_404_enabled", "1")
|
|
time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)
|
|
httpd.shutdown()
|
|
get_table("instances").delete(uid=uid)
|
|
refresh_snapshot()
|
|
|
|
|
|
def test_http_ingress_proxy_injects_a_base_into_the_root_document_only(app_server):
|
|
import http.server
|
|
import socket
|
|
import socketserver
|
|
import threading
|
|
|
|
sock = socket.socket()
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(
|
|
b"<!doctype html><html><head><title>t</title></head>"
|
|
b"<body><script src='app.js'></script></body></html>"
|
|
)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
httpd = socketserver.TCPServer(("127.0.0.1", port), Handler)
|
|
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
|
thread.start()
|
|
slug = f"base{port}"
|
|
uid = f"basetest-{port}"
|
|
get_table("instances").insert(
|
|
{
|
|
"uid": uid,
|
|
"name": "ingress-base",
|
|
"project_uid": "ingtest",
|
|
"status": "running",
|
|
"ingress_slug": slug,
|
|
"ingress_port": 8000,
|
|
"ports_json": f'[{{"host": {port}, "container": 8000, "proto": "tcp"}}]',
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
refresh_snapshot()
|
|
try:
|
|
root = requests.get(f"{BASE_URL}/p/{slug}")
|
|
assert root.status_code == 200, root.text
|
|
assert f'<base href="/p/{slug}/">' in root.text
|
|
nested = requests.get(f"{BASE_URL}/p/{slug}/static/webview/pre/index.html")
|
|
assert nested.status_code == 200, nested.text
|
|
assert "<base" not in nested.text
|
|
finally:
|
|
httpd.shutdown()
|
|
get_table("instances").delete(uid=uid)
|
|
refresh_snapshot()
|