codexpi/tests/test_project_groups.py

from pathlib import Path

from codexpi.project_group_service import project_group_context, publish_project_group_update


def register_project(logged_in, name: str) -> dict:
    client, _, projects, _, headers = logged_in
    root = projects / name.lower().replace(" ", "-")
    root.mkdir()
    response = client.post("/api/projects", headers=headers, json={"name": name, "path": str(root)})
    assert response.status_code == 201
    return response.json()


def test_group_creation_selects_existing_projects_and_requires_csrf(logged_in):
    client, app, _, _, headers = logged_in
    nasflix = register_project(logged_in, "NASflix")
    ginoflex = register_project(logged_in, "Ginoflex")
    tvrework = register_project(logged_in, "TvRework")
    payload = {
        "name": "TV",
        "description": "Applications qui pilotent ou affichent du contenu sur la télévision.",
        "project_ids": [nasflix["id"], ginoflex["id"], tvrework["id"]],
    }

    assert client.post("/api/project-groups", json=payload).status_code == 403
    created = client.post("/api/project-groups", headers=headers, json=payload)

    assert created.status_code == 201
    group = created.json()
    assert group["name"] == "TV"
    assert group["key"] == "tv"
    assert {item["name"] for item in group["members"]} == {"NASflix", "Ginoflex", "TvRework"}
    assert {
        row["execution_group"]
        for row in app.state.db.all(
            "SELECT execution_group FROM projects WHERE id IN (?,?,?)",
            (nasflix["id"], ginoflex["id"], tvrework["id"]),
        )
    } == {"tv"}
    assert client.get("/api/project-groups").json()[0]["description"].startswith("Applications")


def test_group_update_moves_members_and_delete_releases_them(logged_in):
    client, app, _, _, headers = logged_in
    first = register_project(logged_in, "Premier")
    second = register_project(logged_in, "Deuxième")
    third = register_project(logged_in, "Troisième")
    created = client.post(
        "/api/project-groups",
        headers=headers,
        json={"name": "Télévision", "project_ids": [first["id"], second["id"]]},
    ).json()

    updated = client.put(
        f"/api/project-groups/{created['id']}",
        headers=headers,
        json={"name": "TV salon", "description": "Contrats HDMI et télécommande", "project_ids": [first["id"], third["id"]]},
    )

    assert updated.status_code == 200
    assert updated.json()["key"] == "tv-salon"
    assert {item["id"] for item in updated.json()["members"]} == {first["id"], third["id"]}
    assert app.state.db.one("SELECT execution_group FROM projects WHERE id=?", (second["id"],))["execution_group"] == ""

    deleted = client.delete(f"/api/project-groups/{created['id']}", headers=headers)
    assert deleted.status_code == 204
    assert client.get("/api/project-groups").json() == []
    assert app.state.db.one("SELECT execution_group FROM projects WHERE id=?", (first["id"],))["execution_group"] == ""


def test_project_can_belong_to_multiple_groups(logged_in):
    client, app, _, _, headers = logged_in
    tv = register_project(logged_in, "TV commune")
    player = register_project(logged_in, "Lecteur")
    media = register_project(logged_in, "Médiathèque")
    first = client.post(
        "/api/project-groups",
        headers=headers,
        json={"name": "Télévision", "project_ids": [tv["id"], player["id"]]},
    )
    second = client.post(
        "/api/project-groups",
        headers=headers,
        json={"name": "Médias", "project_ids": [player["id"], media["id"]]},
    )

    assert first.status_code == 201
    assert second.status_code == 201
    groups = client.get("/api/project-groups").json()
    assert len(groups) == 2
    memberships = app.state.db.all(
        "SELECT group_id FROM project_group_members WHERE project_id=?",
        (player["id"],),
    )
    assert len(memberships) == 2
    context = project_group_context(app.state.db.path, player["id"])
    assert "[GROUPE DE PROJETS : Télévision]" in context
    assert "[GROUPE DE PROJETS : Médias]" in context


def test_group_rejects_remote_projects(logged_in):
    client, app, _, _, headers = logged_in
    local = register_project(logged_in, "Local")
    remote = register_project(logged_in, "Distant")
    app.state.db.execute("UPDATE projects SET kind='remote' WHERE id=?", (remote["id"],))

    response = client.post(
        "/api/project-groups",
        headers=headers,
        json={"name": "Mixte", "project_ids": [local["id"], remote["id"]]},
    )

    assert response.status_code == 400
    assert "distants" in response.json()["detail"]


def test_group_keeps_members_when_full_access_is_enabled(logged_in):
    client, app, _, _, headers = logged_in
    first = register_project(logged_in, "Premier projet")
    second = register_project(logged_in, "Deuxième projet")
    enabled = client.put(
        f"/api/projects/{second['id']}/full-access",
        headers=headers,
        json={"enabled": True, "acknowledged": True},
    )
    assert enabled.status_code == 200

    created = client.post(
        "/api/project-groups",
        headers=headers,
        json={"name": "Coordonné", "project_ids": [first["id"], second["id"]]},
    )
    assert created.status_code == 201
    enabled_first = client.put(
        f"/api/projects/{first['id']}/full-access",
        headers=headers,
        json={"enabled": True, "acknowledged": True},
    )
    assert enabled_first.status_code == 200
    assert app.state.db.one(
        "SELECT COUNT(*) AS count FROM project_group_members WHERE group_id=?",
        (created.json()["id"],),
    )["count"] == 2

    context = project_group_context(
        app.state.db.path,
        first["id"],
        engine="claude",
        full_access=True,
    )
    assert f"Deuxième projet : {Path(second['path'])}" in context
    assert "aucune frontière technique" in context
    assert "demande explicitement" in context


def test_group_context_exposes_other_code_read_only_and_shared_updates(logged_in):
    client, app, _, _, headers = logged_in
    sender = register_project(logged_in, "NASflix")
    receiver = register_project(logged_in, "TvReWork")
    created = client.post(
        "/api/project-groups",
        headers=headers,
        json={"name": "TV", "description": "Applications connectées à la TV", "project_ids": [sender["id"], receiver["id"]]},
    )
    assert created.status_code == 201
    conversation = client.post(
        f"/api/projects/{sender['id']}/conversations",
        headers=headers,
        json={"title": "Contrat partagé"},
    ).json()
    publish_project_group_update(
        app.state.db.path,
        sender["id"],
        conversation["id"],
        "Le contrat HDMI accepte désormais la commande play.",
    )

    context = project_group_context(app.state.db.path, receiver["id"])

    assert "[GROUPE DE PROJETS : TV]" in context
    assert f"NASflix : {Path(sender['path'])}" in context
    assert "lecture seule" in context
    assert "N’écris jamais" in context
    assert "Le contrat HDMI accepte désormais la commande play" in context
    assert receiver["path"] not in context

    untrusted_context = project_group_context(
        app.state.db.path,
        receiver["id"],
        engine="claude",
    )
    assert str(Path(sender["path"])) not in untrusted_context
    assert "uniquement leurs comptes rendus" in untrusted_context
    assert "Le contrat HDMI accepte désormais la commande play" in untrusted_context

    full_access_context = project_group_context(
        app.state.db.path,
        receiver["id"],
        engine="claude",
        full_access=True,
    )
    assert f"NASflix : {Path(sender['path'])}" in full_access_context
    assert "aucune frontière technique" in full_access_context