enh: implement and test get_datasets_user_owned for CachedAPIInterrogator

Author: paulmueller

dcoraid/dbmodel/db_api_cached.py

@@ -32,7 +32,9 @@ def __init__(self, api, cache_location):
         else:
             user_data = None
 
-        self._mc = MetaCache(self.cache_location)
+        self._mc = MetaCache(directory=self.cache_location,
+                             user_id=self.api.user_id,
+                             )
         self._mc_timestamp_path = self.cache_location / "cache_timestamp"
         self._mc_timestamp_path.touch()
         self._mc_version_path = self.cache_location / "cache_version"
@@ -79,8 +81,9 @@ def get_datasets_user_following(self):
 
     def get_datasets_user_owned(self):
         """Return datasets the user created"""
-        # TODO: Use datasets in self._mc
-        return self.ai.get_datasets_user_owned()
+        own_list = self._mc.datasets_user_owned
+        ds_list = self._mc.datasets
+        return [ds for (ds, byuser) in zip(ds_list, own_list) if byuser]
 
     def get_datasets_user_shared(self):
         """Return datasets shared with the user"""
@@ -134,6 +137,6 @@ def update(self, reset=False, abort_event=None):
                 break
             self._mc.upsert_dataset(ds_dict)
         else:
-            # Only update teh local timestamp if we actually did
+            # Only update the local timestamp if we actually did
             # update the local database.
             self.local_timestamp = new_timestamp

dcoraid/dbmodel/meta_cache.py

@@ -20,6 +20,7 @@ class MetaCache:
     """
     def __init__(self,
                  directory: str | pathlib.Path,
+                 user_id: str = None,
                  circle_ids: list[str] = None,
                  ) -> None:
         """
@@ -28,24 +29,33 @@ def __init__(self,
 
         Parameters
         ----------
-        directory : str | pathlib.Path
+        directory: str | pathlib.Path
             Path to the folder that will hold the ``circle_<org_id>.db`` files.
             The folder is created automatically if it does not exist.
-        circle_ids : list[str]
+        user_id: str
+            The ID of the user operating the database.
+        circle_ids: list[str]
             List of circle IDs that should be taken into consideration.
             If set to None (default), all databases in the `directory`
             are loaded.
         """
         self.base_dir = pathlib.Path(directory).expanduser().resolve()
         self.base_dir.mkdir(parents=True, exist_ok=True)
+        self.user_id = user_id
 
-        # The registry is a dictionary with circle IDs and a list of
-        # dataset IDs as values.
+        # The organization registry is a dictionary with circle IDs and
+        # a list of dataset IDs as values.
         self._registry_org = {}
 
+        # List of booleans indicating whether dataset was created by the user
+        self.datasets_user_owned = []
+
         # Dictionary of databases for persistent storage
         self._databases = {}
 
+        # List of dataset dictionaries
+        self.datasets = []
+
         # Search blob array
         self._srt_blobs = None
 
@@ -249,6 +259,12 @@ def _upsert_dataset_insert(self, ds_dict):
                 db_name=self.base_dir / f"circle_{org_id}.db")
         self._databases[org_id][ds_id] = ds_dict
 
+        # user's dataset list
+        self.datasets_user_owned.insert(
+            new_idx,
+            ds_dict["creator_user_id"] == self.user_id
+            )
+
     def _upsert_dataset_update(self, ds_dict):
         """Update an existing dataset
 

tests/test_dbmodel_api.py

@@ -38,6 +38,14 @@ def test_get_collections():
     assert "figshare-collection" in collections
 
 
+def test_get_datasets_user_owned(tmp_path):
+    ds_dict = common.make_dataset_for_download()
+    api = common.get_api()
+    db = db_api.APIInterrogator(api=api)
+    ds_list = db.get_datasets_user_owned()
+    assert ds_dict in ds_list
+
+
 def test_get_users_anonymous():
     api = CKANAPI(server=common.SERVER)  # anonymous access
     db = db_api.APIInterrogator(api=api)

tests/test_dbmodel_api_cache.py

@@ -0,0 +1,84 @@
+import pathlib
+
+import pytest
+
+from dcoraid.api import CKANAPI, errors
+from dcoraid.dbmodel import CachedAPIInterrogator
+
+from . import common
+
+dpath = pathlib.Path(__file__).parent / "data" / "calibration_beads_47.rtdc"
+
+HAS_FIGSHARE_ACCESS = common.get_test_defaults()["user"] == "dcoraid"
+
+
+@pytest.mark.skipif(not HAS_FIGSHARE_ACCESS,
+                    reason="No access to figshare-import circle")
+def test_get_circles(tmp_path):
+    api = common.get_api()
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    circles = db.get_circles()
+    defaults = common.get_test_defaults()
+    assert defaults["circle"] in circles
+    # requires that the "dcoraid" user is in the figshare-import circle
+    assert "figshare-import" in circles
+
+
+@pytest.mark.skipif(not HAS_FIGSHARE_ACCESS,
+                    reason="No access to figshare-import circle")
+def test_get_collections(tmp_path):
+    api = common.get_api()
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    collections = db.get_collections()
+    defaults = common.get_test_defaults()
+    assert defaults["collection"] in collections
+    # requires that the "dcoraid" user is in figshare-collection collection
+    assert "figshare-collection" in collections
+
+
+def test_get_datasets_user_owned(tmp_path):
+    ds_dict = common.make_dataset_for_download()
+    api = common.get_api()
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    db.update()
+    ds_list = db.get_datasets_user_owned()
+    assert ds_dict in ds_list
+
+
+def test_get_users_anonymous(tmp_path):
+    api = CKANAPI(server=common.SERVER)  # anonymous access
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    with pytest.raises(errors.APIAuthorizationError, match="Access denied"):
+        db.get_users()
+
+
+def test_public_api_interrogator(tmp_path):
+    api = common.get_api()
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    defaults = common.get_test_defaults()
+    assert defaults["circle"] in db.get_circles()
+    assert defaults["collection"] in db.get_collections()
+    assert defaults["user"] in db.get_users()
+
+
+def test_user_data(tmp_path):
+    """Test the user information"""
+    api = common.get_api()
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    data = db.user_data
+    defaults = common.get_test_defaults()
+    assert data["fullname"] == defaults["user_name"], "fullname not correct"
+
+
+@pytest.mark.skipif(not HAS_FIGSHARE_ACCESS,
+                    reason="No access to figshare-import circle")
+def test_get_datasets_user_shared_figshare(tmp_path):
+    # The figshare circle must have the testing user as a member
+    api = common.get_api()
+    db = CachedAPIInterrogator(cache_location=tmp_path, api=api)
+    datasets = db.get_datasets_user_shared()
+    for dd in datasets:
+        if dd["id"] == "89bf2177-ffeb-9893-83cc-b619fc2f6663":
+            break
+    else:
+        assert False, "Search did not return figshare-7771184-v2!"