add buttons to copy and paste geospatial metadata based on eoextent

Author: nuest

CHANGELOG.md

@@ -11,6 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - **Copy/paste geometries between the geoextent tool and the contribution form.** A new "Copy extents" button on `/geoextent/` saves all extracted geometries to the browser; a "Paste geoextents" button on the work landing page (when contributing spatial extent) pastes them into the map as editable layers ready to submit.
 
+- **Admin-only "Publish all unpublished works" button on collection pages** — bulk-flips every Harvested or Contributed work in the collection to Published in one click. Curators see no button (admins-only). Draft / Testing / Withdrawn works are deliberately left untouched.
+
+- **Mountain Wetlands and OpenAlex-as-source harvesters now auto-create a Collection** for the source on first run (mirroring the OAI-PMH path from issue #192). New collections start unpublished so admins can review name/description before exposing them on `/collections/`.
 
 - **Separate "Unpublished works" map layer** for admins (main map) and collection curators (collection pages). Features split into two togglable layers, *Published works (N)* and *Unpublished works (N)* (dashed, muted). Popups for unpublished features show a status badge.
 

tests/test_collections.py

@@ -364,6 +364,121 @@ def test_publish_then_unpublish(self):
         self.assertFalse(self.col.is_published)
 
 
+class PublishCollectionWorksTests(TestCase):
+    """Admin-only bulk action to flip Harvested/Contributed works to Published.
+
+    Curators (non-staff) must be rejected — explicitly tested because curators
+    can otherwise edit collection descriptions and add/remove individual works.
+    """
+
+    def setUp(self):
+        self.client = Client()
+        self.col = Collection.objects.create(identifier='c', name='C', is_published=True)
+        self.admin = User.objects.create_user(
+            username='admin@x.com', email='admin@x.com', password='p123', is_staff=True,
+        )
+        self.curator = User.objects.create_user(
+            username='c@x.com', email='c@x.com', password='p123',
+        )
+        self.col.curators.add(self.curator)
+
+        self.harvested = Work.objects.create(
+            title='H', status='h', doi='10.1234/h',
+            geometry=GeometryCollection(Point(0, 0)),
+        )
+        self.contributed = Work.objects.create(
+            title='C', status='c', doi='10.1234/c',
+            geometry=GeometryCollection(Point(1, 1)),
+        )
+        self.draft = Work.objects.create(
+            title='D', status='d', doi='10.1234/d',
+            geometry=GeometryCollection(Point(2, 2)),
+        )
+        self.withdrawn = Work.objects.create(
+            title='W', status='w', doi='10.1234/w',
+            geometry=GeometryCollection(Point(3, 3)),
+        )
+        self.already_published = Work.objects.create(
+            title='P', status='p', doi='10.1234/p',
+            geometry=GeometryCollection(Point(4, 4)),
+        )
+        for w in (self.harvested, self.contributed, self.draft, self.withdrawn, self.already_published):
+            w.collections.add(self.col)
+
+        # Work in another collection — must not be touched.
+        self.other_col = Collection.objects.create(identifier='other', name='Other', is_published=True)
+        self.outsider_work = Work.objects.create(
+            title='O', status='h', doi='10.1234/o',
+            geometry=GeometryCollection(Point(5, 5)),
+        )
+        self.outsider_work.collections.add(self.other_col)
+
+    def _url(self):
+        return f'/collections/{self.col.id}/publish-works/'
+
+    def test_admin_publishes_harvested_and_contributed_only(self):
+        self.client.force_login(self.admin)
+        resp = self.client.post(self._url())
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.json(), {'success': True, 'published_count': 2})
+
+        for w in (self.harvested, self.contributed):
+            w.refresh_from_db()
+            self.assertEqual(w.status, 'p')
+        # Draft / Withdrawn deliberately untouched.
+        self.draft.refresh_from_db()
+        self.assertEqual(self.draft.status, 'd')
+        self.withdrawn.refresh_from_db()
+        self.assertEqual(self.withdrawn.status, 'w')
+        # Other collection's work untouched.
+        self.outsider_work.refresh_from_db()
+        self.assertEqual(self.outsider_work.status, 'h')
+
+    def test_curator_is_rejected(self):
+        self.client.force_login(self.curator)
+        resp = self.client.post(self._url())
+        # staff_member_required redirects to admin login for non-staff.
+        self.assertEqual(resp.status_code, 302)
+        self.harvested.refresh_from_db()
+        self.assertEqual(self.harvested.status, 'h')
+
+    def test_anonymous_is_rejected(self):
+        resp = self.client.post(self._url())
+        self.assertEqual(resp.status_code, 302)
+        self.harvested.refresh_from_db()
+        self.assertEqual(self.harvested.status, 'h')
+
+    def test_get_is_not_allowed(self):
+        self.client.force_login(self.admin)
+        resp = self.client.get(self._url())
+        self.assertEqual(resp.status_code, 405)
+
+    def test_button_visible_to_admin_with_count(self):
+        self.client.force_login(self.admin)
+        resp = self.client.get(reverse('optimap:collection-page', args=[self.col.identifier]))
+        self.assertEqual(resp.status_code, 200)
+        body = resp.content.decode()
+        self.assertIn('collection-publish-works-btn', body)
+        # 2 works are h/c, count rendered in the button label.
+        self.assertIn('Publish all 2 unpublished works', body)
+
+    def test_button_hidden_when_no_publishable_works(self):
+        # Flip the two candidates to Published — button disappears. The class
+        # name and ``data-publishable-count`` both appear in the admin's JS
+        # handler block, so assert on the human-visible label that only the
+        # rendered button carries.
+        self.harvested.status = 'p'; self.harvested.save()
+        self.contributed.status = 'p'; self.contributed.save()
+        self.client.force_login(self.admin)
+        resp = self.client.get(reverse('optimap:collection-page', args=[self.col.identifier]))
+        self.assertNotIn('Publish all', resp.content.decode())
+
+    def test_button_hidden_for_curator(self):
+        self.client.force_login(self.curator)
+        resp = self.client.get(reverse('optimap:collection-page', args=[self.col.identifier]))
+        self.assertNotIn('Publish all', resp.content.decode())
+
+
 class CuratorAddRemoveTests(TestCase):
     def setUp(self):
         self.client = Client()

tests/test_geocoding.py

@@ -242,28 +242,73 @@ def test_failed_lookups_skip_address_but_count_stays_at_successes(self):
         self.assertIn("Berlin", placename)
         self.assertEqual(n, 1)
 
-    def test_polygon_uses_one_representative_interior_point(self):
-        # A 4-vertex polygon has 5 ring points (closed) — we must NOT
-        # geocode each vertex; one representative point is enough.
+    def test_polygon_samples_envelope_corners_plus_interior(self):
+        # A polygon must sample the four corners of its bounding box plus an
+        # interior point, so cross-border polygons aren't reduced to a single
+        # interior placename. The 4-vertex closing ring is still NOT
+        # geocoded vertex-by-vertex (would explode for high-resolution rings).
         from django.contrib.gis.geos import Polygon
         poly = Polygon(
             ((10.0, 50.0), (11.0, 50.0), (11.0, 51.0), (10.0, 51.0), (10.0, 50.0)),
             srid=4326,
         )
         gc = GeometryCollection(poly, srid=4326)
-        call_count = {"n": 0}
+        seen_coords: list[tuple[float, float]] = []
 
         def lookup(lat, lon):
-            call_count["n"] += 1
+            seen_coords.append((round(lat, 3), round(lon, 3)))
             return {"address": _BERLIN, "display_name": "Berlin"}
 
         with mock.patch.object(
             geocoding, "_reverse_geocode_lookup", side_effect=lookup,
         ):
             placename, country, n = geocoding.geocode_geometry(gc)
-        self.assertEqual(call_count["n"], 1, "polygon should geocode once")
+        # Four corners + one interior point — five lookups, not the five
+        # ring vertices (which would include the duplicated closing point).
+        self.assertEqual(len(seen_coords), 5, f"polygon: corners + interior, got {seen_coords}")
+        self.assertEqual(set([c for c in seen_coords if c in {
+            (50.0, 10.0), (50.0, 11.0), (51.0, 10.0), (51.0, 11.0),
+        }]), {(50.0, 10.0), (50.0, 11.0), (51.0, 10.0), (51.0, 11.0)},
+            "all four envelope corners must be sampled")
         self.assertEqual(country, "DE")
 
+    def test_polygon_spanning_two_countries_lca_falls_back(self):
+        # Bug demoed by work id=10 on the deployed instance: a polygon
+        # straddling Germany and Poland was getting "Mniszki, …, Poland" as
+        # placename because only the centroid was geocoded. With corners
+        # sampled, two corners hit Germany, two hit Poland → LCA collapses
+        # past country (no shared continent in our test fixture either) and
+        # the work no longer claims a misleading specific placename.
+        from django.contrib.gis.geos import Polygon
+        # Envelope corners → (lat, lon):
+        #   (50, 10) DE, (50, 20) PL, (52, 10) DE, (52, 20) PL
+        # Interior (point_on_surface for an axis-aligned rectangle) → (51, 15) — DE.
+        _POLAND = {
+            "country_code": "pl", "country": "Poland",
+            "state": "Łódź Voivodeship", "city": "Mniszki",
+        }
+        lookup = _LookupTable({
+            (50.0, 10.0): {"address": _BERLIN,  "display_name": "Berlin"},
+            (50.0, 20.0): {"address": _POLAND,  "display_name": "Mniszki"},
+            (52.0, 10.0): {"address": _BERLIN,  "display_name": "Berlin"},
+            (52.0, 20.0): {"address": _POLAND,  "display_name": "Mniszki"},
+            (51.0, 15.0): {"address": _BERLIN,  "display_name": "Berlin"},
+        })
+        poly = Polygon(
+            ((10.0, 50.0), (20.0, 50.0), (20.0, 52.0), (10.0, 52.0), (10.0, 50.0)),
+            srid=4326,
+        )
+        gc = GeometryCollection(poly, srid=4326)
+        with mock.patch.object(
+            geocoding, "_reverse_geocode_lookup", side_effect=lookup,
+        ):
+            placename, country, n = geocoding.geocode_geometry(gc)
+        # Two countries appear among the corners → no shared country → both
+        # placename and country must be None, not the centroid's interior.
+        self.assertIsNone(placename, f"expected None, got {placename!r}")
+        self.assertIsNone(country)
+        self.assertEqual(n, 5)
+
     def test_max_points_caps_geocoder_calls(self):
         # 30 points but max_points=5 → only 5 lookups.
         gc = GeometryCollection(

tests/test_harvester_collection_auto_create.py

@@ -0,0 +1,115 @@
+# SPDX-FileCopyrightText: 2026 OPTIMETA and KOMET projects <https://projects.tib.eu/komet>
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""Generalisation of issue #192: every harvest entry point auto-creates a
+``Collection`` for its source on first run if none is pre-assigned. The OAI-PMH
+path is covered by ``test_oai_collection_auto_create.py``; this module covers
+the Mountain Wetlands and OpenAlex-as-source harvesters that previously
+required an admin (or fixture) to pre-seed a Collection.
+"""
+
+from unittest.mock import patch, Mock
+
+from django.test import TestCase
+
+from works.models import Collection, Source
+
+
+class MountainWetlandsAutoCreatesCollectionTests(TestCase):
+    def setUp(self):
+        self.source = Source.objects.create(
+            name='Mountain Wetlands Repository',
+            url_field='https://andes.example/api/v1/items/',
+            source_type='mountain-wetlands',
+        )
+        self.assertIsNone(self.source.collection)
+
+    @patch('works.harvesting.mountain_wetlands._mwr_session')
+    def test_first_harvest_creates_collection_and_links_source(self, mock_session_factory):
+        from works.tasks import harvest_mountain_wetlands
+
+        # Empty page — short-circuit the loop without saving any works.
+        fake_response = Mock()
+        fake_response.ok = True
+        fake_response.status_code = 200
+        fake_response.json.return_value = {'count': 0, 'data': []}
+        mock_session = Mock()
+        mock_session.get.return_value = fake_response
+        mock_session_factory.return_value = mock_session
+
+        harvest_mountain_wetlands(self.source.id)
+
+        self.source.refresh_from_db()
+        self.assertIsNotNone(self.source.collection)
+        self.assertEqual(self.source.collection.identifier, 'mountain-wetlands-repository')
+        self.assertFalse(self.source.collection.is_published)
+        self.assertEqual(Collection.objects.count(), 1)
+
+    @patch('works.harvesting.mountain_wetlands._mwr_session')
+    def test_preassigned_collection_is_left_alone(self, mock_session_factory):
+        from works.tasks import harvest_mountain_wetlands
+
+        preassigned = Collection.objects.create(
+            identifier='preassigned-mw', name='Pre-assigned MW', is_published=True,
+        )
+        self.source.collection = preassigned
+        self.source.save(update_fields=['collection'])
+
+        fake_response = Mock()
+        fake_response.ok = True
+        fake_response.status_code = 200
+        fake_response.json.return_value = {'count': 0, 'data': []}
+        mock_session = Mock()
+        mock_session.get.return_value = fake_response
+        mock_session_factory.return_value = mock_session
+
+        harvest_mountain_wetlands(self.source.id)
+
+        self.source.refresh_from_db()
+        self.assertEqual(self.source.collection_id, preassigned.id)
+        self.source.collection.refresh_from_db()
+        self.assertTrue(self.source.collection.is_published)
+
+
+class OpenalexSourceAutoCreatesCollectionTests(TestCase):
+    def setUp(self):
+        self.source = Source.objects.create(
+            name='AGILE GIScience Series (OpenAlex)',
+            url_field='https://api.openalex.org/sources/S4210203054',
+            source_type='openalex',
+            openalex_id='S4210203054',
+        )
+        self.assertIsNone(self.source.collection)
+
+    @patch('works.harvesting.openalex_source.parse_openalex_response_and_save_works')
+    def test_first_harvest_creates_collection_and_links_source(self, mock_parser):
+        from works.tasks import harvest_openalex_source
+
+        # Skip the API loop entirely — only the wiring around it matters here.
+        mock_parser.return_value = (0, 0)
+
+        harvest_openalex_source(self.source.id, max_records=0)
+
+        self.source.refresh_from_db()
+        self.assertIsNotNone(self.source.collection)
+        self.assertEqual(self.source.collection.identifier, 'agile-giscience-series-openalex')
+        self.assertFalse(self.source.collection.is_published)
+        self.assertEqual(Collection.objects.count(), 1)
+
+    @patch('works.harvesting.openalex_source.parse_openalex_response_and_save_works')
+    def test_preassigned_collection_is_left_alone(self, mock_parser):
+        from works.tasks import harvest_openalex_source
+
+        mock_parser.return_value = (0, 0)
+        preassigned = Collection.objects.create(
+            identifier='preassigned-agile', name='Pre-assigned AGILE', is_published=True,
+        )
+        self.source.collection = preassigned
+        self.source.save(update_fields=['collection'])
+
+        harvest_openalex_source(self.source.id, max_records=0)
+
+        self.source.refresh_from_db()
+        self.assertEqual(self.source.collection_id, preassigned.id)
+        self.source.collection.refresh_from_db()
+        self.assertTrue(self.source.collection.is_published)

works/services/geocoding.py

@@ -161,30 +161,61 @@ def _representative_points(geom, max_points: int = 20) -> list[tuple[float, floa
 
     Each ``Point`` in the geometry is sampled individually (so multi-point
     GeometryCollections are geocoded per-site). Polygons / lines contribute
-    one interior representative point. Bounded so a 500-vertex polygon
+    the four corners of their bounding box plus an interior point — sampling
+    only an interior point would mask cross-border coverage: a Polygon
+    spanning Germany and Poland geocodes through its centroid (which lands
+    in one country) and the LCA blindly reports that country, even though
+    the polygon's corners clearly cross the border. Sampling the corners
+    forces ``_common_address`` to detect the divergence and fall back to
+    the shared ancestor (or to ``None``). Bounded so a 500-vertex polygon
     doesn't trigger 500 Nominatim requests.
     """
     points: list[tuple[float, float]] = []
+    seen: set[tuple[float, float]] = set()
+
+    def _add(lat: float, lon: float):
+        # Quantise to the cache resolution (~100 m) so envelope corners that
+        # round to the same grid cell as an interior point don't waste a
+        # Nominatim hit on the same neighbourhood.
+        key = (round(lat, 3), round(lon, 3))
+        if key in seen:
+            return
+        seen.add(key)
+        points.append((lat, lon))
 
     def _walk(g):
         if len(points) >= max_points:
             return
         gt = g.geom_type
         if gt == "Point":
-            points.append((g.y, g.x))
+            _add(g.y, g.x)
         elif gt in ("MultiPoint", "GeometryCollection",
                     "MultiPolygon", "MultiLineString"):
             for child in g:
                 _walk(child)
                 if len(points) >= max_points:
                     return
         else:
-            # Polygon, LineString, LinearRing — pick one interior point.
+            # Polygon, LineString, LinearRing — sample the envelope corners
+            # plus an interior point. See module docstring above.
+            try:
+                minx, miny, maxx, maxy = g.extent
+            except Exception:  # pragma: no cover — defensive
+                minx = miny = maxx = maxy = None
+            if minx is not None:
+                for lat, lon in (
+                    (miny, minx), (miny, maxx),
+                    (maxy, minx), (maxy, maxx),
+                ):
+                    if len(points) >= max_points:
+                        return
+                    _add(lat, lon)
             try:
                 pt = g.point_on_surface
             except Exception:  # pragma: no cover — defensive
                 pt = g.centroid
-            points.append((pt.y, pt.x))
+            if len(points) < max_points:
+                _add(pt.y, pt.x)
 
     _walk(geom)
     return points[:max_points]