test(backend): add endpoint tests for feedback, stars, and notifications apps (#229)

backend/tests/test_feedback_endpoints.py

@@ -0,0 +1,170 @@
+import pytest
+from django.contrib.gis.geos import Polygon
+from rest_framework.test import APIClient
+
+from accounts.models import OsmUser
+from feedback.models import Feedback
+
+FEEDBACK_URL = "/api/v1/feedback/"
+
+_POLYGON_GEOJSON = {
+    "type": "Polygon",
+    "coordinates": [[[0, 0], [0, 1], [1, 1], [1, 0], [0, 0]]],
+}
+
+
+@pytest.fixture
+def authed_user(db) -> OsmUser:
+    return OsmUser.objects.create(osm_id=42, username="alice")
+
+
+@pytest.fixture
+def client(authed_user: OsmUser) -> APIClient:
+    api = APIClient()
+    api.force_authenticate(user=authed_user)
+    return api
+
+
+@pytest.fixture
+def other_user(db) -> OsmUser:
+    return OsmUser.objects.create(osm_id=43, username="bob")
+
+
+@pytest.fixture
+def other_client(other_user: OsmUser) -> APIClient:
+    api = APIClient()
+    api.force_authenticate(user=other_user)
+    return api
+
+
+def _make_feedback(user: OsmUser, **overrides) -> Feedback:
+    defaults = {
+        "stac_id": "stac-1",
+        "geom": Polygon(((0, 0), (0, 1), (1, 1), (1, 0), (0, 0)), srid=4326),
+        "action": Feedback.Action.ACCEPT,
+        "user": user,
+    }
+    defaults.update(overrides)
+    return Feedback.objects.create(**defaults)
+
+
+def _features(payload: dict) -> list:
+    """Unwrap a paginated GeoJSON FeatureCollection list response."""
+    results = payload.get("results", payload)
+    if isinstance(results, dict) and "features" in results:
+        return results["features"]
+    return results
+
+
+def test_feedback_requires_authentication(db):
+    response = APIClient().get(FEEDBACK_URL)
+    assert response.status_code == 401
+
+
+def test_feedback_create_sets_request_user(client, authed_user):
+    response = client.post(
+        FEEDBACK_URL,
+        data={
+            "type": "Feature",
+            "geometry": _POLYGON_GEOJSON,
+            "properties": {
+                "stac_id": "stac-1",
+                "action": "accept",
+                "comments": "looks right",
+            },
+        },
+        format="json",
+    )
+
+    assert response.status_code == 201
+    feature = response.json()
+    assert feature["properties"]["stac_id"] == "stac-1"
+    assert feature["properties"]["action"] == "accept"
+    assert feature["properties"]["user"]["osm_id"] == authed_user.osm_id
+    assert feature["geometry"]["type"] == "Polygon"
+    assert Feedback.objects.count() == 1
+    assert Feedback.objects.get().user == authed_user
+
+
+def test_feedback_create_rejects_invalid_geometry(client):
+    response = client.post(
+        FEEDBACK_URL,
+        data={
+            "type": "Feature",
+            "geometry": {"type": "Polygon", "coordinates": "not-coordinates"},
+            "properties": {"stac_id": "stac-1", "action": "accept"},
+        },
+        format="json",
+    )
+
+    assert response.status_code == 400
+
+
+def test_feedback_create_rejects_unknown_action(client):
+    response = client.post(
+        FEEDBACK_URL,
+        data={
+            "type": "Feature",
+            "geometry": _POLYGON_GEOJSON,
+            "properties": {"stac_id": "stac-1", "action": "maybe"},
+        },
+        format="json",
+    )
+
+    assert response.status_code == 400
+
+
+def test_feedback_list_filters_by_stac_id_and_action(client, authed_user):
+    _make_feedback(authed_user, stac_id="stac-1")
+    _make_feedback(authed_user, stac_id="stac-2", action=Feedback.Action.REJECT)
+
+    response = client.get(FEEDBACK_URL, {"stac_id": "stac-1"})
+    assert response.status_code == 200
+    features = _features(response.json())
+    assert len(features) == 1
+    assert features[0]["properties"]["stac_id"] == "stac-1"
+
+    response = client.get(FEEDBACK_URL, {"action": "reject"})
+    assert response.status_code == 200
+    features = _features(response.json())
+    assert len(features) == 1
+    assert features[0]["properties"]["stac_id"] == "stac-2"
+
+
+def test_feedback_readable_by_other_users(other_client, authed_user):
+    feedback = _make_feedback(authed_user)
+
+    response = other_client.get(f"{FEEDBACK_URL}{feedback.id}/")
+    assert response.status_code == 200
+    assert response.json()["properties"]["user"]["osm_id"] == authed_user.osm_id
+
+
+def test_feedback_update_denied_for_non_owner(other_client, authed_user):
+    feedback = _make_feedback(authed_user)
+
+    response = other_client.patch(
+        f"{FEEDBACK_URL}{feedback.id}/",
+        data={"properties": {"comments": "hijacked"}},
+        format="json",
+    )
+
+    assert response.status_code == 403
+    feedback.refresh_from_db()
+    assert feedback.comments == ""
+
+
+def test_feedback_owner_can_update_and_delete(client, authed_user):
+    feedback = _make_feedback(authed_user)
+
+    response = client.patch(
+        f"{FEEDBACK_URL}{feedback.id}/",
+        data={"properties": {"comments": "updated"}},
+        format="json",
+    )
+    assert response.status_code == 200
+    feedback.refresh_from_db()
+    assert feedback.comments == "updated"
+
+    response = client.delete(f"{FEEDBACK_URL}{feedback.id}/")
+    assert response.status_code == 204
+    assert Feedback.objects.count() == 0

backend/tests/test_notification_endpoints.py

@@ -0,0 +1,113 @@
+from datetime import timedelta
+
+import pytest
+from django.utils import timezone
+from rest_framework.test import APIClient
+
+from accounts.models import OsmUser
+from notifications.models import Banner, UserNotification
+
+BANNERS_URL = "/api/v1/banners/"
+NOTIFICATIONS_URL = "/api/v1/notifications/me/"
+
+
+@pytest.fixture
+def authed_user(db) -> OsmUser:
+    return OsmUser.objects.create(osm_id=42, username="alice")
+
+
+@pytest.fixture
+def client(authed_user: OsmUser) -> APIClient:
+    api = APIClient()
+    api.force_authenticate(user=authed_user)
+    return api
+
+
+@pytest.fixture
+def other_user(db) -> OsmUser:
+    return OsmUser.objects.create(osm_id=43, username="bob")
+
+
+def test_banner_list_is_public_and_hides_unstarted_banners(db):
+    now = timezone.now()
+    Banner.objects.create(message="live now", start_date=now - timedelta(days=1))
+    Banner.objects.create(message="coming soon", start_date=now + timedelta(days=1))
+
+    response = APIClient().get(BANNERS_URL)
+
+    assert response.status_code == 200
+    messages = [banner["message"] for banner in response.json()["results"]]
+    assert messages == ["live now"]
+
+
+def test_banner_serializer_exposes_displayable_state(db):
+    now = timezone.now()
+    Banner.objects.create(
+        message="expired",
+        start_date=now - timedelta(days=2),
+        end_date=now - timedelta(days=1),
+    )
+
+    response = APIClient().get(BANNERS_URL)
+
+    assert response.status_code == 200
+    banners = response.json()["results"]
+    assert len(banners) == 1
+    # Started-but-expired banners are still listed, flagged not displayable
+    assert banners[0]["is_displayable"] is False
+
+
+def test_notifications_require_authentication(db):
+    response = APIClient().get(NOTIFICATIONS_URL)
+    assert response.status_code == 401
+
+
+def test_notifications_list_only_own(client, authed_user, other_user):
+    UserNotification.objects.create(user=authed_user, message="for alice")
+    UserNotification.objects.create(user=other_user, message="for bob")
+
+    response = client.get(NOTIFICATIONS_URL)
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["count"] == 1
+    assert payload["results"][0]["message"] == "for alice"
+    assert payload["results"][0]["is_read"] is False
+
+
+def test_mark_read_sets_read_state(client, authed_user):
+    notification = UserNotification.objects.create(user=authed_user, message="hi")
+
+    response = client.post(f"{NOTIFICATIONS_URL}{notification.id}/mark-read/")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["is_read"] is True
+    assert body["read_at"] is not None
+    notification.refresh_from_db()
+    assert notification.is_read is True
+    assert notification.read_at is not None
+
+
+def test_mark_read_scoped_to_own_notifications(client, other_user):
+    notification = UserNotification.objects.create(user=other_user, message="not yours")
+
+    response = client.post(f"{NOTIFICATIONS_URL}{notification.id}/mark-read/")
+
+    assert response.status_code == 404
+    notification.refresh_from_db()
+    assert notification.is_read is False
+
+
+def test_mark_all_read_updates_only_own_unread(client, authed_user, other_user):
+    UserNotification.objects.create(user=authed_user, message="one")
+    UserNotification.objects.create(user=authed_user, message="two")
+    other_notification = UserNotification.objects.create(user=other_user, message="other")
+
+    response = client.post(f"{NOTIFICATIONS_URL}mark-all-read/")
+
+    assert response.status_code == 200
+    assert response.json() == {"detail": "ok"}
+    assert UserNotification.objects.filter(user=authed_user, is_read=False).count() == 0
+    other_notification.refresh_from_db()
+    assert other_notification.is_read is False

backend/tests/test_star_endpoints.py

@@ -0,0 +1,102 @@
+import pytest
+from rest_framework.test import APIClient
+
+from accounts.models import OsmUser
+from stars.models import Star
+
+STARS_URL = "/api/v1/stars/"
+
+
+@pytest.fixture
+def authed_user(db) -> OsmUser:
+    return OsmUser.objects.create(osm_id=42, username="alice")
+
+
+@pytest.fixture
+def client(authed_user: OsmUser) -> APIClient:
+    api = APIClient()
+    api.force_authenticate(user=authed_user)
+    return api
+
+
+@pytest.fixture
+def anon_client(db) -> APIClient:
+    return APIClient()
+
+
+def test_star_get_requires_target_id(anon_client):
+    response = anon_client.get(STARS_URL)
+    assert response.status_code == 400
+    assert "target_id" in response.json()["detail"]
+
+
+def test_star_post_requires_target_id(anon_client):
+    response = anon_client.post(STARS_URL)
+    assert response.status_code == 400
+
+
+def test_anonymous_star_dedupes_repeat_clicks(anon_client):
+    response = anon_client.post(f"{STARS_URL}?target_id=model-1")
+    assert response.status_code == 201
+    payload = response.json()
+    assert payload == {
+        "target_id": "model-1",
+        "starred": True,
+        "count": 1,
+        "created": True,
+    }
+
+    # Same client (same IP + user agent) starring again must not double-count
+    response = anon_client.post(f"{STARS_URL}?target_id=model-1")
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["created"] is False
+    assert payload["count"] == 1
+
+
+def test_distinct_anonymous_clients_count_separately(anon_client):
+    anon_client.post(f"{STARS_URL}?target_id=model-1")
+    response = anon_client.post(
+        f"{STARS_URL}?target_id=model-1",
+        HTTP_USER_AGENT="a-different-browser",
+    )
+
+    assert response.status_code == 201
+    assert response.json()["count"] == 2
+
+
+def test_authenticated_star_state(client, authed_user, anon_client):
+    response = client.post(f"{STARS_URL}?target_id=model-2")
+    assert response.status_code == 201
+    assert Star.objects.get(target_id="model-2").user == authed_user
+
+    response = client.get(STARS_URL, {"target_id": "model-2"})
+    assert response.status_code == 200
+    assert response.json() == {"target_id": "model-2", "count": 1, "starred": True}
+
+    # An anonymous viewer sees the count but is not the one who starred
+    response = anon_client.get(STARS_URL, {"target_id": "model-2"})
+    assert response.status_code == 200
+    assert response.json() == {"target_id": "model-2", "count": 1, "starred": False}
+
+
+def test_authenticated_unstar(client):
+    client.post(f"{STARS_URL}?target_id=model-3")
+
+    response = client.delete(f"{STARS_URL}?target_id=model-3")
+    assert response.status_code == 204
+
+    response = client.get(STARS_URL, {"target_id": "model-3"})
+    assert response.json() == {"target_id": "model-3", "count": 0, "starred": False}
+
+
+def test_anonymous_unstar_removes_only_own_star(client, anon_client):
+    client.post(f"{STARS_URL}?target_id=model-4")
+    anon_client.post(f"{STARS_URL}?target_id=model-4")
+    assert Star.objects.filter(target_id="model-4").count() == 2
+
+    response = anon_client.delete(f"{STARS_URL}?target_id=model-4")
+    assert response.status_code == 204
+
+    response = anon_client.get(STARS_URL, {"target_id": "model-4"})
+    assert response.json() == {"target_id": "model-4", "count": 1, "starred": False}