Upgrade the plugin and test

Author: avdata99

ckanext/datastore/tests/sample_dump_plugin.py

@@ -8,17 +8,18 @@
 ALWAYS_INVALID_REASON = "Always invalid format"
 
 
-def invalid_validate(resource_id):
+def invalid_validate(context):
     """Validator that always rejects the format.
 
     Used by tests to confirm that a ``validate`` callable on a
     registered format flows through to the helper, the blueprint, and
-    the resource-read template.
+    the resource-read template. Receives the dump context dict (see
+    :class:`~ckanext.datastore.interfaces.IDatastoreDump`).
     """
     return ALWAYS_INVALID_REASON
 
 
-def valid_validate(resource_id):
+def valid_validate(context):
     """Validator that always approves.
 
     Used by tests to cover the branch where a format declares a
@@ -38,6 +39,10 @@ class SampleDumpPlugin(p.SingletonPlugin):
       (covers the "validator returns reason" path).
     * ``passes`` — a format whose ``validate`` callable always
       approves (covers the "validator returns None" path).
+    * ``capped`` — a format with declarative ``max_rows``/``max_columns``
+      and no ``validate`` callable (covers the framework-evaluated path).
+      ``max_rows`` is 1 so any resource with more than one (post-filter)
+      row trips it.
     * Removes the built-in ``xml`` format via the ``None`` sentinel.
     """
 
@@ -71,5 +76,17 @@ def register_dump_formats(self):
                 "file_extension": "passes",
                 "validate": valid_validate,
             },
+            # Declarative limits evaluated by core (no validate callable).
+            # max_rows=1 lets tests trip the row limit with a 2-row
+            # resource and confirm a filter can bring it back under.
+            "capped": {
+                "label": "Capped",
+                "writer_factory": csv_writer,
+                "records_format": "csv",
+                "content_type": "application/x-capped; charset=utf-8",
+                "file_extension": "capped",
+                "max_rows": 1,
+                "max_columns": 5,
+            },
             "xml": None,
         }

ckanext/datastore/tests/test_dump.py

@@ -2,6 +2,7 @@
 
 import unittest.mock as mock
 import json
+import urllib.parse
 import pytest
 import ckan.plugins.toolkit as toolkit
 import ckan.tests.helpers as helpers
@@ -1114,3 +1115,95 @@ def test_resource_read_page_renders_disabled_dropdown_item(self, app):
         # And the built-in CSV format must still render as an active
         # dropdown-item link (not disabled).
         assert 'class="dropdown-item"' in body
+
+
+@pytest.mark.ckan_config("ckan.plugins", "datastore sample_dump_plugin")
+@pytest.mark.usefixtures("clean_datastore", "with_plugins")
+class TestDatastoreDumpDeclarativeLimits(object):
+    """Tests for declarative ``max_rows``/``max_columns`` controls.
+
+    Uses the ``capped`` format from ``sample_dump_plugin`` (``max_rows``
+    is 1, ``max_columns`` is 5, no ``validate`` callable). These exercise
+    the framework-evaluated availability path and, crucially, that the
+    decision is made against the *filtered* export scope rather than the
+    resource totals.
+    """
+
+    def _make_resource(self, records):
+        resource = factories.Resource(url_type="datastore")
+        helpers.call_action(
+            "datastore_create",
+            resource_id=resource["id"],
+            records=records,
+        )
+        return resource
+
+    def test_helper_marks_over_row_limit_unavailable(self):
+        # 2 rows > max_rows=1 -> capped is unavailable with a
+        # framework-generated reason mentioning rows.
+        resource = self._make_resource(
+            [{"book": "annakarenina"}, {"book": "warandpeace"}])
+
+        formats = toolkit.h.datastore_dump_formats(
+            resource_id=resource["id"])
+        by_name = {f["name"]: f for f in formats}
+
+        assert by_name["capped"]["available"] is False
+        assert "rows" in by_name["capped"]["reason"].lower()
+        # Formats without controls stay available.
+        assert by_name["csv"]["available"] is True
+
+    def test_helper_within_limit_available(self):
+        # 1 row == max_rows=1 (rejected only when strictly greater).
+        resource = self._make_resource([{"book": "annakarenina"}])
+
+        formats = toolkit.h.datastore_dump_formats(
+            resource_id=resource["id"])
+        by_name = {f["name"]: f for f in formats}
+
+        assert by_name["capped"]["available"] is True
+        assert by_name["capped"]["reason"] is None
+
+    def test_dump_endpoint_rejects_over_limit_with_400(self, app):
+        resource = self._make_resource(
+            [{"book": "annakarenina"}, {"book": "warandpeace"}])
+
+        response = app.get(
+            f"/datastore/dump/{resource['id']}?format=capped",
+            status=400,
+        )
+        assert "rows" in response.get_data(as_text=True).lower()
+
+    def test_dump_endpoint_filtered_under_limit_succeeds(self, app):
+        # The resource has 2 rows (over the cap), but a filter reduces
+        # the export to a single row, so the capped format becomes
+        # available and the download succeeds. This is the case a
+        # resource-id-only validator would get wrong.
+        resource = self._make_resource(
+            [{"book": "annakarenina"}, {"book": "warandpeace"}])
+
+        qs = urllib.parse.urlencode({
+            "format": "capped",
+            "filters": json.dumps({"book": "annakarenina"}),
+        })
+        response = app.get(f"/datastore/dump/{resource['id']}?{qs}")
+
+        assert response.status_code == 200
+        assert "annakarenina" in response.get_data(as_text=True)
+        assert "warandpeace" not in response.get_data(as_text=True)
+
+    def test_helper_filtered_scope_marks_available(self):
+        # Same as above but through the helper: passing search_params
+        # that filter to one row flips capped from unavailable to
+        # available.
+        resource = self._make_resource(
+            [{"book": "annakarenina"}, {"book": "warandpeace"}])
+
+        formats = toolkit.h.datastore_dump_formats(
+            resource_id=resource["id"],
+            search_params={"filters": {"book": "annakarenina"}},
+        )
+        by_name = {f["name"]: f for f in formats}
+
+        assert by_name["capped"]["available"] is True
+        assert by_name["capped"]["reason"] is None