From 7269e7e6a4d5441e48e072c13aadf13f0140ae4c Mon Sep 17 00:00:00 2001 From: TrisNol Date: Mon, 18 Sep 2023 20:23:19 +0200 Subject: [PATCH] test(data-extraction): Cover transform.py --- .../models/company.py | 37 +- .../unternehmensregister/load.py | 5 +- .../unternehmensregister/transform.py | 160 +++---- .../unternehmensregister/transform_test.py | 397 ++++++++++++++++++ 4 files changed, 511 insertions(+), 88 deletions(-) diff --git a/src/aki_prj23_transparenzregister/models/company.py b/src/aki_prj23_transparenzregister/models/company.py index 67806a9..11b2ec9 100644 --- a/src/aki_prj23_transparenzregister/models/company.py +++ b/src/aki_prj23_transparenzregister/models/company.py @@ -54,16 +54,24 @@ class CompanyTypeEnum(str, MultiValueEnum): @dataclass -class CompanyID: - """_summary_.""" +class DistrictCourt: + """DistrictCourt.""" - district_court: str + name: str + city: str + + +@dataclass +class CompanyID: + """CompanyID.""" + + district_court: DistrictCourt hr_number: str @dataclass class Location: - """_summary_.""" + """Location.""" city: str street: str | None = None @@ -155,14 +163,29 @@ class YearlyResult: kpis: dict[FinancialKPIEnum, float] +class CurrencyEnum(str, MultiValueEnum): + """Enum of possible currencies.""" + + EURO = "EUR" + DEUTSCHE_MARK = "DM", "DEM" + KEINE_ANGABE = "" + + +class CapitalTypeEnum(str, Enum): + """Enum of possible capital types.""" + + HAFTEINLAGE = "Hafteinlage" + STAMMKAPITAL = "Stammkapital" + GRUNDKAPITAL = "Grundkapital" + + @dataclass class Capital: """Capital of company.""" value: float - currency: str # TODO define Enum - # TODO define Enum - type: str # noqa: A003 + currency: CurrencyEnum + type: CapitalTypeEnum # noqa: A003 @dataclass diff --git a/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/load.py b/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/load.py index 79be6a4..52bf304 100644 --- a/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/load.py +++ b/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/load.py @@ -25,8 +25,9 @@ if __name__ == "__main__": connector = MongoConnector(conn_string) service = CompanyMongoService(connector) - for file in tqdm(glob.glob1("./data/Unternehmensregister/transformed", "*.json")): - path = os.path.join("./data/Unternehmensregister/transformed", file) + base_path = "./Jupyter/API-tests/Unternehmensregister/data/Unternehmensregister" + for file in tqdm(glob.glob1(f"{base_path}/transformed", "*.json")): + path = os.path.join(f"{base_path}/transformed", file) with open(path, encoding="utf-8") as file_object: data = json.loads(file_object.read()) company: Company = Company(**data) diff --git a/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/transform.py b/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/transform.py index a95667f..29db73a 100644 --- a/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/transform.py +++ b/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/transform.py @@ -11,12 +11,15 @@ from tqdm import tqdm from aki_prj23_transparenzregister.models.company import ( Capital, + CapitalTypeEnum, Company, CompanyID, CompanyRelationship, CompanyRelationshipEnum, CompanyToCompanyRelationship, CompanyTypeEnum, + CurrencyEnum, + DistrictCourt, Location, PersonName, PersonToCompanyRelationship, @@ -236,20 +239,21 @@ def map_rechtsform(company_name: str, data: dict) -> CompanyTypeEnum | None: return None -def map_capital(data: dict, company_type: str) -> Capital | None: +def map_capital(data: dict, company_type: CompanyTypeEnum) -> Capital | None: """Extracts the company capital from the given Unternehmensregister export. Args: data (dict): Data export - company_type (str): Type of company (e.g., 'Gesellschaft mit beschränkter Haftung') + company_type (CompanyTypeEnum): Type of company (e.g., 'Gesellschaft mit beschränkter Haftung') Returns: Capital | None: Company Capital if found """ + # Early return + if "Zusatzangaben" not in data["XJustiz_Daten"]["Fachdaten_Register"]: + return None capital: dict = {"Zahl": 0.0, "Waehrung": ""} - if company_type == "Kommanditgesellschaft": - if "Zusatzangaben" not in data["XJustiz_Daten"]["Fachdaten_Register"]: - return None + if company_type == CompanyTypeEnum.KG: capital_type = "Hafteinlage" base = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ "Personengesellschaft" @@ -259,55 +263,50 @@ def map_capital(data: dict, company_type: str) -> Capital | None: # TODO link to persons using Ref_Rollennummer then extract ["Hafteinlage"] as below capital["Zahl"] = capital["Zahl"] + float(entry["Hafteinlage"]["Zahl"]) capital["Waehrung"] = entry["Hafteinlage"]["Waehrung"] - elif type(base) == "dict": + elif isinstance(base, dict): capital = base["Hafteinlage"] elif company_type in [ - "Gesellschaft mit beschränkter Haftung", - "Europäische Aktiengesellschaft (SE)", - "Aktiengesellschaft", - "Kommanditgesellschaft auf Aktien", - "Rechtsform ausländischen Rechts HRB", + CompanyTypeEnum.GMBH, + CompanyTypeEnum.SE, + CompanyTypeEnum.AG, + CompanyTypeEnum.KGaA, + CompanyTypeEnum.AUSLAENDISCHE_RECHTSFORM, + CompanyTypeEnum.OHG, ]: - if "Zusatzangaben" not in data["XJustiz_Daten"]["Fachdaten_Register"]: - return None if ( - "Zusatz_GmbH" - in data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ + "Kapitalgesellschaft" + not in data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"] + ): + base = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ + "Personengesellschaft" + ] + else: + base = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ "Kapitalgesellschaft" ] - ): + if "Zusatz_GmbH" in base: capital_type = "Stammkapital" - capital = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ - "Kapitalgesellschaft" - ]["Zusatz_GmbH"]["Stammkapital"] - elif ( - "Zusatz_Aktiengesellschaft" - in data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ - "Kapitalgesellschaft" - ] - ): + capital = base["Zusatz_GmbH"]["Stammkapital"] + elif "Zusatz_Aktiengesellschaft" in base: capital_type = "Grundkapital" - capital = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][ - "Kapitalgesellschaft" - ]["Zusatz_Aktiengesellschaft"]["Grundkapital"]["Hoehe"] + capital = base["Zusatz_Aktiengesellschaft"]["Grundkapital"]["Hoehe"] elif company_type in [ - "Einzelkaufmann", - "Einzelkauffrau", - "eingetragene Genossenschaft", - "Partnerschaft", - "Einzelkaufmann / Einzelkauffrau", - "Offene Handelsgesellschaft", - "Partnerschaftsgesellschaft", + CompanyTypeEnum.EINZELKAUFMANN, + CompanyTypeEnum.EG, + CompanyTypeEnum.PARTNERSCHAFT, + CompanyTypeEnum.PARTNERGESELLSCHAFT, + CompanyTypeEnum.PARTNERSCHAFTSGESELLSCHAFT, None, ]: return None - else: + # Catch entries having the dict but with null values + if not all(capital.values()): return None return Capital( - **{ - "value": capital["Zahl"], - "currency": capital["Waehrung"], - "type": capital_type, + **{ # type: ignore + "value": float(capital["Zahl"]), + "currency": CurrencyEnum(capital["Waehrung"]), + "type": CapitalTypeEnum(capital_type), } ) @@ -366,17 +365,6 @@ def map_founding_date(data: dict) -> str | None: ) if len(entry_date) == 1: return transform_date_to_iso(entry_date[0]) - - if "Eintragungstext" in data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"] and ( - type(data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"]["Eintragungstext"]) - == "list" - ): - temp = data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"]["Eintragungstext"][ - 0 - ]["Text"] - results = re.findall(r"\d{1,2}\.\d{1,2}\.\d{2,4}", temp) - if len(temp) == 1: - return transform_date_to_iso(results[0]) if ( "Gruendungsmetadaten" in data["XJustiz_Daten"]["Fachdaten_Register"]["Basisdaten_Register"] @@ -402,34 +390,50 @@ def map_company_id(data: dict) -> CompanyID: "hr_number": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ "Instanzdaten" ]["Aktenzeichen"], - "district_court": { - "name": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ - "Beteiligung" - ][1]["Beteiligter"]["Organisation"]["Bezeichnung"][ - "Bezeichnung_Aktuell" - ] - if "Organisation" - in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ - "Beteiligung" - ][1]["Beteiligter"] - else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ - "Beteiligung" - ][1]["Beteiligter"]["Natuerliche_Person"]["Voller_Name"]["Nachname"], - "city": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ - "Beteiligung" - ][1]["Beteiligter"]["Organisation"]["Sitz"]["Ort"] - if "Organisation" - in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ - "Beteiligung" - ][1]["Beteiligter"] - else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ - "Beteiligung" - ][1]["Beteiligter"]["Natuerliche_Person"]["Anschrift"]["Ort"], - }, + "district_court": DistrictCourt( + **{ + "name": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ + "Beteiligung" + ][1]["Beteiligter"]["Organisation"]["Bezeichnung"][ + "Bezeichnung_Aktuell" + ] + if "Organisation" + in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ + "Beteiligung" + ][1]["Beteiligter"] + else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ + "Beteiligung" + ][1]["Beteiligter"]["Natuerliche_Person"]["Voller_Name"][ + "Nachname" + ], + "city": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ + "Beteiligung" + ][1]["Beteiligter"]["Organisation"]["Sitz"]["Ort"] + if "Organisation" + in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ + "Beteiligung" + ][1]["Beteiligter"] + else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][ + "Beteiligung" + ][1]["Beteiligter"]["Natuerliche_Person"]["Anschrift"]["Ort"], + } + ), } ) +def map_last_update(data: dict) -> str: + """Extract last update date from export. + + Args: + data (dict): Unternehmensregister export + + Returns: + str: Last update date + """ + return data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"]["letzte_Eintragung"] + + def map_unternehmensregister_json(data: dict) -> Company: """Processes the Unternehmensregister structured export to a Company by using several helper methods. @@ -446,10 +450,8 @@ def map_unternehmensregister_json(data: dict) -> Company: result["name"] = name_from_beteiligung(data) result["location"] = loc_from_beteiligung(data) - result["last_update"] = data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"][ - "letzte_Eintragung" - ] - # TODO New features --> to be tested + result["last_update"] = map_last_update(data) + result["company_type"] = map_rechtsform(result["name"], data) result["capital"] = map_capital(data, result["company_type"]) result["business_purpose"] = map_business_purpose(data) diff --git a/tests/utils/data_extraction/unternehmensregister/transform_test.py b/tests/utils/data_extraction/unternehmensregister/transform_test.py index 091f861..4865880 100644 --- a/tests/utils/data_extraction/unternehmensregister/transform_test.py +++ b/tests/utils/data_extraction/unternehmensregister/transform_test.py @@ -1,10 +1,18 @@ import json import os from tempfile import TemporaryDirectory +from unittest.mock import Mock, patch from aki_prj23_transparenzregister.models.company import ( + Capital, + CapitalTypeEnum, + Company, + CompanyID, CompanyRelationshipEnum, CompanyToCompanyRelationship, + CompanyTypeEnum, + CurrencyEnum, + DistrictCourt, Location, PersonName, PersonToCompanyRelationship, @@ -206,3 +214,392 @@ def test_map_rechtsform_from_name() -> None: for company_name, expected_result in data: assert transform.map_rechtsform(company_name, {}) == expected_result + + +def test_map_capital_kg_single() -> None: + capital = Capital( + currency=CurrencyEnum.EURO, value=69000, type=CapitalTypeEnum.HAFTEINLAGE # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Personengesellschaft": { + "Zusatz_KG": { + "Daten_Kommanditist": { + "Hafteinlage": { + "Zahl": str(capital.value), + "Waehrung": capital.currency, + }, + } + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.KG) # type: ignore + assert result == capital + + +def test_map_capital_kg_sum() -> None: + capital = Capital( + currency=CurrencyEnum.EURO, value=20000, type=CapitalTypeEnum.HAFTEINLAGE # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Personengesellschaft": { + "Zusatz_KG": { + "Daten_Kommanditist": [ + { + "Hafteinlage": { + "Zahl": str(10000), + "Waehrung": capital.currency, + } + }, + { + "Hafteinlage": { + "Zahl": str(10000), + "Waehrung": capital.currency, + }, + }, + ] + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.KG) # type: ignore + assert result == capital + + +def test_map_capital_no_fachdaten() -> None: + data = {"XJustiz_Daten": {"Fachdaten_Register": {}}} # type: ignore + + result = transform.map_capital(data, CompanyTypeEnum.KG) # type: ignore + assert result is None + + +def test_map_capital_gmbh() -> None: + capital = Capital( + currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Kapitalgesellschaft": { + "Zusatz_GmbH": { + "Stammkapital": { + "Zahl": str(capital.value), + "Waehrung": capital.currency, + }, + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.GMBH) # type: ignore + assert result == capital + + +def test_map_capital_ag() -> None: + capital = Capital( + currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.GRUNDKAPITAL # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Kapitalgesellschaft": { + "Zusatz_Aktiengesellschaft": { + "Grundkapital": { + "Hoehe": { + "Zahl": str(capital.value), + "Waehrung": capital.currency, + } + }, + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.SE) # type: ignore + assert result == capital + + +def test_map_capital_personengesellschaft() -> None: + capital = Capital( + currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Personengesellschaft": { + "Zusatz_GmbH": { + "Stammkapital": { + "Zahl": str(capital.value), + "Waehrung": capital.currency, + }, + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.OHG) # type: ignore + assert result == capital + + +def test_map_capital_einzelkaufmann() -> None: + capital = Capital( + currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Personengesellschaft": { + "Zusatz_GmbH": { + "Stammkapital": { + "Zahl": str(capital.value), + "Waehrung": capital.currency, + }, + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.EINZELKAUFMANN) # type: ignore + assert result is None + + +def test_map_capital_partial_null_values() -> None: + capital = Capital( + currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore + ) + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Zusatzangaben": { + "Personengesellschaft": { + "Zusatz_GmbH": { + "Stammkapital": { + "Zahl": None, + "Waehrung": capital.currency, + }, + } + } + } + } + } + } + + result = transform.map_capital(data, CompanyTypeEnum.OHG) # type: ignore + assert result is None + + +def test_map_business_purpose() -> None: + business_purpose = "Handel mit Betäubungsmitteln aller Art" + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Basisdaten_Register": { + "Gegenstand_oder_Geschaeftszweck": business_purpose + } + } + } + } + + result = transform.map_business_purpose(data) + assert result == business_purpose + + +def test_map_business_purpose_no_result() -> None: + data: dict = {"XJustiz_Daten": {}} + + result = transform.map_business_purpose(data) + assert result is None + + +def test_transform_date_to_iso() -> None: + date = "10.10.1111" + expected_result = "1111-10-10" + result = transform.transform_date_to_iso(date) + assert result == expected_result + + +def test_transform_date_to_iso_2_char_year() -> None: + date = "10.10.98" + expected_result = "1998-10-10" + result = transform.transform_date_to_iso(date) + assert result == expected_result + + +def test_map_founding_date_from_tag_der_ersten_eintragung() -> None: + data = { + "some entry": "Tag der ersten Eintragung: 01.05.2004", + "some other entry": "hfjdoöiashföahöf iodsazo8 5z4o fdsha8oü gfdsö", + } + expected_result = "2004-05-01" + result = transform.map_founding_date(data) + assert result == expected_result + + +def test_map_founding_date_from_gesellschaftsvertrag() -> None: + data = { + "some entry": "hfjdoöiashföahöf iodsazo8 5z4o fdsha8oü gfdsö", + "some other entry": "Das Wesen der Rekursion ist der Selbstaufruf Gesellschaftsvertrag vom 22.12.1996 Hallo Welt", + } + expected_result = "1996-12-22" + result = transform.map_founding_date(data) + assert result == expected_result + + +def test_map_founding_date_from_gruendungsdatum() -> None: + data = { + "XJustiz_Daten": { + "Fachdaten_Register": { + "Basisdaten_Register": { + "Gruendungsmetadaten": {"Gruendungsdatum": "1998-01-01"} + } + } + } + } + expected_result = "1998-01-01" + result = transform.map_founding_date(data) + assert result == expected_result + + +def test_map_founding_date_no_result() -> None: + data: dict = {"XJustiz_Daten": {"Fachdaten_Register": {"Basisdaten_Register": {}}}} + result = transform.map_founding_date(data) + assert result is None + + +def test_map_company_id() -> None: + district_court = DistrictCourt("Amtsgericht Ulm", "Ulm") + company_id = CompanyID(district_court, "HRA 4711") + data = { + "XJustiz_Daten": { + "Grunddaten": { + "@XJustizVersion": "1.20.0", + "Verfahrensdaten": { + "Instanzdaten": { + "Aktenzeichen": company_id.hr_number, + }, + "Beteiligung": [ + {}, + { + "Beteiligter": { + "Organisation": { + "Bezeichnung": { + "Bezeichnung_Aktuell": district_court.name + }, + "Sitz": { + "Ort": district_court.city, + }, + } + }, + }, + ], + }, + }, + } + } + result = transform.map_company_id(data) + assert result == company_id + + +def test_map_last_update() -> None: + date = "2024-01-01" + data = { + "XJustiz_Daten": {"Fachdaten_Register": {"Auszug": {"letzte_Eintragung": date}}} + } + result = transform.map_last_update(data) + assert result == date + + +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_company_id" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.name_from_beteiligung" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.loc_from_beteiligung" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_last_update" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_rechtsform" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_capital" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_business_purpose" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_founding_date" +) +@patch( + "aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.parse_stakeholder" +) +def test_map_unternehmensregister_json( # noqa: PLR0913 + mock_map_parse_stakeholder: Mock, + mock_map_founding_date: Mock, + mock_map_business_purpose: Mock, + mock_map_capital: Mock, + mock_map_rechtsform: Mock, + mock_map_last_update: Mock, + mock_loc_from_beteiligung: Mock, + mock_map_name_from_beteiligung: Mock, + mock_map_company_id: Mock, +) -> None: + expected_result = Company( + **{ + "id": Mock(), # type: ignore + "name": Mock(), + "location": Mock(), + "last_update": Mock(), + "company_type": Mock(), + "capital": Mock(), + "business_purpose": Mock(), + "founding_date": Mock(), + "relationships": [Mock()], + } + ) + + mock_map_company_id.return_value = expected_result.id + mock_map_name_from_beteiligung.return_value = expected_result.name + mock_loc_from_beteiligung.return_value = expected_result.location + mock_map_last_update.return_value = expected_result.last_update + mock_map_rechtsform.return_value = expected_result.company_type + mock_map_capital.return_value = expected_result.capital + mock_map_business_purpose.return_value = expected_result.business_purpose + mock_map_founding_date.return_value = expected_result.founding_date + mock_map_parse_stakeholder.return_value = expected_result.relationships[0] + + data: dict = { + "XJustiz_Daten": { + "Grunddaten": {"Verfahrensdaten": {"Beteiligung": [{}, {}, {}]}} + } + } + + result = transform.map_unternehmensregister_json(data) + assert result == expected_result