diff --git a/backend/tests/test_feedback_endpoints.py b/backend/tests/test_feedback_endpoints.py new file mode 100644 index 00000000..adae92bf --- /dev/null +++ b/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 diff --git a/backend/tests/test_notification_endpoints.py b/backend/tests/test_notification_endpoints.py new file mode 100644 index 00000000..a0e4ee7a --- /dev/null +++ b/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 diff --git a/backend/tests/test_star_endpoints.py b/backend/tests/test_star_endpoints.py new file mode 100644 index 00000000..2432aa30 --- /dev/null +++ b/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}