codexpi/tests/test_pwa_factory.py

from __future__ import annotations

import threading
import urllib.request
import zipfile
import sqlite3
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

from codexpi.database import Database
from codexpi.pwa_factory_service import PwaFactoryService


def test_legacy_portfolio_status_constraint_is_migrated_without_losing_rows(tmp_path: Path):
    path = tmp_path / "legacy.sqlite3"
    with sqlite3.connect(path) as db:
        db.execute(
            "CREATE TABLE projects(id TEXT PRIMARY KEY,name TEXT NOT NULL,path TEXT NOT NULL UNIQUE,owner_id TEXT,"
            "create_repository INTEGER NOT NULL DEFAULT 0,engine TEXT NOT NULL DEFAULT 'codex',source_file TEXT NOT NULL DEFAULT '',"
            "additional_files TEXT NOT NULL DEFAULT '[]',full_access INTEGER NOT NULL DEFAULT 0,model TEXT NOT NULL DEFAULT '',"
            "reasoning_effort TEXT NOT NULL DEFAULT 'medium',kind TEXT NOT NULL DEFAULT 'local',"
            "archived INTEGER NOT NULL DEFAULT 0,created_at TEXT NOT NULL)"
        )
        db.execute(
            "CREATE TABLE portfolio_projects("
            "id TEXT PRIMARY KEY,slug TEXT NOT NULL UNIQUE,workspace_id TEXT REFERENCES projects(id) ON DELETE SET NULL,"
            "owner_id TEXT,name TEXT NOT NULL,description TEXT NOT NULL,category TEXT NOT NULL,icon TEXT NOT NULL DEFAULT '◇',"
            "repository TEXT NOT NULL DEFAULT '',branch TEXT NOT NULL DEFAULT 'main',target_directory TEXT NOT NULL,"
            "system_prompt TEXT NOT NULL,initial_prompt TEXT NOT NULL,status TEXT NOT NULL DEFAULT 'CREATED' "
            "CHECK(status IN ('CREATED','TESTING','READY_V0','BLOCKED','FAILED')),priority INTEGER NOT NULL DEFAULT 100,"
            "created_at TEXT NOT NULL,last_activity_at TEXT NOT NULL)"
        )
        db.execute(
            "INSERT INTO portfolio_projects VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            ("one", "one", None, None, "One", "", "TEST", "◇", "", "main", "/tmp/one", "", "", "READY_V0", 1, "now", "now"),
        )
    Database(path).initialize()
    with sqlite3.connect(path) as db:
        db.execute("UPDATE portfolio_projects SET status='SCAFFOLD' WHERE id='one'")
        assert db.execute("SELECT status FROM portfolio_projects WHERE id='one'").fetchone()[0] == "SCAFFOLD"
        assert db.execute("PRAGMA foreign_key_check").fetchall() == []


def test_pwa_project_persists_pages_and_can_be_reopened(tmp_path: Path):
    database = Database(tmp_path / "codexpi.sqlite3")
    database.initialize()
    factory = PwaFactoryService(database)

    created = factory.create(
        "owner-1",
        {
            "name": "Mon Atelier",
            "template": "dashboard",
            "primary_color": "#6d5dfc",
            "background_color": "#f7f5ff",
        },
    )
    updated = factory.update(
        created["id"],
        "owner-1",
        {
            "pages": [
                {
                    "id": "accueil",
                    "title": "Accueil",
                    "body": "Mes indicateurs réels",
                    "components": [{"type": "stats", "title": "Cette semaine"}],
                },
                {
                    "id": "notes",
                    "title": "Notes",
                    "body": "Mes notes persistantes",
                    "components": [{"type": "notes", "title": "Bloc-notes"}],
                },
            ]
        },
    )

    reopened = PwaFactoryService(Database(tmp_path / "codexpi.sqlite3")).get(created["id"], "owner-1")
    assert reopened == updated
    assert [page["title"] for page in reopened["pages"]] == ["Accueil", "Notes"]
    assert factory.get(created["id"], "another-owner") is None


def test_export_is_a_standalone_runnable_pwa(tmp_path: Path):
    database = Database(tmp_path / "codexpi.sqlite3")
    database.initialize()
    factory = PwaFactoryService(database)
    project = factory.create("owner-1", {"name": "Tâches Maison", "template": "tasks"})

    archive = factory.export_zip(project["id"], "owner-1")
    exported = tmp_path / "exported"
    with zipfile.ZipFile(archive) as bundle:
        assert {
            "index.html",
            "app.js",
            "styles.css",
            "manifest.webmanifest",
            "service-worker.js",
            "README.md",
            "icon.svg",
        } <= set(bundle.namelist())
        bundle.extractall(exported)

    class QuietHandler(SimpleHTTPRequestHandler):
        def log_message(self, *_args):
            pass

    server = ThreadingHTTPServer(("127.0.0.1", 0), lambda *args, **kwargs: QuietHandler(*args, directory=exported, **kwargs))
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{server.server_port}/", timeout=3) as response:
            page = response.read().decode("utf-8")
        with urllib.request.urlopen(
            f"http://127.0.0.1:{server.server_port}/manifest.webmanifest", timeout=3
        ) as response:
            manifest = response.read().decode("utf-8")
    finally:
        server.shutdown()
        thread.join(timeout=3)

    assert response.status == 200
    assert "Tâches Maison" in page
    assert '"display": "standalone"' in manifest
    assert "localStorage" in (exported / "app.js").read_text(encoding="utf-8")
    assert "serviceWorker.register" in (exported / "index.html").read_text(encoding="utf-8")


def test_pwa_factory_api_acceptance(logged_in):
    client, _, _, _, headers = logged_in
    created = client.post(
        "/api/pwa-factory/projects",
        headers=headers,
        json={"name": "Catalogue Loutre", "template": "catalog", "primary_color": "#0f766e"},
    )
    assert created.status_code == 201
    project = created.json()

    pages = project["pages"] + [
        {
            "id": "contact",
            "title": "Contact",
            "body": "Écrivez-nous",
            "components": [{"type": "text", "title": "Coordonnées"}],
        }
    ]
    saved = client.put(
        f"/api/pwa-factory/projects/{project['id']}", headers=headers, json={"pages": pages}
    )
    assert saved.status_code == 200
    assert saved.json()["pages"][-1]["title"] == "Contact"

    reopened = client.get(f"/api/pwa-factory/projects/{project['id']}")
    assert reopened.status_code == 200
    assert reopened.json()["pages"] == pages

    exported = client.get(f"/api/pwa-factory/projects/{project['id']}/export")
    assert exported.status_code == 200
    assert exported.headers["content-type"] == "application/zip"
    assert exported.content.startswith(b"PK")

    page = client.get("/pwa-factory/")
    assert page.status_code == 200
    assert "Créer une application" in page.text
    assert "PWA Factory" in page.text