fix validators and start writing tests

IsaacMilarky · IsaacMilarky · commit d28fcd4c8034 · 2025-10-16T15:31:32.000-05:00
Signed-off-by: Isaac Milarsky &lt;imilarsky@gmail.com&gt;
diff --git a/backend/npdfhir/tests.py b/backend/npdfhir/tests.py
@@ -2,10 +2,17 @@
 from rest_framework import status
 from rest_framework.test import APITestCase, APIClient
 from django.test.runner import DiscoverRunner
+from django.test import TestCase
 from django.db import connection
 from .cache import cacheData # I can't explain why, but we need to import cacheData here. I think we can remove this once we move to the docker db setup
 from fhir.resources.bundle import Bundle
 from pydantic import ValidationError
+from .models import Nucc, C80PracticeCodes
+from .validators import NPDValueSet, NPDPractitioner
+from fhir.resources.valueset import ValueSet
+from fhir.resources.practitioner import Practitioner
+
+
 
 def get_female_npis(npi_list):
     """
@@ -249,3 +256,98 @@ def test_retrieve_nonexistent(self):
         url = reverse("fhir-practitioner-detail", args=[999999])
         response = self.client.get(url)
         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+
+class NPDValueSetValidatorTests(TestCase):
+
+    def test_verify_codes_valid(self):
+        """✅ Should pass when all codes exist in the DB."""
+
+        data = {
+            "resourceType" : "ValueSet",
+            "id" : "FHIR-version",
+            "meta" : {
+                "lastUpdated" : "2025-10-16T16:45:52.699+00:00",
+                "profile" : ["http://hl7.org/fhir/StructureDefinition/shareablevalueset"]
+            },
+            "text" : {
+                "status" : "generated",
+                "div" : "<div xmlns=\"http://www.w3.org/1999/xhtml\"><p class=\"res-header-id\"><b>Generated Narrative: ValueSet FHIR-version</b></p><a name=\"FHIR-version\"> </a><a name=\"hcFHIR-version\"> </a><div style=\"display: inline-block; background-color: #d9e0e7; padding: 6px; margin: 4px; border: 1px solid #8da1b4; border-radius: 5px; line-height: 60%\"><p style=\"margin-bottom: 0px\"/><p style=\"margin-bottom: 0px\">Profile: <a href=\"shareablevalueset.html\">Shareable ValueSet</a></p></div><ul><li>Include all codes defined in <a href=\"codesystem-FHIR-version.html\"><code>http://hl7.org/fhir/FHIR-version</code></a> version <span title=\"Version is not explicitly stated, which means it is fixed to the version provided in this specification\">&#x1F4E6;6.0.0-ballot3</span></li></ul></div>"
+            },
+            "extension" : [{
+                "url" : "http://hl7.org/fhir/StructureDefinition/structuredefinition-wg",
+                "valueCode" : "fhir"
+            },
+            {
+                "url" : "http://hl7.org/fhir/StructureDefinition/structuredefinition-standards-status",
+                "valueCode" : "normative"
+            },
+            {
+                "url" : "http://hl7.org/fhir/StructureDefinition/structuredefinition-normative-version",
+                "valueCode" : "4.0.0"
+            },
+            {
+                "url" : "http://hl7.org/fhir/StructureDefinition/structuredefinition-fmm",
+                "valueInteger" : 5
+            }],
+            "url" : "http://hl7.org/fhir/ValueSet/FHIR-version",
+            "identifier" : [{
+                "system" : "urn:ietf:rfc:3986",
+                "value" : "urn:oid:2.16.840.1.113883.4.642.3.1309"
+            }],
+            "version" : "6.0.0-ballot3",
+            "name" : "FHIRVersion",
+            "title" : "FHIRVersion",
+            "status" : "active",
+            "experimental" : False,
+            "date" : "2025-10-16T16:45:52+00:00",
+            "publisher" : "HL7 International / FHIR Infrastructure",
+            "contact" : [{
+                "telecom" : [{
+                "system" : "url",
+                "value" : "http://www.hl7.org/Special/committees/fiwg"
+                }]
+            }],
+            "description" : "All published FHIR Versions.",
+            "jurisdiction" : [{
+                "coding" : [{
+                "system" : "http://unstats.un.org/unsd/methods/m49/m49.htm",
+                "code" : "001",
+                "display" : "World"
+                }]
+            }],
+            "immutable" : True,
+            "compose": {
+                "include": [
+                    {
+                        "system": "http://nucc.org/provider-taxonomy",
+                        "concept": [{"code": "101Y00000X"}],
+                    },
+                    {
+                        "system": "http://snomed.info/sct",
+                        "concept": [{"code": "419772000"}],
+                    },
+                ]
+            },
+        }
+
+        model = NPDValueSet(**data)
+
+        self.assertIsInstance(model, ValueSet)
+
+    def test_verify_codes_invalid(self):
+        """🚫 Should raise ValueError when a code is not found."""
+        data = {
+            "resourceType": "ValueSet",
+            "compose": {
+                "include": [
+                    {
+                        "system": "http://nucc.org/provider-taxonomy",
+                        "concept": [{"code": "BAD_CODE"}],
+                    }
+                ]
+            },
+        }
+
+        with self.assertRaises(ValidationError) as ctx:
+            NPDValueSet(**data)
diff --git a/backend/npdfhir/validators.py b/backend/npdfhir/validators.py
@@ -9,144 +9,146 @@
 from fhir.resources.valueset import ValueSet
 from fhir.resources.practitioner import Practitioner
 from pydantic import ValidationError
-from pydantic.functional_validators import model_validator
+from pydantic import model_validator
 from django.db import connection
 from .models import Nucc, C80PracticeCodes
 
 
-@model_validator(mode="after")
-def verify_codes(self):
-    """
-    Verifies the code items present in the valueset 
-
-    Args:
-        self (ValueSet): Pydantic Object representing a FHIR ValueSet
+class NPDValueSet(ValueSet):
+    @model_validator(mode="after")
+    def verify_codes(self):
+        """
+        Verifies the code items present in the valueset 
 
-    Raises:
-        ValidationError: raises a validationError if the codes are not part of a 
-        known valid code
-    """
+        Args:
+            self (ValueSet): Pydantic Object representing a FHIR ValueSet
 
-    def get_nucc_codes_from_db():
-        """
-        Function that fetches nucc codes from the db
+        Raises:
+            ValidationError: raises a validationError if the codes are not part of a 
+            known valid code
         """
 
-        data = dict(
-            Nucc.objects.values_list("code", "display_name")
-        )
-
-        return data
+        def get_nucc_codes_from_db():
+            """
+            Function that fetches nucc codes from the db
+            """
 
+            data = dict(
+                Nucc.objects.values_list("code", "display_name")
+            )
 
-    def get_c80_codes_from_db():
-        """
-        Function that fetches c80 codes from db
-        """
+            return data
 
-        data = dict(
-            C80PracticeCodes.objects.values_list("code", "display_name")
-        )
 
-        return data
+        def get_c80_codes_from_db():
+            """
+            Function that fetches c80 codes from db
+            """
 
-    nucc_codes = get_nucc_codes_from_db()
-    c80_codes = get_c80_codes_from_db()
+            data = dict(
+                C80PracticeCodes.objects.values_list("code", "display_name")
+            )
 
-    for include_index, code_item in enumerate(self.compose['include']):
-        #Check if code in ValueSet is part of nucc
+            return data
 
-        for concept_index, code_value in enumerate(code_item['concept']):
-            try:
-                if "nucc.org/provider-taxonomy" in code_item['system']:
-                    assert code_value['code'] in nucc_codes
-                elif "snomed.info/sct" in code_item['system']:
-                    assert code_value['code'] in c80_codes
-            except AssertionError as e:
-                raise ValueError(
-                    f"Code {code_value['code']} is not a valid {code_value['system']} code!"
-                    f"(at include[{include_index}].concept[{concept_index}])"
-                ) from e
-    
-    return self
+        nucc_codes = get_nucc_codes_from_db()
+        c80_codes = get_c80_codes_from_db()
 
-#ValueSet.add_root_validator(verify_codes, pre=False)
-ValueSet.model_rebuild(validators=[verify_codes])
+        for include_index, code_item in enumerate(self.compose.include):
+            #Check if code in ValueSet is part of nucc
 
-@model_validator(mode="after")
-def verify_identifier_code_from_data(self):
-    """
-    Iterates through json data and verifies the identifiers present
+            for concept_index, code_value in enumerate(code_item.concept):
+                try:
+                    if "nucc.org/provider-taxonomy" in code_item.system:
+                        assert code_value.code in nucc_codes
+                    elif "snomed.info/sct" in code_item.system:
+                        assert code_value.code in c80_codes
+                except AssertionError as e:
+                    raise ValueError(
+                        f"Code {code_value.code} is not a valid {code_item.system} code!"
+                        f"(at include[{include_index}].concept[{concept_index}])"
+                    ) from e
+        
+        return self
 
-    Args:
-        data (dict): json dictionary representing fhir data
-    """
+#ValueSet.add_root_validator(verify_codes, pre=False)
+#ValueSet.model_rebuild(validators=[verify_codes])
 
-    def verify_npi(npi,position):
+class NPDPractitioner(Practitioner):
+    @model_validator(mode="after")
+    def verify_identifier_code_from_data(self):
         """
-        Checks the npi code against various checks and throws errors if they fail.
+        Iterates through json data and verifies the identifiers present
 
         Args:
-            npi (str): npi identifier string
-            position (int): index where npi is found in the json
-
-        Raises:
-            ValidationError: Error thrown when npi value is out of range
-            ValidationError: Error thrown when npi value fails luhn algo
+            data (dict): json dictionary representing fhir data
         """
 
-        def is_valid_npi_format(npi_value):
+        def verify_npi(npi,position):
             """
-            Checks npi value string for correct range and digits
+            Checks the npi code against various checks and throws errors if they fail.
 
             Args:
-                npi_value (str): npi identifier string
+                npi (str): npi identifier string
+                position (int): index where npi is found in the json
 
-            Returns:
-                bool: True if value is valid
+            Raises:
+                ValidationError: Error thrown when npi value is out of range
+                ValidationError: Error thrown when npi value fails luhn algo
             """
 
-            digits_only = re.sub(r'\D', '', str(npi_value))
-            npi_num = int(digits_only)
-            return 999999999 <= npi_num <= 10000000000
+            def is_valid_npi_format(npi_value):
+                """
+                Checks npi value string for correct range and digits
 
-        #NPI luhn algo defined here:
-        #https://www.cms.gov/Regulations-and-Guidance/Administrative-Simplification/NationalProvIdentStand/Downloads/NPIcheckdigit.pdf
-        def npi_check_luhn_algorithm(npi_value):
-            """
-            Checks npi value string based on given luhn algorithm
+                Args:
+                    npi_value (str): npi identifier string
 
-            Transforms every other digit and sums them together plus 24
+                Returns:
+                    bool: True if value is valid
+                """
 
-            If the result is divisible by 10 then the npi value is valid
+                digits_only = re.sub(r'\D', '', str(npi_value))
+                npi_num = int(digits_only)
+                return 999999999 <= npi_num <= 10000000000
 
-            Args:
-                npi_value (str): npi identifier string
-            """
+            #NPI luhn algo defined here:
+            #https://www.cms.gov/Regulations-and-Guidance/Administrative-Simplification/NationalProvIdentStand/Downloads/NPIcheckdigit.pdf
+            def npi_check_luhn_algorithm(npi_value):
+                """
+                Checks npi value string based on given luhn algorithm
+
+                Transforms every other digit and sums them together plus 24
+
+                If the result is divisible by 10 then the npi value is valid
+
+                Args:
+                    npi_value (str): npi identifier string
+                """
 
-            def transform(d):
-                return d * 2 if d < 5 else d * 2 - 9
+                def transform(d):
+                    return d * 2 if d < 5 else d * 2 - 9
 
-            digits_only = re.sub(r'\D', '', str(npi_value))
+                digits_only = re.sub(r'\D', '', str(npi_value))
 
-            digits = [int(ch) for ch in digits_only[:10]]
+                digits = [int(ch) for ch in digits_only[:10]]
 
-            total = sum(
-                transform(d) if i % 2 == 0 else d
-                for i, d in enumerate(digits)
-            ) + 24
+                total = sum(
+                    transform(d) if i % 2 == 0 else d
+                    for i, d in enumerate(digits)
+                ) + 24
 
-            return total % 10 == 0
+                return total % 10 == 0
 
-        if not is_valid_npi_format(npi):
-            raise ValueError(f"NPI '{npi}' (identifier[{position}]) invalid format")
-        if not npi_check_luhn_algorithm(npi):
-            raise ValueError(f"NPI '{npi}' (identifier[{position}]) failed Luhn check")
+            if not is_valid_npi_format(npi):
+                raise ValueError(f"NPI '{npi}' (identifier[{position}]) invalid format")
+            if not npi_check_luhn_algorithm(npi):
+                raise ValueError(f"NPI '{npi}' (identifier[{position}]) failed Luhn check")
 
-    for identifier_index, identifier in enumerate(self.identifier):
-        if "hl7.org/fhir/sid/us-npi" in identifier['system']:
-            verify_npi(identifier['value'],identifier_index)
-    
-    return self
+        for identifier_index, identifier in enumerate(self.identifier):
+            if "hl7.org/fhir/sid/us-npi" in identifier['system']:
+                verify_npi(identifier['value'],identifier_index)
+        
+        return self
 
-Practitioner.model_rebuild(validators=[verify_identifier_code_from_data])
+#Practitioner.model_rebuild(validators=[verify_identifier_code_from_data])
