webcarto/backend/tests/test_api.py

import json
import pytest
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, create_engine, Session
from sqlalchemy.pool import StaticPool
from app.main import app
from app.database import get_session
from app.models import Dataset, Feature, FeatureVersion, MapSettings  # noqa: F401

# SQLite in-memory avec StaticPool pour garder la même connexion
engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)


def override_get_session():
    with Session(engine) as session:
        yield session


app.dependency_overrides[get_session] = override_get_session
client = TestClient(app)

SAMPLE_GEOJSON = {
    "type": "FeatureCollection",
    "features": [
        {
            "type": "Feature",
            "geometry": {"type": "Point", "coordinates": [2.35, 48.85]},
            "properties": {"name": "Paris", "description": "Capitale de la France"},
        },
        {
            "type": "Feature",
            "geometry": {"type": "Point", "coordinates": [5.37, 43.30]},
            "properties": {"name": "Marseille"},
        },
    ],
}


@pytest.fixture(autouse=True)
def setup_db():
    SQLModel.metadata.create_all(engine)
    yield
    SQLModel.metadata.drop_all(engine)


def test_health():
    r = client.get("/api/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}


def test_list_datasets_empty():
    r = client.get("/api/datasets")
    assert r.status_code == 200
    assert r.json() == []


def test_import_geojson():
    geojson_str = json.dumps(SAMPLE_GEOJSON)
    r = client.post(
        "/api/datasets/import",
        data={"geojson": geojson_str},
        files={"file": ("test.geojson", geojson_str.encode(), "application/json")},
    )
    assert r.status_code == 200
    data = r.json()
    assert data["name"] == "test"
    assert data["feature_count"] == 2
    assert data["id"] is not None


def test_import_invalid_geojson():
    r = client.post(
        "/api/datasets/import",
        data={"geojson": '{"type": "Point"}'},
        files={"file": ("bad.geojson", b'{"type": "Point"}', "application/json")},
    )
    assert r.status_code == 400


def test_get_dataset():
    geojson_str = json.dumps(SAMPLE_GEOJSON)
    r = client.post(
        "/api/datasets/import",
        data={"geojson": geojson_str},
        files={"file": ("villes.geojson", geojson_str.encode(), "application/json")},
    )
    ds_id = r.json()["id"]

    r = client.get(f"/api/datasets/{ds_id}")
    assert r.status_code == 200
    data = r.json()
    assert data["name"] == "villes"
    assert len(data["features"]) == 2
    assert data["features"][0]["properties"]["name"] == "Paris"


def test_update_feature():
    geojson_str = json.dumps(SAMPLE_GEOJSON)
    r = client.post(
        "/api/datasets/import",
        data={"geojson": geojson_str},
        files={"file": ("test.geojson", geojson_str.encode(), "application/json")},
    )
    ds_id = r.json()["id"]

    r = client.get(f"/api/datasets/{ds_id}")
    feature_id = r.json()["features"][0]["id"]

    r = client.put(
        f"/api/features/{feature_id}",
        json={"properties": {"name": "Paris modifié", "description": "Mis à jour"}},
    )
    assert r.status_code == 200
    assert r.json()["version"] == 1

    r = client.get(f"/api/datasets/{ds_id}")
    f = next(f for f in r.json()["features"] if f["id"] == feature_id)
    assert f["properties"]["name"] == "Paris modifié"


def test_export_dataset():
    geojson_str = json.dumps(SAMPLE_GEOJSON)
    r = client.post(
        "/api/datasets/import",
        data={"geojson": geojson_str},
        files={"file": ("test.geojson", geojson_str.encode(), "application/json")},
    )
    ds_id = r.json()["id"]

    r = client.post(f"/api/datasets/{ds_id}/export?format=geojson")
    assert r.status_code == 200
    fc = r.json()
    assert fc["type"] == "FeatureCollection"
    assert len(fc["features"]) == 2


def test_dataset_not_found():
    r = client.get("/api/datasets/9999")
    assert r.status_code == 404


def test_feature_not_found():
    r = client.put("/api/features/9999", json={"properties": {"name": "x"}})
    assert r.status_code == 404


def test_delete_feature():
    geojson_str = json.dumps(SAMPLE_GEOJSON)
    r = client.post(
        "/api/datasets/import",
        data={"geojson": geojson_str},
        files={"file": ("test.geojson", geojson_str.encode(), "application/json")},
    )
    ds_id = r.json()["id"]
    r = client.get(f"/api/datasets/{ds_id}")
    feature_id = r.json()["features"][0]["id"]

    r = client.delete(f"/api/features/{feature_id}")
    assert r.status_code == 200
    assert r.json()["ok"] is True

    # Vérifier que la feature n'existe plus
    r = client.get(f"/api/datasets/{ds_id}")
    assert len(r.json()["features"]) == 1
    assert r.json()["feature_count"] == 1


def test_delete_dataset():
    geojson_str = json.dumps(SAMPLE_GEOJSON)
    r = client.post(
        "/api/datasets/import",
        data={"geojson": geojson_str},
        files={"file": ("test.geojson", geojson_str.encode(), "application/json")},
    )
    ds_id = r.json()["id"]

    r = client.delete(f"/api/datasets/{ds_id}")
    assert r.status_code == 200
    assert r.json()["ok"] is True

    # Vérifier que le dataset n'existe plus
    r = client.get(f"/api/datasets/{ds_id}")
    assert r.status_code == 404