Feature/getorganized

.pylintrc

@@ -2,4 +2,5 @@
 disable = 
   C0301, # Line too long
   I1101, E1101, # C-modules members
-  R0913, R0917 # Too many arguments
+  R0913, R0917, # Too many arguments
+  W0223 # Abstract class functions missing

changelog.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- getorganized: Added functions and tests for using GetOrganized.
+
 ## [2.11.1] - 2025-04-28
 
 ### Fixed

itk_dev_shared_components/getorganized/go_api.py

@@ -0,0 +1,154 @@
+"""Functions for working with the GetOrganized API."""
+
+import json
+from urllib.parse import urljoin
+from typing import Literal
+
+from requests import Session
+from requests_ntlm import HttpNtlmAuth
+
+
+def create_session(username: str, password: str) -> Session:
+    """Create a session for accessing GetOrganized API.
+
+    Args:
+        username: Username for login.
+        password: Password for login.
+
+    Returns:
+        Return the session object
+    """
+    session = Session()
+    session.headers.setdefault("Content-Type", "application/json")
+    session.auth = HttpNtlmAuth(username, password)
+    return session
+
+
+def upload_document(*, apiurl: str, file: bytearray, case_id: str, filename: str, agent_name: str | None = None, date_string: str | None = None, session: Session, doc_category: str | None = None, case_type: str = Literal["EMN", "GEO"], overwrite: bool = True) -> tuple[str, Session]:
+    """Upload a document to Get Organized.
+
+    Args:
+        apiurl: Base url for API.
+        session: Session token for request.
+        file: Bytearray of file to upload.
+        case_id: Case ID already present in GO.
+        filename: Name of file when saved in GO.
+        agent_name: Agent name, used for creating a folder in GO. Defaults to None.
+        date_string: A date to add as metadata to GetOrganized. Defaults to None.
+
+    Returns:
+        Return response text.
+    """
+    url = apiurl + "/_goapi/Documents/AddToCase"
+    payload = {
+        "Bytes": list(file),
+        "CaseId": case_id,
+        "SiteUrl": urljoin(apiurl, f"/case{case_type}/{case_id}"),
+        "ListName": "Dokumenter",
+        "FolderPath": agent_name,
+        "FileName": filename,
+        "Metadata": f"<z:row xmlns:z='#RowsetSchema' ows_Dato='{date_string}' ows_Kategori='{doc_category}'/>",
+        "Overwrite": overwrite
+    }
+    response = session.post(url, data=json.dumps(payload), timeout=60)
+    response.raise_for_status()
+    return response.text
+
+
+def delete_document(apiurl: str, document_id: int, session: Session) -> tuple[str, Session]:
+    """Delete a document from GetOrganized.
+
+    Args:
+        apiurl: Url of the GetOrganized API.
+        session: Session object used for logging in.
+        document_id: ID of the document to delete.
+
+    Returns:
+        Return the response.
+    """
+    url = urljoin(apiurl, "/_goapi/Documents/ByDocumentId/")
+    payload = {
+        "DocId": document_id,
+        "ForceDelete": True
+    }
+    response = session.delete(url, timeout=60, data=json.dumps(payload))
+    response.raise_for_status()
+    return response
+
+
+def create_case(session: Session, apiurl: str, title: str, category: str, department: str, kle: str, case_type: str = Literal["EMN", "GEO"]) -> str:
+    """Create a case in GetOrganized.
+
+    Args:
+        apiurl: Url for the GetOrganized API.
+        session: Session object to access API.
+        title: Title of the case being created.
+        category: Case category to create the case for.
+        department: Department for the case.
+        kle: KLE number for the case (https://www.kle-online.dk/emneplan/00/)
+
+    Returns:
+        Return the caseID of the created case.
+    """
+    url = urljoin(apiurl, "/_goapi/Cases/")
+    payload = {
+        'CaseTypePrefix': case_type,
+        'MetadataXml': f'''<z:row xmlns:z="#RowsetSchema"
+                            ows_Title="{title}"
+                            ows_CaseStatus="Åben"
+                            ows_CaseCategory="{category}"
+                            ows_Afdeling="{department}"
+                            ows_KLENummer="{kle}"/>''',
+        'ReturnWhenCaseFullyCreated': False
+    }
+    response = session.post(url, data=json.dumps(payload), timeout=60)
+    response.raise_for_status()
+    return response.json()["CaseID"]
+
+
+def get_case_metadata(session: Session, apiurl: str, case_id: str) -> str:
+    """Get metadata for a GetOrganized case, to look through parameters and values.
+
+    Args:
+        session: Session token.
+        apiurl: Base URL for the API.
+        case_id: Case ID to get metadata on.
+
+    Returns:
+        Return the metadata for the case as an XML string.
+    """
+    url = urljoin(apiurl, f"/_goapi/Cases/Metadata/{case_id}")
+    response = session.get(url, timeout=60)
+    response.raise_for_status()
+    return response.json()["Metadata"]
+
+
+def find_case(session: Session, apiurl: str, case_title: str, case_type: str = Literal["EMN", "GEO"]) -> list[str]:
+    """Search for an existing case in GO with the given case title.
+    The search finds any case that contains the given title in its title.
+
+    Args:
+        case_title: The title to search for.
+        session: Session object to access the API.
+
+    Returns:
+        The case id of the found case(s) if any.
+    """
+    url = apiurl + "/_goapi/Cases/FindByCaseProperties"
+    payload = {
+        "FieldProperties": [
+            {
+                "InternalName": "ows_Title",
+                "Value": case_title,
+                "ComparisonType": "Contains",
+            }
+        ],
+        "CaseTypePrefixes": [case_type],
+        "LogicalOperator": "AND",
+        "ExcludeDeletedCases": True
+    }
+    response = session.post(url, data=json.dumps(payload), timeout=60)
+    response.raise_for_status()
+    cases = response.json()['CasesInfo']
+
+    return [case['CaseID'] for case in cases]

pyproject.toml

@@ -22,7 +22,8 @@ dependencies = [
   "requests == 2.*",
   "beautifulsoup4 == 4.*",
   "selenium == 4.*",
-  "uiautomation == 2.*"
+  "uiautomation == 2.*",
+  "requests_ntlm == 1.*"
 ]
 
 [project.urls]

tests/readme.md

@@ -74,9 +74,20 @@ Note: The NOVA_CVR_CASE and NOVA_CPR_CASE variables require cases to be created
 
 CVR_CREDS = 'login;password'
 
-### EFLYT
+### eFlyt
 
-EFLYT_LOGIN = 'username,password'
+EFLYT_LOGIN = 'login,password'
 TEST_CPR = 'XXXXXXXXXX'
 TEST_CASE = '123456' # A test case with multiple current inhabitants, with relations registered
 TEST_CASE_NOONE = '56789' # A test case without any current inhabitants
+
+### GetOrganized
+
+GO_LOGIN = 'login;password'
+GO_APIURL = 'https://test.go.aarhuskommune.dk'
+GO_CATEGORY="Åben for alle",
+GO_DEPARTMENT="916;#Backoffice - Drift og Økonomi",
+GO_KLE="318;#25.02.00 Ejendomsbeskatning i almindelighed"
+
+These GO-variables are found by fetching metadata for a case created in the GO interface with the correct setup for a specific process.
+Use get_case_metadata on a known case ID with the required setup, and use those when implementing GetOrganized.

tests/test_getorganized/test_case.py

@@ -0,0 +1,61 @@
+"""Tests related to the GetOrganized module."""
+
+import unittest
+import os
+import re
+import json
+from uuid import uuid4
+
+from dotenv import load_dotenv
+
+from itk_dev_shared_components.getorganized import go_api
+
+load_dotenv()
+
+
+class CaseTest(unittest.TestCase):
+    """Test the Case functionality of GetOrganized integration"""
+    test_case = None
+
+    @classmethod
+    def setUpClass(cls):
+        user, password = os.getenv("GO_LOGIN").split(",")
+
+        cls.session = go_api.create_session(user, password)
+        uuid = uuid4()
+        cls.test_case = go_api.create_case(session=cls.session,
+                                           apiurl=os.getenv("GO_APIURL"),
+                                           title=uuid,
+                                           case_type="EMN",
+                                           category=os.getenv("GO_CATEGORY"),
+                                           department=os.getenv("GO_DEPARTMENT"),
+                                           kle=os.getenv("GO_KLE")
+                                           )
+
+    def test_case_created(self):
+        """Test case is created."""
+        self.assertIsNotNone(self.test_case)
+
+    def test_document(self):
+        """Test upload and delete of a document."""
+        test_data = bytearray(b"Testdata")
+        document = go_api.upload_document(session=self.session, apiurl=os.getenv("GO_APIURL"), case_id=self.test_case, filename="Testfil", file=test_data)
+        self.assertIsNotNone(document)
+        response = go_api.delete_document(session=self.session, apiurl=os.getenv("GO_APIURL"), document_id=json.loads(document)['DocId'])
+        self.assertEqual(response.status_code, 200)
+
+    def test_find_case(self):
+        """Test finding a case and getting metadata."""
+        metadata = go_api.get_case_metadata(self.session, os.getenv("GO_APIURL"), self.test_case)
+        self.assertIsNotNone(metadata)
+
+        test_case_title = re.match('.*ows_Title="([^"]+)"', metadata)[1]
+        case_found = go_api.find_case(session=self.session, apiurl=os.getenv("GO_APIURL"), case_title=test_case_title, case_type="EMN")
+        if isinstance(case_found, list):
+            self.assertIn(self.test_case, case_found)
+        else:
+            self.assertEqual(self.test_case, case_found)
+
+
+if __name__ == '__main__':
+    unittest.main()