Skip Iris results without a risk score instead of raising TypeError

domaintools/filters.py

@@ -142,7 +142,12 @@ def __call__(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
 
         filtered_result = []
         for result in self._results:
-            domain_risk_score = result.get("domain_risk", {}).get("risk_score")
+            domain_risk = result.get("domain_risk") or {}
+            domain_risk_score = domain_risk.get("risk_score")
+            if domain_risk_score is None:
+                # skip uncomparable risk score
+                continue
+
             if domain_risk_score > self._threshold:
                 filtered_result.append(result)
 

tests/test_api.py

@@ -465,6 +465,32 @@ def test_iris_investigate():
         assert result["domain"] in ["amazon.com", "google.com"]
 
 
+def test_iris_investigate_risk_score_threshold_with_scoreless_result():
+    """A result without a `domain_risk` block must be skipped, not blow up the filtering."""
+    scored_domain = iris_investigate_fixture()["results"][0]
+    scored_domain["domain_risk"] = {"risk_score": 80}
+
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {
+        "response": {
+            "results": [{"domain": "no-domain-risk.com"}, scored_domain],
+            "results_count": 2,
+        }
+    }
+
+    with patch("domaintools.base_results.Client") as mock_client:
+        mock_client.return_value.__enter__.return_value.post.return_value = mock_response
+
+        investigation_results = api.iris_investigate(
+            domains=["no-domain-risk.com", scored_domain["domain"]],
+            risk_score_threshold=70,
+        )
+
+        assert investigation_results["results_count"] == 1
+        assert investigation_results["results"][0]["domain"] == scored_domain["domain"]
+
+
 def test_iris_investigate_irisql_calls_results_with_irisql():
     query = "# IrisQL-1.0\nDOMAIN CONTAINS \"phishing\""
     with patch.object(api, "_results") as mock_results:

tests/test_filters.py

@@ -16,7 +16,9 @@ def setUp(self):
         domaintools_iris_result = iris_investigate_data.domaintools()
         int_chase_iris_result = iris_investigate_data.int_chase()
 
-        test_results = {"results": domaintools_iris_result["results"] + int_chase_iris_result["results"]}
+        self.results = domaintools_iris_result["results"] + int_chase_iris_result["results"]
+
+        test_results = {"results": self.results}
 
         self.dt_res_filter = DTResultFilter(result_set=test_results)
 
@@ -33,6 +35,32 @@ def test_filter_by_riskscore(self):
         assert len(result) == 1
         assert result[0]["domain"] == "int-chase.com"
 
+    def test_filter_by_riskscore_skips_results_without_risk_score(self):
+        """Test that results carrying no usable risk score are skipped instead of raising"""
+        risk_score_threshold = 69
+
+        scoreless_results = {
+            "results": self.results
+            + [
+                {"domain": "no-domain-risk.com"},
+                {"domain": "null-domain-risk.com", "domain_risk": None},
+                {
+                    "domain": "no-risk-score.com",
+                    "domain_risk": {"components": [{"name": "zerolist", "risk_score": 0}]},
+                },
+                {"domain": "null-risk-score.com", "domain_risk": {"risk_score": None}},
+            ]
+        }
+
+        result = DTResultFilter(result_set=scoreless_results).by(
+            [
+                filter_by_riskscore(threshold=risk_score_threshold),
+            ]
+        )
+
+        assert len(result) == 1
+        assert result[0]["domain"] == "int-chase.com"
+
     def test_filter_younger_than_expire_date(self):
         """Test filter result younger than the expire date"""
         younger_than = "2024-02-24"