More tests

test_urlchecker_http.py

@@ -8,7 +8,7 @@ import mock
 import peewee
 
 
-class UrlCheckerStatusTestCase(unittest.TestCase):
+class UrlCheckerHttpTestCase(unittest.TestCase):
     def setUp(self):
         self.db = LogDB(":memory:")
         self.db.createTables()
@@ -306,7 +306,7 @@ class UrlCheckerStatusTestCase(unittest.TestCase):
 
 def suite():
     suite = unittest.TestSuite()
-    suite.addTest(unittest.makeSuite(UrlCheckerStatusTestCase))
+    suite.addTest(unittest.makeSuite(UrlCheckerHttpTestCase))
     return suite
 
 

test_urlchecker_network.py

+import unittest
+from urlchecker_db import LogDB
+import urlchecker_network
+from urlchecker_network import logNetwork, isTcpPortOpen
+from urlchecker_status import logStatus
+import mock
+import peewee
+
+
+class UrlCheckerNetworkTestCase(unittest.TestCase):
+    def setUp(self):
+        self.db = LogDB(":memory:")
+        self.db.createTables()
+
+    ################################################
+    # logNetwork
+    ################################################
+    def test_logNetwork_insertFirst(self):
+        ip = "127.0.0.1"
+        port = 1234
+        transport = "foobar"
+        state = "bar"
+        status_id = logStatus(self.db, "foo")
+
+        result = logNetwork(self.db, ip, transport, port, state, status_id)
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == transport
+        assert self.db.NetworkChange.get().state == state
+        assert self.db.NetworkChange.get().status_id == status_id
+
+    def test_logNetwork_insertOnlyOnePerStatusIdIPPortTransport(self):
+        ip = "127.0.0.1"
+        port = 1234
+        transport = "foobar"
+        state = "bar"
+        status_id = logStatus(self.db, "foo")
+
+        result = logNetwork(self.db, ip, transport, port, state, status_id)
+        try:
+          logNetwork(self.db, ip, transport, port, state + '.', status_id)
+        except peewee.IntegrityError:
+          assert self.db.NetworkChange.select().count() == 1
+          assert self.db.NetworkChange.get().status_id == status_id
+        else:
+          raise NotImplementedError('Expected IntegrityError')
+
+    def test_logNetwork_skipIdenticalPreviousValues(self):
+        ip = "127.0.0.1"
+        port = 1234
+        transport = "foobar"
+        state = "bar"
+        status_id = logStatus(self.db, "foo")
+        result = logNetwork(self.db, ip, transport, port, state, status_id)
+
+        status_id_2 = logStatus(self.db, "foo")
+        result_2 = logNetwork(self.db, ip, transport, port, state, status_id_2)
+        assert result_2 == result
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == transport
+        assert self.db.NetworkChange.get().state == state
+        assert self.db.NetworkChange.get().status_id == status_id
+
+    def test_logNetwork_insertWhenDifferentState(self):
+        ip = "127.0.0.1"
+        port = 1234
+        transport = "foobar"
+        state = "bar"
+        status_id = logStatus(self.db, "foo")
+        result = logNetwork(self.db, ip, transport, port, state, status_id)
+
+        state_2 = state + "."
+        status_id_2 = logStatus(self.db, "foo")
+        result_2 = logNetwork(self.db, ip, transport, port, state_2, status_id_2)
+        assert result_2 != result
+        assert self.db.NetworkChange.select().count() == 2
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id).ip == ip
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id).port == port
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id).transport == transport
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id).state == state
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id).status_id == status_id
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id_2).ip == ip
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id_2).port == port
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id_2).transport == transport
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id_2).state == state_2
+        assert self.db.NetworkChange.get(self.db.NetworkChange.status == status_id_2).status_id == status_id_2
+
+    def test_logNetwork_insertDifferentKeys(self):
+        ip = "127.0.0.1"
+        ip_2 = ip + "2"
+        port = 1234
+        port_2 = port + 1
+        transport = "foobar"
+        transport_2 = transport + '.'
+        state = "bar"
+        status_id = logStatus(self.db, "foo")
+
+        logNetwork(self.db, ip, transport, port, state, status_id)
+        logNetwork(self.db, ip_2, transport, port, state, status_id)
+        logNetwork(self.db, ip, transport_2, port, state, status_id)
+        logNetwork(self.db, ip, transport, port_2, state, status_id)
+        logNetwork(self.db, ip_2, transport_2, port, state, status_id)
+        logNetwork(self.db, ip_2, transport, port_2, state, status_id)
+        logNetwork(self.db, ip, transport_2, port_2, state, status_id)
+        logNetwork(self.db, ip_2, transport_2, port_2, state, status_id)
+
+        assert self.db.NetworkChange.select().count() == 8
+
+    ################################################
+    # isTcpPortOpen
+    ################################################
+    def test_isTcpPortOpen_open(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          result = isTcpPortOpen(self.db, ip, port, status_id)
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == 'TCP'
+        assert self.db.NetworkChange.get().state == 'open'
+        assert self.db.NetworkChange.get().status_id == status_id
+        assert result == True
+
+    def test_isTcpPortOpen_connectionRefused(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          def sideEffect(*args, **kw):
+            raise ConnectionRefusedError()
+          mock_socket.return_value.connect.side_effect = sideEffect
+
+          result = isTcpPortOpen(self.db, ip, port, status_id)
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == 'TCP'
+        assert self.db.NetworkChange.get().state == 'closed'
+        assert self.db.NetworkChange.get().status_id == status_id
+        assert result == False
+
+    def test_isTcpPortOpen_timeout(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          def sideEffect(*args, **kw):
+            raise urlchecker_network.socket.timeout()
+          mock_socket.return_value.connect.side_effect = sideEffect
+
+          result = isTcpPortOpen(self.db, ip, port, status_id)
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == 'TCP'
+        assert self.db.NetworkChange.get().state == 'filtered'
+        assert self.db.NetworkChange.get().status_id == status_id
+        assert result == False
+
+    def test_isTcpPortOpen_noRoute(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          def sideEffect(*args, **kw):
+            raise OSError(urlchecker_network.errno.EHOSTUNREACH, 'foo')
+          mock_socket.return_value.connect.side_effect = sideEffect
+
+          result = isTcpPortOpen(self.db, ip, port, status_id)
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == 'TCP'
+        assert self.db.NetworkChange.get().state == 'filtered'
+        assert self.db.NetworkChange.get().status_id == status_id
+        assert result == False
+
+    def test_isTcpPortOpen_noNetwork(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          def sideEffect(*args, **kw):
+            raise OSError(urlchecker_network.errno.ENETUNREACH, 'foo')
+          mock_socket.return_value.connect.side_effect = sideEffect
+
+          result = isTcpPortOpen(self.db, ip, port, status_id)
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+        assert self.db.NetworkChange.select().count() == 1
+        assert self.db.NetworkChange.get().ip == ip
+        assert self.db.NetworkChange.get().port == port
+        assert self.db.NetworkChange.get().transport == 'TCP'
+        assert self.db.NetworkChange.get().state == 'unreachable'
+        assert self.db.NetworkChange.get().status_id == status_id
+        assert result == False
+
+    def test_isTcpPortOpen_OSError(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          def sideEffect(*args, **kw):
+            raise OSError()
+          mock_socket.return_value.connect.side_effect = sideEffect
+
+          try:
+            isTcpPortOpen(self.db, ip, port, status_id)
+          except OSError:
+            assert self.db.NetworkChange.select().count() == 0
+          else:
+            raise NotImplementedError('Expected OSError')
+
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+    def test_isTcpPortOpen_Exception(self):
+        ip = "127.0.0.1"
+        port = 1234
+        status_id = logStatus(self.db, "foo")
+
+        with mock.patch("urlchecker_network.socket.socket") as mock_socket:
+          def sideEffect(*args, **kw):
+            raise Exception()
+          mock_socket.return_value.connect.side_effect = sideEffect
+
+          try:
+            isTcpPortOpen(self.db, ip, port, status_id)
+          except Exception:
+            assert self.db.NetworkChange.select().count() == 0
+          else:
+            raise NotImplementedError('Expected OSError')
+
+          assert mock_socket.call_count == 1
+
+          assert mock_socket.return_value.settimeout.call_count == 1
+          mock_socket.return_value.settimeout.assert_called_with(2)
+
+          assert mock_socket.return_value.connect.call_count == 1
+          mock_socket.return_value.connect.assert_called_with((ip, port))
+
+          assert mock_socket.return_value.close.call_count == 1
+
+def suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(UrlCheckerNetworkTestCase))
+    return suite
+
+
+if __name__ == "__main__":
+    unittest.main(defaultTest="suite")

urlchecker_db.py

@@ -43,12 +43,12 @@ class LogDB:
         # Store remote network status
         class NetworkChange(BaseModel):
             status = peewee.ForeignKeyField(Status)
-            ip = peewee.TextField(index=True)
+            ip = peewee.TextField()
             transport = peewee.TextField()
             port = peewee.IntegerField()
             state = peewee.TextField()
-            # class Meta:
-            #     primary_key = peewee.CompositeKey("status", "ip", "transport", "port")
+            class Meta:
+                primary_key = peewee.CompositeKey("status", "ip", "transport", "port")
 
         class DnsChange(BaseModel):
             status = peewee.ForeignKeyField(Status)
@@ -59,8 +59,8 @@ class LogDB:
 
         class HttpCodeChange(BaseModel):
             status = peewee.ForeignKeyField(Status)
-            ip = peewee.TextField(index=True)
-            url = peewee.TextField(index=True)
+            ip = peewee.TextField()
+            url = peewee.TextField()
             status_code = peewee.IntegerField()
             class Meta:
                 primary_key = peewee.CompositeKey("status", "ip", "url")

urlchecker_network.py

@@ -30,7 +30,7 @@ def logNetwork(db, ip, transport, port, state, status_id):
                 port=port,
                 state=state,
             )
-        return previous_entry.id
+        return previous_entry.status
 
 
 def isTcpPortOpen(db, ip, port, status_id):
@@ -53,7 +53,11 @@ def isTcpPortOpen(db, ip, port, status_id):
             # OSError: [Errno 101] Network is unreachable
             state = "unreachable"
         else:
+            sock.close()
             raise
+    except:
+      sock.close()
+      raise
 
     sock.close()