Update admins test

Rename function
OCHA-DAP · Jan 29, 2025 · 1c30c06 · 1c30c06
1 parent 6a08cb6
commit 1c30c06
Show file tree

Hide file tree

Showing 4 changed files with 61 additions and 24 deletions.
diff --git a/src/hapi/pipelines/database/admins.py b/src/hapi/pipelines/database/admins.py
@@ -202,7 +202,7 @@ def get_admin2_ref(
         return ref
 
     @classmethod
-    def get_max_admin_from_headers(cls, hxltag_to_header: Dict) -> int:
+    def get_max_admin_from_hxltags(cls, hxltag_to_header: Dict) -> int:
         max_admin_level = 0
         for hxltag in hxltag_to_header:
             match = cls.admin_name_regex.match(hxltag)

diff --git a/src/hapi/pipelines/database/humanitarian_needs.py b/src/hapi/pipelines/database/humanitarian_needs.py
@@ -49,7 +49,7 @@ def populate(self) -> None:
             url = resource["url"]
             headers, rows = reader.get_tabular_rows(url, dict_form=True)
             hxltag_to_header = invert_dictionary(next(rows))
-            max_admin_level = self._admins.get_max_admin_from_headers(
+            max_admin_level = self._admins.get_max_admin_from_hxltags(
                 hxltag_to_header
             )
             # Admin 1 PCode,Admin 2 PCode,Sector,Gender,Age Group,Disabled,Population Group,Population,In Need,Targeted,Affected,Reached

diff --git a/src/hapi/pipelines/database/operational_presence.py b/src/hapi/pipelines/database/operational_presence.py
@@ -44,7 +44,7 @@ def populate(self) -> None:
         url = resource["url"]
         headers, rows = reader.get_tabular_rows(url, dict_form=True)
         hxltag_to_header = invert_dictionary(next(rows))
-        max_admin_level = self._admins.get_max_admin_from_headers(
+        max_admin_level = self._admins.get_max_admin_from_hxltags(
             hxltag_to_header
         )
         resources_to_ignore = []

diff --git a/tests/test_main.py b/tests/test_main.py
@@ -122,88 +122,125 @@ def test_admin(self, configuration, folder, pipelines):
         count = session.scalar(select(func.count(DBAdmin2.id)))
         check.equal(count, 32102)
         admins = pipelines._admins
-        max_admin_level = admins.get_max_admin_from_headers(
+        max_admin_level = admins.get_max_admin_from_hxltags(
             [
-                "A",
-                "B",
-                "Admin 1 Name",
-                "c",
-                "123",
-                "Admin 3 Name",
-                "4",
-                "Admin 2 Name",
+                "#A",
+                "#B",
+                "adm1+name",
+                "#c",
+                "#lala5+name",
+                "#adm3+name",
+                "#e",
+                "#adm2+name",
             ]
         )
         check.equal(max_admin_level, 3)
         row = {"a": 1, "Country ISO3": "AFG"}
-        admin_level = admins.get_admin_level_from_row(row, max_admin_level)
+        hxltag_to_header = {
+            "#lala": "a",
+            "#country+code": "Country ISO3",
+            "#adm1+name": "",
+            "#adm2+name": "",
+            "#adm3+name": "",
+        }
+        admin_level = admins.get_admin_level_from_row(
+            hxltag_to_header, row, max_admin_level
+        )
         check.equal(admin_level, 0)
+        hxltag_to_header["#adm1+name"] = "Admin 1 Name"
         row = {"a": 1, "Country ISO3": "AFG", "Admin 1 Name": "ABC"}
-        admin_level = admins.get_admin_level_from_row(row, max_admin_level)
+        admin_level = admins.get_admin_level_from_row(
+            hxltag_to_header, row, max_admin_level
+        )
         check.equal(admin_level, 1)
+        hxltag_to_header["#adm2+name"] = "Admin 2 Name"
+        hxltag_to_header["#adm3+name"] = "Admin 3 Name"
         row = {
             "a": 1,
             "Country ISO3": "AFG",
             "Admin 3 Name": "ABC",
             "Admin 2 Name": "ABC",
         }
-        admin_level = admins.get_admin_level_from_row(row, max_admin_level)
+        admin_level = admins.get_admin_level_from_row(
+            hxltag_to_header, row, max_admin_level
+        )
         check.equal(admin_level, 3)
+        hxltag_to_header["#adm1+code"] = "Admin 1 PCode"
+        hxltag_to_header["#adm2+code"] = "Admin 2 PCode"
+        hxltag_to_header["#adm3+code"] = "Admin 3 PCode"
         row = {
             "a": 1,
             "Country ISO3": "AFG",
             "Admin 1 PCode": "",
             "Admin 2 PCode": "",
             "Admin 3 PCode": "",
         }
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 3)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 3
+        )
         check.equal(admin2_ref, None)
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AFG-XXX-XXX"
         row["Admin 1 Name"] = "ABC"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AFG-XXX-XXX"
         del row["Admin 1 Name"]
         row["Admin 1 PCode"] = "AF01"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AF01-XXX"
         row["Admin 1 Name"] = "ABC"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AF01-XXX"
         row["Admin 2 Name"] = "ABC"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AF01-XXX"
         del row["Admin 1 Name"]
         del row["Admin 2 Name"]
         row["Admin 2 PCode"] = "AF0101"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AF0101"
         row["Admin 1 Name"] = "ABC"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )
         assert code == "AF0101"
         row["Admin 2 Name"] = "ABC"
-        admin2_ref = admins.get_admin2_ref_from_row(row, "Test", "Test", 2)
+        admin2_ref = admins.get_admin2_ref_from_row(
+            hxltag_to_header, row, "Test", "Test", 2
+        )
         code = session.scalar(
             select(DBAdmin2.code).where(DBAdmin2.id == admin2_ref)
         )