test(data-extraction): Include first unit tests

tests/models/company_test.py

@@ -1,7 +1,12 @@
 """Test Models.company."""
 
 
-from aki_prj23_transparenzregister.models.company import Company, CompanyID, Location
+from aki_prj23_transparenzregister.models.company import (
+    Capital,
+    Company,
+    CompanyID,
+    Location,
+)
 
 
 def test_to_dict() -> None:
@@ -10,12 +15,17 @@ def test_to_dict() -> None:
     location = Location(
         city="Insmouth", house_number="19", street="Harbor", zip_code="1890"
     )
+    capital = Capital(currency="BTC", type="Virtual assets", value=42)
     company = Company(
         id=company_id,
         last_update="Tomorrow",
         location=location,
         name="BLANK GmbH",
         relationships=[],
+        business_purpose="Blockchain and NFTs",
+        capital=capital,
+        company_type="Something",
+        founding_date="Yesterday",
     )
 
     assert company.to_dict() == {
@@ -32,4 +42,12 @@ def test_to_dict() -> None:
         },
         "name": "BLANK GmbH",
         "relationships": [],
+        "business_purpose": "Blockchain and NFTs",
+        "capital": {
+            "value": capital.value,
+            "currency": capital.currency,
+            "type": capital.type,
+        },
+        "company_type": "Something",
+        "founding_date": "Yesterday",
     }

tests/utils/data_extraction/__init__.py

@@ -1 +0,0 @@
-"""Tests for data_extraction."""

tests/utils/data_extraction/unternehmensregister/extract_test.py

@@ -0,0 +1,81 @@
+import glob
+import os
+from tempfile import TemporaryDirectory
+
+from aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister import (
+    extract,
+)
+
+
+def prepare_temporary_dir(directory: str, formats: list[str]) -> None:
+    for index in range(len(formats)):
+        test_file = os.path.join(directory, f"file-{index}.{formats[index]}")
+        with open(test_file, "w") as file:
+            file.write(f"Hello There {index}")
+
+
+def test_rename_latest_file() -> None:
+    with TemporaryDirectory() as temp_dir:
+        # Create some test files in the temporary directory
+        test_file1 = os.path.join(temp_dir, "file1.xml")
+        test_file2 = os.path.join(temp_dir, "file2.xml")
+        test_file3 = os.path.join(temp_dir, "file3.xml")
+
+        # Create files with different modification times
+        with open(test_file1, "w") as f:
+            f.write("Content 1")
+        with open(test_file2, "w") as f:
+            f.write("Content 2")
+        with open(test_file3, "w") as f:
+            f.write("Content 3")
+
+        # Rename the latest file to 'new_file.xml'
+        extract.rename_latest_file(temp_dir, "new_file.xml")
+
+        glob.glob1(temp_dir, "*.xml")
+        # Verify that 'file3.xml' is renamed to 'new_file.xml'
+        assert not os.path.exists(test_file3)
+        assert os.path.exists(os.path.join(temp_dir, "new_file.xml"))
+
+        # Verify that 'file1.xml' and 'file2.xml' are still present
+        assert os.path.exists(test_file1)
+        assert os.path.exists(test_file2)
+
+        # Verify that renaming with a different pattern works
+        with open(test_file1, "w") as f:
+            f.write("Content 4")
+        with open(os.path.join(temp_dir, "file4.txt"), "w") as f:
+            f.write("Content 5")
+
+        # Rename the latest .txt file to 'new_file.txt'
+        extract.rename_latest_file(temp_dir, "new_file.txt", pattern="*.txt")
+
+        # Verify that 'file4.txt' is renamed to 'new_file.txt'
+        assert not os.path.exists(os.path.join(temp_dir, "file4.txt"))
+        assert os.path.exists(os.path.join(temp_dir, "new_file.txt"))
+
+        # Verify that 'file1.xml' is still present and unchanged
+        with open(test_file1) as f:
+            assert f.read() == "Content 4"
+
+
+def test_get_num_files_default_pattern() -> None:
+    with TemporaryDirectory() as temp_dir:
+        prepare_temporary_dir(temp_dir, ["xml", "xml", "xml"])
+
+        expected_result = 3
+        assert extract.get_num_files(temp_dir) == expected_result
+
+
+def test_get_num_files_different_pattern() -> None:
+    with TemporaryDirectory() as temp_dir:
+        prepare_temporary_dir(temp_dir, ["xml", "txt", "json"])
+
+        num_files = extract.get_num_files(temp_dir, "*.txt")
+        assert num_files == 1
+
+
+def test_wait_for_download_condition() -> None:
+    with TemporaryDirectory() as temp_dir:
+        prepare_temporary_dir(temp_dir, ["xml", "txt"])
+        assert extract.wait_for_download_condition(temp_dir, 2) is False

tests/utils/data_extraction/unternehmensregister/transform_test.py

@@ -0,0 +1,208 @@
+import json
+import os
+from tempfile import TemporaryDirectory
+
+from aki_prj23_transparenzregister.models.company import (
+    CompanyRelationshipEnum,
+    CompanyToCompanyRelationship,
+    Location,
+    PersonName,
+    PersonToCompanyRelationship,
+    RelationshipRoleEnum,
+)
+from aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister import (
+    transform,
+)
+
+
+def test_transform_xml_to_json() -> None:
+    with TemporaryDirectory() as temp_source_dir:
+        with open(os.path.join(temp_source_dir, "test.xml"), "w") as file:
+            xml_input = """<?xml version="1.0" encoding="UTF-8"?>
+<test>
+  <message>Hello World!</message>
+</test>
+            """
+            file.write(xml_input)
+        with TemporaryDirectory() as temp_target_dir:
+            transform.transform_xml_to_json(temp_source_dir, temp_target_dir)
+            with open(os.path.join(temp_target_dir, "test.json")) as file:
+                json_output = json.load(file)
+                assert json_output == {"test": {"message": "Hello World!"}}
+
+
+def test_parse_stakeholder_org_hidden_in_person() -> None:
+    data = {
+        "Beteiligter": {
+            "Natuerliche_Person": {
+                "Voller_Name": {"Vorname": None, "Nachname": "Some Company KG"},
+                "Anschrift": {"Ort": "Area 51"},
+            }
+        },
+        "Rolle": {"Rollenbezeichnung": {"content": "Kommanditist(in)"}},
+    }
+    expected_result = CompanyToCompanyRelationship(
+        role=RelationshipRoleEnum.KOMMANDITIST,  # type: ignore
+        description="Some Company KG",
+        type=CompanyRelationshipEnum.COMPANY,
+        location=Location(**{"city": "Area 51"}),
+    )
+    assert transform.parse_stakeholder(data) == expected_result
+
+
+def test_parse_stakeholder_person() -> None:
+    data = {
+        "Beteiligter": {
+            "Natuerliche_Person": {
+                "Voller_Name": {"Vorname": "Stephen", "Nachname": "King"},
+                "Anschrift": {"Ort": "Maine"},
+                "Geburt": {"Geburtsdatum": "1947-09-21"},
+            }
+        },
+        "Rolle": {"Rollenbezeichnung": {"content": "Geschäftsleiter(in)"}},
+    }
+    expected_result = PersonToCompanyRelationship(
+        role=RelationshipRoleEnum.GESCHAEFTSLEITER,  # type: ignore
+        date_of_birth="1947-09-21",
+        name=PersonName(**{"firstname": "Stephen", "lastname": "King"}),
+        type=CompanyRelationshipEnum.PERSON,
+        location=Location(**{"city": "Maine"}),
+    )
+    assert transform.parse_stakeholder(data) == expected_result
+
+
+def test_parse_stakeholder_org() -> None:
+    data = {
+        "Beteiligter": {
+            "Organisation": {
+                "Bezeichnung": {"Bezeichnung_Aktuell": "Transparenzregister kG"},
+                "Anschrift": {
+                    "Ort": "Iserlohn",
+                    "Strasse": "Hauptstrasse",
+                    "Hausnummer": "42",
+                    "Postleitzahl": "58636",
+                },
+                "Geburt": {"Geburtsdatum": "1947-09-21"},
+            }
+        },
+        "Rolle": {"Rollenbezeichnung": {"content": "Geschäftsführender Direktor"}},
+    }
+    expected_result = CompanyToCompanyRelationship(
+        description="Transparenzregister kG",
+        role=RelationshipRoleEnum.DIREKTOR,  # type: ignore
+        type=CompanyRelationshipEnum.COMPANY,
+        location=Location(
+            **{
+                "city": "Iserlohn",
+                "zip_code": "58636",
+                "house_number": "42",
+                "street": "Hauptstrasse",
+            }
+        ),
+    )
+    assert transform.parse_stakeholder(data) == expected_result
+
+
+def test_parse_stakeholder_no_result() -> None:
+    data: dict = {"Beteiligter": {}}
+    assert transform.parse_stakeholder(data) is None
+
+
+def test_loc_from_beteiligung() -> None:
+    data = {
+        "XJustiz_Daten": {
+            "Grunddaten": {
+                "Verfahrensdaten": {
+                    "Beteiligung": [
+                        {
+                            "Beteiligter": {
+                                "Beteiligtennummer": "1",
+                                "Organisation": {
+                                    "Bezeichnung": {
+                                        "Bezeichnung_Aktuell": "1 A Autenrieth Kunststofftechnik GmbH & Co. KG"
+                                    },
+                                    "Sitz": {
+                                        "Ort": "Heroldstatt",
+                                        "Staat": {
+                                            "@xsi:type": "WL_Staaten",
+                                            "@wl_version": "1.5",
+                                            "@wl_fassung": "2",
+                                            "content": "DE",
+                                        },
+                                    },
+                                    "Anschrift": {
+                                        "Strasse": "Gewerbestraße",
+                                        "Hausnummer": "8",
+                                        "Postleitzahl": "72535",
+                                        "Ort": "Heroldstatt",
+                                    },
+                                },
+                            }
+                        },
+                    ]
+                }
+            }
+        }
+    }
+
+    expected_result = Location(
+        city="Heroldstatt", house_number="8", street="Gewerbestraße", zip_code="72535"
+    )
+    assert transform.loc_from_beteiligung(data) == expected_result
+
+
+def test_name_from_beteiligung() -> None:
+    data = {
+        "XJustiz_Daten": {
+            "Grunddaten": {
+                "Verfahrensdaten": {
+                    "Beteiligung": [
+                        {
+                            "Beteiligter": {
+                                "Beteiligtennummer": "1",
+                                "Organisation": {
+                                    "Bezeichnung": {
+                                        "Bezeichnung_Aktuell": "1 A Autenrieth Kunststofftechnik GmbH & Co. KG"
+                                    },
+                                },
+                            }
+                        },
+                    ]
+                }
+            }
+        }
+    }
+
+    expected_result = "1 A Autenrieth Kunststofftechnik GmbH & Co. KG"
+    assert transform.name_from_beteiligung(data) == expected_result
+
+
+def test_map_rechtsform() -> None:
+    data = {
+        "XJustiz_Daten": {
+            "Fachdaten_Register": {
+                "Basisdaten_Register": {
+                    "Aktuelles_Satzungsdatum": "1952-07-15",
+                    "Rechtstraeger": {
+                        "Rechtsform": {
+                            "content": "Gesellschaft mit beschränkter Haftung"
+                        },
+                    },
+                }
+            }
+        }
+    }
+    expected_result = "Gesellschaft mit beschränkter Haftung"
+    assert transform.map_rechtsform("", data) == expected_result
+
+
+def test_map_rechtsform_from_name() -> None:
+    data = [
+        ("GEA Farm Technologies GmbH", "Gesellschaft mit beschränkter Haftung"),
+        ("Atos SE", "Europäische Aktiengesellschaft (SE)"),
+        ("Bilkenroth KG", "Kommanditgesellschaft"),
+        ("jfoiahfo8sah 98548902 öhz ö", None),
+    ]
+
+    for company_name, expected_result in data:
+        assert transform.map_rechtsform(company_name, {}) == expected_result