1 files changed, 126 insertions, 0 deletions
diff --git a/tests/test_contacts.py b/tests/test_contacts.py
index a6ddabe..0ac197d 100644
--- a/tests/test_contacts.py
+++ b/tests/test_contacts.py
@@ -109,5 +109,131 @@ class Worklist(unittest.TestCase):
         self.assertEqual(contacts.worklist(iocs), [])
 
 
+class Resolve(unittest.TestCase):
+    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
+    IPV6 = {"services": []}
+    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}
+
+    def _bootstraps(self):
+        return {"ipv4": self.IPV4, "ipv6": self.IPV6, "dns": self.DNS}
+
+    def test_an_ip_resolves_to_its_abuse_desk(self):
+        def fetch(url):
+            return {
+                "handle": "NET-1",
+                "entities": [{
+                    "roles": ["abuse"],
+                    "vcardArray": ["vcard", [
+                        ["version", {}, "text", "4.0"],
+                        ["email", {}, "text", "abuse@example.invalid"],
+                    ]],
+                }],
+            }
+
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["iocs"], ["ioc-1"])
+        self.assertEqual(result[0]["query"], "198.51.100.7")
+        self.assertEqual(result[0]["abuse"], ["abuse@example.invalid"])
+        self.assertEqual(result[0]["handle"], "NET-1")
+        self.assertNotIn("error", result[0])
+
+    def test_no_abuse_role_records_a_reason_not_an_error(self):
+        def fetch(url):
+            return {"handle": "NET-2", "entities": []}
+
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(result[0]["abuse"], [])
+        self.assertEqual(result[0]["error"], "no abuse role published")
+
+    def test_no_rdap_server_records_a_reason(self):
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        iocs = [{"id": "ioc-1", "type": "domain", "value": "example.test"}]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(result[0]["abuse"], [])
+        self.assertIn("no rdap server", result[0]["error"])
+
+    def test_a_network_failure_is_per_query_and_does_not_stop_the_run(self):
+        def fetch(url):
+            if "198.51.100.7" in url:
+                raise OSError("connection timed out")
+            return {
+                "handle": "DOM-1",
+                "entities": [{
+                    "roles": ["abuse"],
+                    "vcardArray": ["vcard", [
+                        ["version", {}, "text", "4.0"],
+                        ["email", {}, "text", "abuse@example.invalid"],
+                    ]],
+                }],
+            }
+
+        iocs = [
+            {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"},
+            {"id": "ioc-2", "type": "domain", "value": "example.invalid"},
+        ]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(len(result), 2)
+        failed = [r for r in result if r["query"] == "198.51.100.7"][0]
+        worked = [r for r in result if r["query"] == "example.invalid"][0]
+        self.assertIn("connection timed out", failed["error"])
+        self.assertEqual(worked["abuse"], ["abuse@example.invalid"])
+
+    def test_one_query_per_host_however_many_iocs(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "DOM-1", "entities": []}
+
+        iocs = [
+            {"id": f"ioc-{n}", "type": "url",
+             "value": f"https://a.example.invalid/page{n}"}
+            for n in range(20)
+        ]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(len(result[0]["iocs"]), 20)
+
+    def test_no_query_ever_carries_a_path(self):
+        """THE FOURTH PROPERTY, asserted at the transport."""
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "DOM-1", "entities": []}
+
+        iocs = [{
+            "id": "ioc-1", "type": "url",
+            "value": "https://a.example.invalid/verify/victim%40example.org?e=x",
+        }]
+        contacts.resolve(iocs, bootstraps=self._bootstraps(), fetch=fetch)
+
+        for url in calls:
+            self.assertNotIn("victim", url)
+            self.assertNotIn("verify", url)
+            self.assertNotIn("%40", url)
+            self.assertNotIn("?", url)
+
+
 if __name__ == "__main__":
     unittest.main()