diff --git a/mediacloud/api.py b/mediacloud/api.py index abb42de..a8443d9 100644 --- a/mediacloud/api.py +++ b/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() diff --git a/mediacloud/mgmt.py b/mediacloud/mgmt.py new file mode 100644 index 0000000..dcd4da4 --- /dev/null +++ b/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 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", + ) diff --git a/mediacloud/test/api_mgmt_test.py b/mediacloud/test/api_mgmt_test.py new file mode 100644 index 0000000..f38c741 --- /dev/null +++ b/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)