Directory Management API tweak + test

mediacloud/api.py

@@ -60,9 +60,13 @@ def _query(self, endpoint: str, params: Optional[Dict] = None, method: str = 'GE
             r = self._session.get(endpoint_url, params=params, timeout=self.TIMEOUT_SECS)
         elif method == 'POST':
             r = self._session.post(endpoint_url, json=params, timeout=self.TIMEOUT_SECS)
+        elif method == "DELETE":
+            return self._session.delete(endpoint_url, params=params, timeout=self.TIMEOUT_SECS)
+        elif method == "PATCH":
+            return self._session.patch(endpoint_url, json=params, timeout=self.TIMEOUT_SECS)
         else:
             raise RuntimeError(f"Unsupported method of '{method}'")
-        if r.status_code != 200:
+        if r.status_code // 100 != 2:  # create operations return 201
             raise mediacloud.error.APIResponseError(r, params, r.json())
 
         return r.json()

mediacloud/mgmt.py

@@ -0,0 +1,135 @@
+"""
+Directory Management API
+
+Nod to Monty Python
+It's not Camelot, only a model...
+
+update/patch methods not yet tested!!!
+
+ALL arguments are keyword only, for safety!
+"""
+from typing import TypeAlias
+
+from mediacloud.api import BaseApi
+
+_EMPTY = object()
+
+_Params: TypeAlias = dict
+
+
+class DirectoryManagementApi(BaseApi):  # XXX maybe extend DirectoryApi???
+    """
+    Class for Directory Management
+    """
+
+    def _params(self, what: str, kws: dict, params: list[str]) -> _Params:
+        """
+        helper for _{collection,source}_params helpers
+        """
+        ret: _Params = {}
+        kwcopy = kws.copy()
+        for p in params:
+            v = kwcopy.pop(p, _EMPTY)
+            if v is not _EMPTY:
+                ret[p] = v
+        if kwcopy:
+            extra = ",".join(kwcopy.keys())
+            raise ValueError(f"Unknown {what} params {extra}")
+        return ret
+
+    ################ CollectionsViewSet
+
+    def _collection_params(self, kws: dict) -> _Params:
+        """
+        helper for collection_{create,update}
+        """
+        return self._params("collection", kws,
+                            ['name', 'notes', 'public', 'featured', 'managed', 'monitored'])
+
+    def collection_create(self, **kwargs) -> dict:
+        params = self._collection_params(kwargs)
+        if 'name' not in params or not params['name']:
+            raise ValueError("collection_create must have 'name'")
+        return self._query('sources/collections/', params, "POST")
+
+    def collection_copy(self, *, collection_id: int, name: str) -> dict:
+        # name defaults to "original name (copy)", but be fussy:
+        if not name:
+            raise ValueError("collection_copy must have 'name'")
+        params = {'collection_id': collection_id, 'name': name}
+        return self._query('sources/collections/copy-collection/', params, "POST")
+
+    def collection_update(self, *, collection_id: int, **kwargs) -> dict:
+        params = self._collection_params(kwargs)
+        if not params:
+            raise ValueError("no parameters for collection_update?")
+        return self._query(f'sources/collections/{collection_id}/', params, "PATCH")
+
+    # for testing/cleanup:
+    def collection_delete(self, collection_id: int) -> dict:
+        return self._query(f'sources/collections/{collection_id}/', None, "DELETE")
+
+    def collection_source_list(self, *, collection_id: int) -> list[dict]:
+        """
+        returns list of source objects for a collection
+        """
+        # XXX should do pagination!!!!!!!
+        ret = self._query(f'sources/sources/?collection_id={collection_id}', None, "GET")
+        # returns {'count': n, 'next': ..., 'previous': ..., 'results': [{...}, ...]}
+        return ret['results']
+
+    ################ SourcesViewSet
+
+    def _source_params(self, kws: dict) -> _Params:
+        """
+        helper for source_{create,update}
+        """
+        return self._params("source", kws,
+                            ['name', 'label', 'homepage', 'platform',
+                             'url_search_string', 'notes', 'media_type',
+                             'pub_state', 'pub_country', 'primary_language'])
+
+    def source_create(self, **kwargs) -> dict:
+        params = self._source_params(kwargs)
+        for p in ['name', 'homepage']:
+            if p not in params:
+                raise ValueError(f"source_create requires '{p}'")
+        return self._query('sources/sources/', params, "POST")
+
+    def source_update(self, *, source_id: int, **kwargs) -> dict:
+        params = self._source_params(kwargs)
+        if not params:
+            raise ValueError("no parameters for source_update?")
+        params['id'] = source_id
+        # currently causes Internal Server Error and <Response [500]> returned with body:
+        # {"detail":"{'homepage': [ErrorDetail(string='This field is required.', code='required')]}"}
+        # which isn't what I expect from "PATCH"!!!
+        return self._query(f'sources/sources/{source_id}/', params, "PATCH")
+
+    # for testing/cleanup:
+    def source_delete(self, source_id: int) -> dict:
+        return self._query(f'sources/sources/{source_id}/', None, "DELETE")
+
+    ################ SourcesCollectionsViewSet
+
+    def source_collection_list(self, *, source_id: int) -> list[dict]:
+        """
+        returns list of collection objects for a source
+        """
+        ret = self._query(f'sources/sources-collections/{source_id}/', None, "GET")
+        return ret['collections']
+
+    # mcweb sourcesCollectionsApi.js calls this createSourceCollectionAssociation
+    def source_collection_create(self, *, source_id: int, collection_id: int) -> dict:
+        params = {'source_id': source_id, 'collection_id': collection_id}
+        return self._query('sources/sources-collections/', params, "POST")
+
+    # mcweb sourcesCollectionsApi.js calls this deleteSourceCollectionAssociation
+    # XXX endpoint seems to take collection=bool query parameter??
+    # (if not set to true, expects collection_id parameter??)
+    def source_collection_delete(self, *, source_id: int, collection_id: int) -> dict:
+        return self._query(
+            f'sources/sources-collections/{source_id}/?collection_id={collection_id}',
+            None,
+            "DELETE",
+        )

mediacloud/test/api_mgmt_test.py

@@ -0,0 +1,108 @@
+"""
+Minimal tests for DirectoryManagementApi (mediacloud.mgmt).
+
+Live tests follow the same two-token pattern as api_search_test: MC_API_TOKEN (normal) and
+MC_API_ADMIN_TOKEN (privileged). Staging integration requires both to be set.
+"""
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from mediacloud.error import APIResponseError
+from mediacloud.mgmt import DirectoryManagementApi
+
+# Directory management integration checks against Tarbell staging (see mediacloud.org deployment).
+MCWEB_STAGING_API_BASE = "https://mcweb-staging.tarbell.mediacloud.org/api/"
+
+
+class DirectoryManagementApiUnitTest(unittest.TestCase):
+    """Pure unit tests (no network)."""
+
+    def setUp(self):
+        self._api = DirectoryManagementApi("test-token")
+
+    def test_source_collection_delete_uses_http_delete(self):
+        with patch.object(self._api, "_query", autospec=True) as mock_query:
+            mock_query.return_value = MagicMock(status_code=204)
+            self._api.source_collection_delete(source_id=42, collection_id=99)
+        mock_query.assert_called_once_with(
+            "sources/sources-collections/42/?collection_id=99",
+            None,
+            "DELETE",
+        )
+
+    def test_collection_create_requires_name(self):
+        with self.assertRaises(ValueError):
+            self._api.collection_create(notes="only notes")
+
+    def test_collection_create_rejects_unknown_kwargs(self):
+        with self.assertRaises(ValueError) as ctx:
+            self._api.collection_create(name="x", bogus_field=1)
+        self.assertIn("Unknown collection params", str(ctx.exception))
+
+    def test_collection_update_rejects_unknown_kwargs(self):
+        with self.assertRaises(ValueError):
+            self._api.collection_update(collection_id=1, name="ok", extra=2)
+
+
+@pytest.mark.skipif(not os.getenv("MC_API_TOKEN"), reason="MC_API_TOKEN not set")
+class DirectoryManagementApiLiveTest(unittest.TestCase):
+    """Smoke against the configured API (default: production search host)."""
+
+    def setUp(self):
+        self._mgmt = DirectoryManagementApi(os.environ["MC_API_TOKEN"])
+
+    def test_user_profile(self):
+        profile = self._mgmt.user_profile()
+        self.assertIsInstance(profile, dict)
+        self.assertGreater(len(profile), 0)
+
+
+@pytest.mark.skipif(
+    not os.getenv("MC_API_TOKEN") or not os.getenv("MC_API_ADMIN_TOKEN"),
+    reason="MC_API_TOKEN and MC_API_ADMIN_TOKEN required (same as api_search_test)",
+)
+class DirectoryManagementStagingIntegrationTest(unittest.TestCase):
+    """
+    Live checks against https://mcweb-staging.tarbell.mediacloud.org/ (staging API).
+
+    Uses MC_API_TOKEN for the 403 assertion (normal key must not be allowed to create
+    collections). Uses MC_API_ADMIN_TOKEN for a parallel read sanity check on staging.
+    """
+
+    def setUp(self):
+        self._mc_api_key = os.environ["MC_API_TOKEN"]
+        self._mc_api_admin_key = os.environ["MC_API_ADMIN_TOKEN"]
+        if self._mc_api_key == self._mc_api_admin_key:
+            self.skipTest("MC_API_TOKEN and MC_API_ADMIN_TOKEN must differ for staging mgmt tests")
+
+        self._prev_base = DirectoryManagementApi.BASE_API_URL
+        DirectoryManagementApi.BASE_API_URL = MCWEB_STAGING_API_BASE
+        self.addCleanup(self._restore_base_url)
+        self._mgmt = DirectoryManagementApi(self._mc_api_key)
+        self._admin_mgmt = DirectoryManagementApi(self._mc_api_admin_key)
+
+    def _restore_base_url(self):
+        DirectoryManagementApi.BASE_API_URL = self._prev_base
+
+    def test_staging_user_profile_reaches_server(self):
+        """Sanity: normal token is accepted on staging before we assert on management errors."""
+        profile = self._mgmt.user_profile()
+        self.assertIsInstance(profile, dict)
+        self.assertGreater(len(profile), 0)
+
+    def test_staging_admin_user_profile_reaches_server(self):
+        """Sanity: admin token is accepted on staging (same pattern as BaseSearchTest)."""
+        profile = self._admin_mgmt.user_profile()
+        self.assertIsInstance(profile, dict)
+        self.assertGreater(len(profile), 0)
+
+    def test_non_admin_collection_create_is_forbidden_on_staging(self):
+        with self.assertRaises(APIResponseError) as ctx:
+            self._mgmt.collection_create(
+                name="api-client integration (should not be created)",
+                notes="DirectoryManagementStagingIntegrationTest",
+            )
+        self.assertEqual(ctx.exception.response.status_code, 403)