"""The public website (app/routers/public_site.py): site settings, putting a page on the site, what a public page may hold, how pages are served to visitors and search engines, HTML pages and the .zip site import.""" import io import json import uuid import zipfile import pytest from app.core.html_sanitize import SanitizeReport, sanitize_document, sanitize_fragment from app.crud import public_site as site_crud from tests.test_pages import _create_role_and_token FAKE_INDEX = 'FlowFabric
' @pytest.fixture(autouse=True) def _fake_app_index(monkeypatch): async def fake() -> str: return FAKE_INDEX monkeypatch.setattr(site_crud, "app_index_html", fake) def _slug() -> str: return f"site-{uuid.uuid4().hex[:8]}" async def _site(client, headers, **extra) -> str: slug = _slug() resp = await client.put("/tenants/me/site", headers=headers, json={"enabled": True, "slug": slug, "site_name": "Acme Transport", **extra}) assert resp.status_code == 200, resp.text return slug async def _public_page(client, headers, name: str, blocks: list, *, seo: dict | None = None, publish: bool = True) -> dict: resp = await client.post("/pages", headers=headers, json={"name": name, "blocks": blocks, **({"seo": seo} if seo else {})}) assert resp.status_code == 201, resp.text page = resp.json() resp = await client.patch(f"/pages/{page['id']}/site", headers=headers, json={"is_public": True}) assert resp.status_code == 200, resp.text if publish: resp = await client.post(f"/pages/{page['id']}/publish", headers=headers, json={}) assert resp.status_code == 200, resp.text return resp.json() WELCOME = [ {"id": "h", "type": "page_header", "label": "Welcome to Acme", "subtitle": "Buses across Kerala", "layout": {"x": 0, "y": 0, "w": 800, "h": 80}}, {"id": "p", "type": "text", "text": "We run three hundred routes every day, from Kasaragod to Thiruvananthapuram.", "layout": {"x": 0, "y": 100, "w": 800, "h": 60}}, {"id": "img", "type": "image", "image_url": "https://cdn.example.com/bus.jpg", "alt_text": "A red bus", "layout": {"x": 0, "y": 180, "w": 400, "h": 200}}, ] # ---- sanitizing --------------------------------------------------------------------------------------------- def test_sanitize_document_removes_everything_that_runs(): report = SanitizeReport() out = sanitize_document( """

Hi

ab t """, report, ) lowered = out.lower() for bad in ("Hi there

') == '

Hi there

' # ---- site settings ------------------------------------------------------------------------------------------ async def test_site_settings_round_trip_and_rules(client, auth_headers): resp = await client.put("/tenants/me/site", headers=auth_headers, json={"enabled": True}) assert resp.status_code == 422 resp = await client.put("/tenants/me/site", headers=auth_headers, json={"enabled": True, "slug": "Not A Slug"}) assert resp.status_code == 422 slug = _slug() resp = await client.put( "/tenants/me/site", headers=auth_headers, json={ "enabled": True, "slug": slug, "domains": ["https://WWW.Acme-Example.test/", "www.acme-example.test"], "google_site_verification": '', }, ) assert resp.status_code == 200, resp.text body = resp.json() assert body["slug"] == slug and body["path"] == f"/site/{slug}/" assert body["domains"] == ["www.acme-example.test"] assert body["google_site_verification"] == "abc123XYZ" page = (await client.post("/pages", headers=auth_headers, json={"name": "Private"})).json() resp = await client.put("/tenants/me/site", headers=auth_headers, json={"enabled": True, "slug": slug, "home_page_id": page["id"]}) assert resp.status_code == 422 assert "isn't public" in resp.json()["detail"] async def test_site_slug_is_unique_across_workspaces(client, auth_headers, test_tenant): from app.models.identity import Tenant from tests.conftest import AsyncSessionLocal slug = _slug() async with AsyncSessionLocal() as db, db.begin(): other = Tenant(name="Pytest other site", status="active", site_slug=slug, site_settings={"enabled": True}) db.add(other) await db.flush() other_id = other.id try: resp = await client.put("/tenants/me/site", headers=auth_headers, json={"enabled": True, "slug": slug}) assert resp.status_code == 409 finally: async with AsyncSessionLocal() as db, db.begin(): await db.execute(Tenant.__table__.delete().where(Tenant.id == other_id)) # ---- putting pages on the site ------------------------------------------------------------------------------ async def test_page_with_data_blocks_cannot_go_public(client, auth_headers): resp = await client.post( "/pages", headers=auth_headers, json={"name": "Data", "blocks": [{"id": "l", "type": "data_list", "data_source": f"master:{uuid.uuid4()}"}]}, ) page = resp.json() resp = await client.patch(f"/pages/{page['id']}/site", headers=auth_headers, json={"is_public": True}) assert resp.status_code == 422 assert resp.json()["detail"]["code"] == "not_public_safe" # A button that starts a workflow needs a signed-in person too. resp = await client.patch( f"/pages/{page['id']}", headers=auth_headers, json={"blocks": [{"id": "b", "type": "button", "label": "Apply", "action": {"type": "trigger_workflow", "workflow_id": str(uuid.uuid4())}}]}, ) assert resp.status_code == 200 resp = await client.patch(f"/pages/{page['id']}/site", headers=auth_headers, json={"is_public": True}) assert resp.status_code == 422 assert "Apply" in resp.json()["detail"]["message"] async def test_publishing_a_public_page_rechecks_its_blocks(client, auth_headers): await _site(client, auth_headers) page = await _public_page(client, auth_headers, "About us", WELCOME) assert page["is_public"] is True and page["slug"] == "about-us" await client.patch(f"/pages/{page['id']}", headers=auth_headers, json={"blocks": [{"id": "f", "type": "form", "form_id": str(uuid.uuid4())}]}) resp = await client.post(f"/pages/{page['id']}/publish", headers=auth_headers, json={}) assert resp.status_code == 422 async def test_page_slug_clash(client, auth_headers): a = (await client.post("/pages", headers=auth_headers, json={"name": "Contact"})).json() b = (await client.post("/pages", headers=auth_headers, json={"name": "Contact"})).json() first = await client.patch(f"/pages/{a['id']}/site", headers=auth_headers, json={"is_public": True}) assert first.json()["slug"] == "contact" # Picked automatically: the next free one. second = await client.patch(f"/pages/{b['id']}/site", headers=auth_headers, json={"is_public": True}) assert second.json()["slug"] == "contact-2" # Asked for by name: refused. clash = await client.patch(f"/pages/{b['id']}/site", headers=auth_headers, json={"is_public": True, "slug": "contact"}) assert clash.status_code == 409 async def test_putting_a_page_on_the_site_needs_publish_permission(client, test_tenant): token = await _create_role_and_token(test_tenant["tenant_id"], {"pages": ["view", "create", "edit"]}) headers = {"Authorization": f"Bearer {token}"} page = (await client.post("/pages", headers=headers, json={"name": "Mine"})).json() resp = await client.patch(f"/pages/{page['id']}/site", headers=headers, json={"is_public": True}) assert resp.status_code == 403 # ---- serving blocks pages ----------------------------------------------------------------------------------- async def test_blocks_page_is_served_with_seo_and_readable_content(client, auth_headers): slug = await _site(client, auth_headers, default_og_image="https://cdn.example.com/card.png") page = await _public_page( client, auth_headers, "Routes", WELCOME, seo={"title": "Bus routes", "keywords": "bus, kerala", "json_ld": json.dumps({"@type": "Thing", "name": ""})}, ) resp = await client.get(f"/site/{slug}/{page['slug']}") assert resp.status_code == 200, resp.text body = resp.text assert "Bus routes | Acme Transport" in body # Blank description: the start of the page's own text. assert '' in body assert '' in body assert '' in body # The web app's own scripts start, inside the shell, with the page's content for crawlers. assert '' in body assert f'data-ff-site="{slug}"' in body assert "

Welcome to Acme

" in body and 'alt="A red bus"' in body # Structured data can't close its own

Fly with us

" "
" ) resp = await client.patch(f"/pages/{page['id']}", headers=auth_headers, json={"html_document": doc, "seo": {"title": "Fly", "description": "Cheap flights"}}) assert resp.status_code == 200, resp.text saved = resp.json()["html_document"] assert "Fly | Acme Transport" in text and "Old title" not in text assert '' in text and "content='old'" not in text and 'content="old"' not in text assert '

Fly with us

' in text assert "/assets/app.js" not in text # html pages are served whole, without the web app async def test_html_document_needs_the_html_permission(client, test_tenant): token = await _create_role_and_token(test_tenant["tenant_id"], {"pages": ["view", "create", "edit"]}) headers = {"Authorization": f"Bearer {token}"} page = (await client.post("/pages", headers=headers, json={"name": "Mine", "content_mode": "html"})).json() resp = await client.patch(f"/pages/{page['id']}", headers=headers, json={"html_document": "

Hi

"}) assert resp.status_code == 403 def _site_zip() -> bytes: buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr( "mysite/index.html", "Acme Home" "LogoAbout" "", ) zf.writestr("mysite/about.html", "About Acme

About

Home") zf.writestr("mysite/css/site.css", "@import 'base.css'; .hero{background:url('../img/logo.png') no-repeat}") zf.writestr("mysite/css/base.css", "body{margin:0}") zf.writestr("mysite/img/logo.png", b"\x89PNG\r\n\x1a\nfake") zf.writestr("mysite/js/app.js", "alert(1)") zf.writestr("__MACOSX/._index.html", "junk") return buf.getvalue() async def test_zip_site_import_makes_exact_pages_with_assets(client, auth_headers, test_tenant): resp = await client.post("/pages/import-html", headers=auth_headers, files={"file": ("mysite.zip", _site_zip(), "application/zip")}) assert resp.status_code == 201, resp.text result = resp.json() assert [p["name"] for p in result["pages"]] == ["Acme Home", "About Acme"] assert result["assets"] == 1 assert any("JavaScript" in n for n in result["notes"]) assert any("missing.png" in n for n in result["notes"]) home_id, about_id = result["pages"][0]["id"], result["pages"][1]["id"] home = (await client.get(f"/pages/{home_id}", headers=auth_headers)).json() assert home["content_mode"] == "html" and home["slug"] == "home" doc = home["html_document"] assert "Pasted", "page_id": page["id"]}) assert resp.status_code == 201, resp.text assert resp.json()["pages"] == [{"id": page["id"], "name": "Keep my name", "slug": None}] after = (await client.get(f"/pages/{page['id']}", headers=auth_headers)).json() assert after["content_mode"] == "html" and "

Pasted

" in after["html_document"] async def test_public_assets(client, auth_headers, test_tenant): resp = await client.post("/pages/assets", headers=auth_headers, files={"file": ("hero.png", b"\x89PNG\r\n\x1a\n", "image/png")}) assert resp.status_code == 201, resp.text asset = resp.json() assert asset["url"].startswith(f"/site/_files/{test_tenant['tenant_id']}/{asset['id']}/") assert (await client.get(asset["url"])).status_code == 200 refused = await client.post("/pages/assets", headers=auth_headers, files={"file": ("x.html", b"