Merge branch 'wip/update-single-value' into wip/kuhaba

.github/workflows/pr-tests.yml

@@ -29,7 +29,7 @@ jobs:
         run: poetry install
 
       - name: Formatting with ruff
-        run: poetry run ruff format .
+        run: poetry run ruff format --check .
 
       - name: Linting with ruff
         run: poetry run ruff check .
@@ -38,6 +38,6 @@ jobs:
         run: poetry run mypy .
 
       - name: unittests
-        run: poetry run python -m unittest discover tests
+        run: poetry run pytest -s tests
 
 

dsp_permissions_scripts/models/errors.py

@@ -17,3 +17,18 @@ def __str__(self) -> str:
 @dataclass
 class PermissionsAlreadyUpToDate(Exception):
     message: str = "The submitted permissions are the same as the current ones"
+
+
+@dataclass
+class SpecifiedPropsEmptyError(ValueError):
+    message: str = "specified_props must not be empty if retrieve_values is 'specified_props'"
+
+
+@dataclass
+class SpecifiedPropsNotEmptyError(ValueError):
+    message: str = "specified_props must be empty if retrieve_values is not 'specified_props'"
+
+
+@dataclass
+class OapRetrieveConfigEmptyError(ValueError):
+    message: str = "retrieve_resources cannot be False if retrieve_values is 'none'"

dsp_permissions_scripts/models/group.py

@@ -8,7 +8,6 @@
 
 
 class Group(BaseModel):
-
     model_config = ConfigDict(frozen=True)
 
     val: str

dsp_permissions_scripts/oap/oap_get.py

@@ -69,19 +69,22 @@ def _get_next_page(
     if "@graph" in result:
         oaps = []
         for r in result["@graph"]:
-            oaps.append(_get_oap_of_one_resource(r, oap_config))
+            if oap := _get_oap_of_one_resource(r, oap_config):
+                oaps.append(oap)
         return True, oaps
 
     # result contains only 1 resource: return it, then stop (there will be no more resources)
     if "@id" in result:
-        oaps = [_get_oap_of_one_resource(result, oap_config)]
+        oaps = []
+        if oap := _get_oap_of_one_resource(result, oap_config):
+            oaps.append(oap)
         return False, oaps
 
     # there are no more resources
     return False, []
 
 
-def _get_oap_of_one_resource(r: dict[str, Any], oap_config: OapRetrieveConfig) -> Oap:
+def _get_oap_of_one_resource(r: dict[str, Any], oap_config: OapRetrieveConfig) -> Oap | None:
     if oap_config.retrieve_resources:
         scope = create_scope_from_string(r["knora-api:hasPermissions"])
         resource_oap = ResourceOap(scope=scope, resource_iri=r["@id"])
@@ -95,7 +98,10 @@ def _get_oap_of_one_resource(r: dict[str, Any], oap_config: OapRetrieveConfig) -
     else:
         value_oaps = _get_value_oaps(r, oap_config.specified_props)
 
-    return Oap(resource_oap=resource_oap, value_oaps=value_oaps)
+    if resource_oap or value_oaps:
+        return Oap(resource_oap=resource_oap, value_oaps=value_oaps)
+    else:
+        return None
 
 
 def _get_value_oaps(resource: dict[str, Any], restrict_to_props: list[str] | None = None) -> list[ValueOap]:

dsp_permissions_scripts/oap/oap_model.py

@@ -6,6 +6,9 @@
 from pydantic import ConfigDict
 from pydantic import model_validator
 
+from dsp_permissions_scripts.models.errors import OapRetrieveConfigEmptyError
+from dsp_permissions_scripts.models.errors import SpecifiedPropsEmptyError
+from dsp_permissions_scripts.models.errors import SpecifiedPropsNotEmptyError
 from dsp_permissions_scripts.models.scope import PermissionScope
 
 
@@ -19,6 +22,12 @@ class Oap(BaseModel):
     resource_oap: ResourceOap | None
     value_oaps: list[ValueOap]
 
+    @model_validator(mode="after")
+    def check_consistency(self) -> Oap:
+        if not self.resource_oap and not self.value_oaps:
+            raise ValueError("An OAP must have at least one resource_oap or one value_oap")
+        return self
+
 
 class ResourceOap(BaseModel):
     """Model representing an object access permission of a resource"""
@@ -46,17 +55,22 @@ class ValueOap(BaseModel):
 
 
 class OapRetrieveConfig(BaseModel):
-
     model_config = ConfigDict(frozen=True)
 
     retrieve_resources: bool = True
     retrieve_values: Literal["all", "specified_props", "none"] = "none"
     specified_props: list[str] = []
 
     @model_validator(mode="after")
-    def check_consistency(self) -> OapRetrieveConfig:
+    def check_specified_props(self) -> OapRetrieveConfig:
         if self.retrieve_values == "specified_props" and not self.specified_props:
-            raise ValueError("specified_props must not be empty if retrieve_values is 'specified_props'")
+            raise SpecifiedPropsEmptyError()
         if self.retrieve_values != "specified_props" and self.specified_props:
-            raise ValueError("specified_props must be empty if retrieve_values is not 'specified_props'")
+            raise SpecifiedPropsNotEmptyError()
+        return self
+
+    @model_validator(mode="after")
+    def check_config_empty(self) -> OapRetrieveConfig:
+        if not self.retrieve_resources and self.retrieve_values == "none":
+            raise OapRetrieveConfigEmptyError()
         return self

dsp_permissions_scripts/oap/oap_set.py

@@ -116,7 +116,7 @@ def _update_batch(batch: tuple[Oap, ...], dsp_client: DspClient) -> list[str]:
     return failed_iris
 
 
-def _write_failed_res_iris_to_file(
+def _write_failed_iris_to_file(
     failed_iris: list[str],
     shortcode: str,
     host: str,
@@ -128,7 +128,7 @@ def _write_failed_res_iris_to_file(
 
 
 def _launch_thread_pool(oaps: list[Oap], nthreads: int, dsp_client: DspClient) -> list[str]:
-    all_failed_iris = []
+    all_failed_iris: list[str] = []
     with ThreadPoolExecutor(max_workers=nthreads) as pool:
         jobs = [pool.submit(_update_batch, batch, dsp_client) for batch in itertools.batched(oaps, 100)]
         for result in as_completed(jobs):
@@ -157,7 +157,7 @@ def apply_updated_oaps_on_server(
     if failed_iris:
         timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
         filename = f"FAILED_RESOURCES_AND_VALUES_{timestamp}.txt"
-        _write_failed_res_iris_to_file(
+        _write_failed_iris_to_file(
             failed_iris=sorted(failed_iris),
             shortcode=shortcode,
             host=host,

pyproject.toml

@@ -18,6 +18,8 @@ types-requests = "^2.31.0.20240406"
 python-dotenv = "^1.0.1"
 ruff = "^0.3.7"
 pre-commit = "^3.7.0"
+pytest = "^8.1.1"
+pytest-unordered = "^0.6.0"
 
 [build-system]
 build-backend = "poetry.core.masonry.api"
@@ -52,6 +54,14 @@ select = [
     "B023",     # flake8-bugbear: function-uses-loop-variable
     "FIX",      # flake8-fixme: checks for FIXME, TODO, XXX, etc.
 ]
+ignore = [
+    "ISC001",   # flake8-implicit-str-concat: single-line-implicit-string-concatenation  # incompatible with the formatter
+]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*" = [
+    "S101",     # flake8-bandit: use of assert
+]
 
 [tool.ruff.lint.pydocstyle]
 convention = "google"

tests/test_ap.py

@@ -1,13 +1,11 @@
-import unittest
+import pytest
 
 from dsp_permissions_scripts.ap.ap_model import Ap
 from dsp_permissions_scripts.ap.ap_model import ApValue
 from dsp_permissions_scripts.models import group
 
-# ruff: noqa: PT009 (pytest-unittest-assertion) (remove this line when pytest is used instead of unittest)
-# ruff: noqa: PT027 (pytest-unittest-raises-assertion) (remove this line when pytest is used instead of unittest)
 
-class TestAp(unittest.TestCase):
+class TestAp:
     ap = Ap(
         forGroup=group.UNKNOWN_USER,
         forProject="http://rdfh.ch/projects/0001",
@@ -17,16 +15,20 @@ class TestAp(unittest.TestCase):
 
     def test_add_permission(self) -> None:
         self.ap.add_permission(ApValue.ProjectAdminRightsAllPermission)
-        self.assertIn(ApValue.ProjectAdminRightsAllPermission, self.ap.hasPermissions)
+        assert ApValue.ProjectAdminRightsAllPermission in self.ap.hasPermissions
 
     def test_add_permission_already_exists(self) -> None:
-        with self.assertRaises(ValueError):
+        with pytest.raises(ValueError):  # noqa: PT011 (exception too broad)
             self.ap.add_permission(ApValue.ProjectAdminGroupAllPermission)
 
     def test_remove_permission(self) -> None:
         self.ap.remove_permission(ApValue.ProjectAdminGroupAllPermission)
-        self.assertNotIn(ApValue.ProjectAdminGroupAllPermission, self.ap.hasPermissions)
+        assert ApValue.ProjectAdminGroupAllPermission not in self.ap.hasPermissions
 
     def test_remove_permission_not_exists(self) -> None:
-        with self.assertRaises(ValueError):
+        with pytest.raises(ValueError):  # noqa: PT011 (exception too broad)
             self.ap.remove_permission(ApValue.ProjectAdminAllPermission)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])