Whois: handle connection error.

src/surykatka/domain.py

@@ -125,31 +125,36 @@ def queryWhois(db, status_id, domain_text):
     # Error trying to connect to socket: closing socket
     _stdout = sys.stdout
     sys.stdout = open(os.devnull, "w")
-    whois_dict = whois.whois(domain_text)
-    sys.stdout = _stdout
-
-    arg_list = []
-    for arg in [
-        whois_dict.registrar,
-        whois_dict.whois_server,
-        whois_dict.creation_date,
-        whois_dict.updated_date,
-        whois_dict.expiration_date,
-        whois_dict.name_servers,
-        whois_dict.status,
-        whois_dict.emails,
-        whois_dict.dnssec,
-        whois_dict.name,
-        whois_dict.org,
-        whois_dict.address,
-        whois_dict.city,
-        whois_dict.state,
-        whois_dict.zipcode,
-        whois_dict.country,
-    ]:
-        if type(arg) == list:
-            arg = arg[0]
-        arg_list.append(arg)
+    try:
+        whois_dict = whois.whois(domain_text)
+    except ConnectionResetError:
+        arg_list = [""] * 16
+        whois_dict = {}
+    else:
+        arg_list = []
+        for arg in [
+            whois_dict.registrar,
+            whois_dict.whois_server,
+            whois_dict.creation_date,
+            whois_dict.updated_date,
+            whois_dict.expiration_date,
+            whois_dict.name_servers,
+            whois_dict.status,
+            whois_dict.emails,
+            whois_dict.dnssec,
+            whois_dict.name,
+            whois_dict.org,
+            whois_dict.address,
+            whois_dict.city,
+            whois_dict.state,
+            whois_dict.zipcode,
+            whois_dict.country,
+        ]:
+            if type(arg) == list:
+                arg = arg[0]
+            arg_list.append(arg)
+    finally:
+        sys.stdout = _stdout
 
     logWhoisQuery(db, status_id, domain_text, *arg_list)
     return whois_dict

tests/test_domain.py

@@ -847,6 +847,43 @@ class SurykatkaDomainTestCase(unittest.TestCase):
         assert self.db.DomainChange.get().status_id == status_id
         assert result == mock_answer
 
+    def test_queryWhois_ConnectionResetError(self):
+        domain = "http://example.org"
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("surykatka.domain.whois.whois") as mock_whois:
+
+            def sideEffect(*args, **kw):
+                raise ConnectionResetError()
+
+            mock_whois.side_effect = sideEffect
+
+            result = queryWhois(self.db, status_id, domain)
+
+            assert mock_whois.call_count == 1
+            mock_whois.assert_called_with(domain)
+
+        assert self.db.DomainChange.select().count() == 1
+        assert self.db.DomainChange.get().domain == domain
+        assert self.db.DomainChange.get().registrar == ""
+        assert self.db.DomainChange.get().whois_server == ""
+        assert self.db.DomainChange.get().creation_date == ""
+        assert self.db.DomainChange.get().updated_date == ""
+        assert self.db.DomainChange.get().expiration_date == ""
+        assert self.db.DomainChange.get().name_servers == ""
+        assert self.db.DomainChange.get().whois_status == ""
+        assert self.db.DomainChange.get().emails == ""
+        assert self.db.DomainChange.get().dnssec == ""
+        assert self.db.DomainChange.get().name == ""
+        assert self.db.DomainChange.get().org == ""
+        assert self.db.DomainChange.get().address == ""
+        assert self.db.DomainChange.get().city == ""
+        assert self.db.DomainChange.get().state == ""
+        assert self.db.DomainChange.get().zipcode == ""
+        assert self.db.DomainChange.get().country == ""
+        assert self.db.DomainChange.get().status_id == status_id
+        assert result == {}
+
     ################################################
     # packDns
     ################################################