Extract group (#65)

* Added extraction of team from the GROUP_CONTEXT enviornment variable

* Changing env variable name

* Added tests

* Handled the error for when no DAPLA_GROUP_CONTEXT is found

* precommit

* Update src/dapla_metadata/datasets/user_info.py

Co-authored-by: Miles Mason Winther <42948872+mmwinther@users.noreply.github.com>

* Updated tests

* Added doctest

* Updated error handling

* Changed error

* Changed test

* Showing owner when opening dataset

* Moved function to utils

* Updating version

* Correcting typo

---------

Co-authored-by: Jorgen-5 <rlj@ssb.no>
Co-authored-by: Miles Mason Winther <42948872+mmwinther@users.noreply.github.com>

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dapla-toolbelt-metadata"
-version = "0.2.6"
+version = "0.3.0"
 description = "Dapla Toolbelt Metadata"
 authors = ["Team Metadata <metadata@ssb.no>"]
 license = "MIT"

src/dapla_metadata/datasets/config.py

@@ -41,7 +41,7 @@ def _load_dotenv_file() -> None:
 def _get_config_item(item: str) -> str | None:
     """Get a config item. Makes sure all access is logged."""
     _load_dotenv_file()
-    value = os.getenv(item)
+    value = os.environ.get(item)
     logger.debug("Config accessed. %s", item)
     return value
 
@@ -78,3 +78,8 @@ def get_dapla_service() -> DaplaService | None:
 def get_oidc_token() -> str | None:
     """Get the JWT token from the environment."""
     return _get_config_item("OIDC_TOKEN")
+
+
+def get_group_context() -> str | None:
+    """Get the JWT token from the environment."""
+    return _get_config_item("DAPLA_GROUP_CONTEXT")

src/dapla_metadata/datasets/core.py

@@ -42,6 +42,7 @@
     num_obligatory_variables_fields_completed,
 )
 from dapla_metadata.datasets.utility.utils import override_dataset_fields
+from dapla_metadata.datasets.utility.utils import set_dataset_owner
 from dapla_metadata.datasets.utility.utils import set_default_values_dataset
 from dapla_metadata.datasets.utility.utils import set_default_values_variables
 
@@ -189,6 +190,7 @@ def _extract_metadata_from_files(self) -> None:
             self._set_metadata(existing_metadata or extracted_metadata)
         set_default_values_variables(self.variables)
         set_default_values_dataset(self.dataset)
+        set_dataset_owner(self.dataset)
         self._create_variables_lookup()
 
     def _get_existing_file_path(

src/dapla_metadata/datasets/user_info.py

@@ -86,3 +86,30 @@ def get_user_info_for_current_platform() -> UserInfo:
             "Was not possible to retrieve user information! Some fields may not be set.",
         )
         return UnknownUserInfo()
+
+
+def get_owner() -> str:
+    """Returns the owner read from the GROUP_CONTEXT environment variable."""
+    if group := config.get_group_context():
+        return parse_team_name(group)
+    msg = "DAPLA_GROUP_CONTEXT environment variable not found"
+    raise OSError(msg)
+
+
+def parse_team_name(group: str) -> str:
+    """Parses the group to get the current team.
+
+    >>> parse_team_name(dapla-metadata-developers)
+    (dapla-metadata)
+
+    >>> parse_team_name(dapla-metadata-data-admins)
+    (dapla-metadata)
+
+    >>> parse_team_name(dapla-metadata)
+    (dapla)
+
+    >>> parse_team_name(dapla-metadata-not-real-name)
+    (dapla-metadata-not-real)
+    """
+    parts = group.split("-")
+    return "-".join(parts[:-2] if group.endswith("data-admins") else parts[:-1])

src/dapla_metadata/datasets/utility/utils.py

@@ -14,6 +14,7 @@
 from datadoc_model.model import DataSetState
 from datadoc_model.model import VariableRole
 
+from dapla_metadata.datasets import user_info
 from dapla_metadata.datasets.utility.constants import (
     DATASET_FIELDS_FROM_EXISTING_METADATA,
 )
@@ -139,6 +140,18 @@ def set_default_values_dataset(dataset: model.Dataset) -> None:
         dataset.contains_personal_data = False
 
 
+def set_dataset_owner(dataset: model.Dataset) -> None:
+    """Sets the owner of the dataset from the DAPLA_GROUP_CONTEXT enviornment variable.
+
+    Args:
+        dataset: The dataset object to set default values on.
+    """
+    try:
+        dataset.owner = user_info.get_owner()
+    except OSError:
+        logger.exception("Failed to find environment variable DAPLA_GROUP_CONTEXT")
+
+
 def set_variables_inherit_from_dataset(
     dataset: model.Dataset,
     variables: list,

tests/datasets/constants.py

@@ -52,6 +52,8 @@
 
 DAPLA_REGION = "DAPLA_REGION"
 
+DAPLA_GROUP_CONTEXT = "DAPLA_GROUP_CONTEXT"
+
 DAPLA_SERVICE = "DAPLA_SERVICE"
 
 DATADOC_METADATA_MODULE = "dapla_metadata.datasets"

tests/datasets/test_user_info.py

@@ -12,6 +12,7 @@
 from dapla_metadata.datasets.user_info import UserInfo
 from dapla_metadata.datasets.utility.enums import DaplaRegion
 from dapla_metadata.datasets.utility.enums import DaplaService
+from tests.datasets.constants import DAPLA_GROUP_CONTEXT
 from tests.datasets.constants import DAPLA_REGION
 from tests.datasets.constants import DAPLA_SERVICE
 from tests.datasets.constants import JUPYTERHUB_USER
@@ -110,3 +111,56 @@ def test_dapla_lab_user_info_short_email_no_email_in_jwt(
 ):
     monkeypatch.setenv("OIDC_TOKEN", fake_jwt)
     assert DaplaLabUserInfo().short_email is None
+
+
+@pytest.mark.parametrize(
+    ("environment_variable_name", "environment_variable_value", "expected_team"),
+    [
+        (DAPLA_GROUP_CONTEXT, "dapla-metadata-developers", "dapla-metadata"),
+        (
+            DAPLA_GROUP_CONTEXT,
+            "dapla-metadata-data-admins",
+            "dapla-metadata",
+        ),
+        (DAPLA_GROUP_CONTEXT, "dapla-metadata-dev", "dapla-metadata"),
+        (DAPLA_GROUP_CONTEXT, "dapla-metadata", "dapla"),
+        (
+            DAPLA_GROUP_CONTEXT,
+            "dapla-metadata-not-a-real-group",
+            "dapla-metadata-not-a-real",
+        ),
+    ],
+)
+def test_get_owner(
+    monkeypatch: pytest.MonkeyPatch,
+    environment_variable_name: str,
+    environment_variable_value: str,
+    expected_team: str,
+):
+    if environment_variable_name:
+        monkeypatch.setenv(environment_variable_name, environment_variable_value)
+    assert user_info.get_owner() == expected_team
+
+
+@pytest.mark.parametrize(
+    ("environment_variable_name", "environment_variable_value"),
+    [
+        (None, None),
+        (DAPLA_GROUP_CONTEXT, ""),
+    ],
+)
+def test_get_owner_errors(
+    monkeypatch: pytest.MonkeyPatch,
+    environment_variable_name: str,
+    environment_variable_value: str,
+):
+    if environment_variable_name:
+        monkeypatch.setenv(environment_variable_name, environment_variable_value)
+
+    with pytest.raises(
+        OSError,
+        match="DAPLA_GROUP_CONTEXT environment variable not found",
+    ) as exc_info:  # Step 1: Expect an exception
+        user_info.get_owner()
+
+    assert str(exc_info.value) == "DAPLA_GROUP_CONTEXT environment variable not found"