Merge pull request #95 from itk-dev-rpa/feature/nova-group-caseworker

Feature/nova group caseworker

Author: ghbm-itk

changelog.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- kmd_nova: Added support for case worker groups/units.
+
 ### Fixed
 
 - Child folders in Graph are now found, even with more than 10 child folders to go through.

itk_dev_shared_components/kmd_nova/nova_cases.py

@@ -2,14 +2,13 @@
 to the KMD Nova api."""
 
 import uuid
-import base64
 import urllib.parse
 
 import requests
 
 from itk_dev_shared_components.kmd_nova.authentication import NovaAccess
-from itk_dev_shared_components.kmd_nova.nova_objects import NovaCase, CaseParty, JournalNote, Caseworker, Department
-from itk_dev_shared_components.kmd_nova.util import datetime_from_iso_string
+from itk_dev_shared_components.kmd_nova.nova_objects import NovaCase, CaseParty, Department
+from itk_dev_shared_components.kmd_nova.util import datetime_from_iso_string, extract_caseworker
 
 
 def get_case(case_uuid: str, nova_access: NovaAccess) -> NovaCase:
@@ -125,7 +124,7 @@ def _get_nova_cases(nova_access: NovaAccess, payload: dict) -> list[NovaCase]:
             kle_number = case_dict['caseClassification']['kleNumber']['code'],
             proceeding_facet = case_dict['caseClassification']['proceedingFacet']['code'],
             sensitivity = case_dict["sensitivity"]["sensitivity"],
-            caseworker = _extract_case_worker(case_dict),
+            caseworker = extract_caseworker(case_dict),
             security_unit=security_unit,
             responsible_department=responsible_department
         )
@@ -189,6 +188,11 @@ def _create_payload(*, case_uuid: str = None, identification: str = None, identi
                     "novaUserId": True,
                     "fullName": True,
                     "racfId": True
+                },
+                "losIdentity": {
+                    "novaUnitId": True,
+                    "fullName": True,
+                    "administrativeUnitId": True
                 }
             },
             "responsibleDepartment": {
@@ -235,29 +239,6 @@ def _extract_departments(case_dict: dict) -> tuple[Department, Department]:
     return security_unit, responsible_department
 
 
-def _extract_case_worker(case_dict: dict) -> Caseworker | None:
-    """Extract the case worker from a HTTP request response.
-    If the case worker is in a unexpected format, None is returned.
-
-    Args:
-        case_dict: The dictionary describing the case.
-
-    Returns:
-        A case worker object describing the case worker if any.
-    """
-    if 'caseworker' in case_dict:
-        try:
-            return Caseworker(
-                uuid = case_dict['caseworker']['kspIdentity']['novaUserId'],
-                name = case_dict['caseworker']['kspIdentity']['fullName'],
-                ident = case_dict['caseworker']['kspIdentity']['racfId']
-            )
-        except KeyError:
-            return None
-
-    return None
-
-
 def _extract_case_parties(case_dict: dict) -> list[CaseParty]:
     """Extract the case parties from a HTTP request response.
 
@@ -281,29 +262,6 @@ def _extract_case_parties(case_dict: dict) -> list[CaseParty]:
     return parties
 
 
-def _extract_journal_notes(case_dict: dict) -> list:
-    """Extract the journal notes from a HTTP request response.
-
-    Args:
-        case_dict: The dictionary describing the journal note.
-
-    Returns:
-        A journal note object describing the journal note.
-    """
-    notes = []
-    for note_dict in case_dict['journalNotes']['journalNotes']:
-        note = JournalNote(
-            uuid = note_dict['uuid'],
-            title = note_dict['journalNoteAttributes']['title'],
-            journal_date = note_dict['journalNoteAttributes']['journalNoteDate'],
-            note_format = note_dict['journalNoteAttributes']['format'],
-            note = base64.b64decode(note_dict['journalNoteAttributes']['note']),
-            approved = note_dict['journalNoteAttributes'].get('approved', False)
-        )
-        notes.append(note)
-    return notes
-
-
 def add_case(case: NovaCase, nova_access: NovaAccess):
     """Add a case to KMD Nova. The case will be created as 'Active'.
 
@@ -371,12 +329,20 @@ def add_case(case: NovaCase, nova_access: NovaAccess):
     }
 
     if case.caseworker:
-        payload['caseworker'] = {
-            "kspIdentity": {
-                "racfId": case.caseworker.ident,
-                "fullName": case.caseworker.name
+        if case.caseworker.type == 'user':
+            payload['caseworker'] = {
+                "kspIdentity": {
+                    "racfId": case.caseworker.ident,
+                    "fullName": case.caseworker.name
+                }
+            }
+        elif case.caseworker.type == 'group':
+            payload['caseworker'] = {
+                "losIdentity": {
+                    "administrativeUnitId": case.caseworker.ident,
+                    "fullName": case.caseworker.name
+                }
             }
-        }
 
     headers = {'Content-Type': 'application/json', 'Authorization': f"Bearer {nova_access.get_bearer_token()}"}
 

itk_dev_shared_components/kmd_nova/nova_documents.py

@@ -10,8 +10,8 @@
 import requests
 
 from itk_dev_shared_components.kmd_nova.authentication import NovaAccess
-from itk_dev_shared_components.kmd_nova.nova_objects import Document, Caseworker
-from itk_dev_shared_components.kmd_nova. util import datetime_from_iso_string
+from itk_dev_shared_components.kmd_nova.nova_objects import Document
+from itk_dev_shared_components.kmd_nova.util import datetime_from_iso_string, extract_caseworker
 
 
 def get_documents(case_uuid: str, nova_access: NovaAccess) -> list[Document]:
@@ -56,16 +56,6 @@ def get_documents(case_uuid: str, nova_access: NovaAccess) -> list[Document]:
 
     documents = []
     for document_dict in response.json()['documents']:
-
-        if 'caseworker' in document_dict:
-            caseworker = Caseworker(
-                uuid = document_dict['caseworker']['kspIdentity']['novaUserId'],
-                name = document_dict['caseworker']['kspIdentity']['fullName'],
-                ident = document_dict['caseworker']['kspIdentity']['racfId']
-            )
-        else:
-            caseworker = None
-
         doc = Document(
             uuid = document_dict['documentUuid'],
             document_number = document_dict['documentNumber'],
@@ -78,7 +68,7 @@ def get_documents(case_uuid: str, nova_access: NovaAccess) -> list[Document]:
             file_extension = document_dict['fileExtension'],
             category_name = document_dict.get('documentCategoryName'),
             category_uuid = document_dict.get('documentCategoryUuid'),
-            caseworker=caseworker
+            caseworker=extract_caseworker(document_dict)
         )
         documents.append(doc)
 
@@ -197,12 +187,20 @@ def attach_document_to_case(case_uuid: str, document: Document, nova_access: Nov
     }
 
     if document.caseworker:
-        payload['caseworker'] = {
-            "kspIdentity": {
-                "racfId": document.caseworker.ident,
-                "fullName": document.caseworker.name
+        if document.caseworker.type == 'user':
+            payload['caseworker'] = {
+                "kspIdentity": {
+                    "racfId": document.caseworker.ident,
+                    "fullName": document.caseworker.name
+                }
+            }
+        elif document.caseworker.type == 'group':
+            payload['caseworker'] = {
+                "losIdentity": {
+                    "administrativeUnitId": document.caseworker.ident,
+                    "fullName": document.caseworker.name
+                }
             }
-        }
 
     headers = {'Content-Type': 'application/json', 'Authorization': f"Bearer {nova_access.get_bearer_token()}"}
     response = requests.post(url, params=params, headers=headers, json=payload, timeout=60)

itk_dev_shared_components/kmd_nova/nova_notes.py

@@ -18,6 +18,7 @@ def add_text_note(case_uuid: str, note_title: str, note_text: str, caseworker: C
         case_uuid: The uuid of the case to add the journal note to.
         note_title: The title of the note.
         note_text: The text content of the note.
+        caseworker: The author of the note.
         approved: Whether the journal note should be marked as approved in Nova.
         nova_access: The NovaAccess object used to authenticate.
 

itk_dev_shared_components/kmd_nova/nova_objects.py

@@ -22,6 +22,7 @@ class Caseworker:
     uuid: str
     name: str
     ident: str
+    type: Literal['user', 'group'] = 'user'
 
 
 @dataclass(slots=True, kw_only=True)

itk_dev_shared_components/kmd_nova/nova_tasks.py

@@ -35,14 +35,18 @@ def attach_task_to_case(case_uuid: str, task: Task, nova_access: NovaAccess) ->
         "caseUuid": case_uuid,
         "title": task.title,
         "description": task.description,  # Optional
-        "caseworkerPersonId": task.caseworker.uuid,  # Optional
         "statusCode": task.status_code,
         "deadline": datetime_to_iso_string(task.deadline),
         "startDate": datetime_to_iso_string(task.started_date),  # Optional
         "closeDate": datetime_to_iso_string(task.closed_date),  # Optional
         "taskTypeName": "Aktivitet"
     }
 
+    if task.caseworker.type == 'user':
+        payload["caseworkerPersonId"] = task.caseworker.uuid
+    elif task.caseworker.type == 'group':
+        payload["caseworkerGroupId"] = task.caseworker.uuid
+
     headers = {'Content-Type': 'application/json', 'Authorization': f"Bearer {nova_access.get_bearer_token()}"}
     response = requests.post(url, params=params, headers=headers, json=payload, timeout=60)
     response.raise_for_status()
@@ -114,7 +118,16 @@ def _extract_caseworker(task_dict: dict) -> Caseworker | None:
         return Caseworker(
             uuid = task_dict['caseWorker']['id'],
             ident = task_dict['caseWorker']['ident'],
-            name = task_dict['caseWorker']['name']
+            name = task_dict['caseWorker']['name'],
+            type='user'
+        )
+
+    if 'caseWorkerGroup' in task_dict:
+        return Caseworker(
+            uuid = task_dict['caseWorkerGroup']['id'],
+            ident = None,
+            name = task_dict['caseWorkerGroup']['name'],
+            type='group'
         )
 
     return None
@@ -143,14 +156,18 @@ def update_task(task: Task, case_uuid: str, nova_access: NovaAccess):
         "caseUuid": case_uuid,
         "title": task.title,
         "description": task.description,
-        "caseworkerPersonId": task.caseworker.uuid,
         "statusCode": task.status_code,
         "deadline": datetime_to_iso_string(task.deadline),
         "startDate": datetime_to_iso_string(task.started_date),
         "closeDate": datetime_to_iso_string(task.closed_date),
         "taskType": "Aktivitet"
     }
 
+    if task.caseworker.type == 'user':
+        payload["caseworkerPersonId"] = task.caseworker.uuid
+    elif task.caseworker.type == 'group':
+        payload["caseworkerGroupId"] = task.caseworker.uuid
+
     headers = {'Content-Type': 'application/json', 'Authorization': f"Bearer {nova_access.get_bearer_token()}"}
     response = requests.put(url, params=params, headers=headers, json=payload, timeout=60)
     response.raise_for_status()

itk_dev_shared_components/kmd_nova/util.py

@@ -3,6 +3,8 @@
 from datetime import datetime
 from typing import Optional
 
+from itk_dev_shared_components.kmd_nova.nova_objects import Caseworker
+
 
 def datetime_from_iso_string(date_string: Optional[str]) -> Optional[datetime]:
     """A helper function to convert an ISO date string to a datetime.
@@ -34,3 +36,37 @@ def datetime_to_iso_string(datetime_: Optional[datetime]) -> Optional[str]:
         return datetime_.isoformat()
 
     return None
+
+
+def extract_caseworker(response_dict: dict) -> Caseworker | None:
+    """Extract the case worker from a HTTP request response.
+    If the case worker is in a unexpected format, None is returned.
+
+    Args:
+        response_dict: The dictionary describing the response object.
+
+    Returns:
+        A case worker object describing the case worker if any.
+    """
+    try:
+        if 'caseworker' in response_dict:
+            if 'kspIdentity' in response_dict['caseworker']:
+                return Caseworker(
+                    uuid = response_dict['caseworker']['kspIdentity']['novaUserId'],
+                    name = response_dict['caseworker']['kspIdentity']['fullName'],
+                    ident = response_dict['caseworker']['kspIdentity']['racfId'],
+                    type='user'
+                )
+
+            if 'losIdentity' in response_dict['caseworker']:
+                return Caseworker(
+                    uuid = response_dict['caseworker']['losIdentity']['novaUnitId'],
+                    name = response_dict['caseworker']['losIdentity']['fullName'],
+                    ident = str(response_dict['caseworker']['losIdentity']['administrativeUnitId']),
+                    type='group'
+                )
+
+    except KeyError:
+        pass
+
+    return None

tests/readme.md

@@ -62,9 +62,11 @@ NOVA_PARTY = '6101009805,Test Test'
 NOVA_DEPARTMENT = '{"id": 818485,"name": "Borgerservice","user_key": "4BBORGER"}'
 NOVA_CREDENTIALS = 'nova_login,nova_password'
 NOVA_USER = '{"name": "svcitkopeno svcitkopeno", "ident": "AZX0080", "uuid": "0bacdddd-5c61-4676-9a61-b01a18cec1d5"}'
+NOVA_USER_GROUP = '{"name": "ÅÅÅ_Frontoffice", "ident": "819697", "uuid": "144d7ab7-302f-4c62-83d2-dcdefcd92dea"}'
 
 NOVA_CVR_CASE = '{"cvr": "55133018", "case_title": "rpa_testcase", "case_number": "S2024-25614"}'
 NOVA_CPR_CASE = '{"cpr": "6101009805", "case_title": "Meget_Unik_Case_Overskrift", "case_number": "S2023-61078"}'
+NOVA_GROUP_CASE = '{"cpr": "6101009805", "case_title": "Testsag brugergruppe", "case_number": "S2025-5318"}'
 
 Note: The NOVA_CVR_CASE and NOVA_CPR_CASE variables require cases to be created in Nova, and the parameters set from those cases.
 

tests/test_nova_api/test_cases.py

@@ -61,10 +61,6 @@ def test_get_cvr_cases(self):
         case_title = cvr_case['case_title']
         case_number = cvr_case['case_number']
 
-        cases = nova_cases.get_cvr_cases(cvr=cvr, nova_access=self.nova_access)
-        self.assertIsInstance(cases[0], NovaCase)
-        self.assertEqual(cases[0].case_parties[0].identification, cvr)
-
         cases = nova_cases.get_cvr_cases(case_number=case_number, nova_access=self.nova_access)
         self.assertEqual(cases[0].case_number, case_number)
 
@@ -147,6 +143,56 @@ def test_add_case(self):
         self.assertEqual(nova_case.responsible_department.id, case.responsible_department.id)
         self.assertEqual(nova_case.security_unit.id, case.security_unit.id)
 
+    def test_user_groups(self):
+        """Test getting and adding a case with a user group as caseworker."""
+        # Get case
+        group_case = json.loads(os.environ['NOVA_GROUP_CASE'])
+        cpr = group_case['cpr']
+        case_number = group_case['case_number']
+        caseworker_dict = json.loads(os.environ['NOVA_USER_GROUP'])
+        caseworker = Caseworker(
+            type='group',
+            **caseworker_dict
+        )
+
+        cases = nova_cases.get_cases(cpr=cpr, case_number=case_number, nova_access=self.nova_access)
+
+        self.assertEqual(len(cases), 1)
+        nova_case = cases[0]
+        self.assertIsInstance(nova_case, NovaCase)
+
+        self.assertEqual(nova_case.caseworker, caseworker)
+
+        # Add case
+        nova_party = os.getenv('NOVA_PARTY').split(',')
+        party = CaseParty(
+            role="Primær",
+            identification_type="CprNummer",
+            identification=nova_party[0],
+            name=nova_party[1]
+        )
+
+        department_dict = json.loads(os.environ['NOVA_DEPARTMENT'])
+        department = Department(
+            **department_dict
+        )
+
+        case = NovaCase(
+            uuid=str(uuid.uuid4()),
+            title=f"Test {datetime.now()}",
+            case_date=datetime.now(),
+            progress_state="Opstaaet",
+            case_parties=[party],
+            kle_number="23.05.01",
+            proceeding_facet="G01",
+            sensitivity="Fortrolige",
+            caseworker=caseworker,
+            responsible_department=department,
+            security_unit=department
+        )
+
+        nova_cases.add_case(case, self.nova_access)
+
 
 if __name__ == '__main__':
     unittest.main()