[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[tor-commits] [ooni-probe/master] Add unittests for txscapy



commit 93f8f1626f4fee4519e7c8ee9c6da74c847135da
Author: Arturo Filastò <art@xxxxxxxxx>
Date:   Fri Jan 31 15:04:18 2014 +0100

    Add unittests for txscapy
---
 ooni/tests/test_txscapy.py |   46 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 46 insertions(+)

diff --git a/ooni/tests/test_txscapy.py b/ooni/tests/test_txscapy.py
new file mode 100644
index 0000000..337c399
--- /dev/null
+++ b/ooni/tests/test_txscapy.py
@@ -0,0 +1,46 @@
+from mock import MagicMock
+from twisted.internet import defer
+from twisted.trial import unittest
+
+from ooni.utils import txscapy
+
+class TestTxScapy(unittest.TestCase):
+    def setUp(self):
+        # if not txscapy.hasRawSocketPermission():
+        #     self.skipTest("No raw socket permissions...")
+        self.scapy_factory = txscapy.ScapyFactory('auto')
+        self.scapy_factory.super_socket = MagicMock()
+        self.scapy_factory.super_socket.ins.fileno.return_value = -1
+
+    def tearDown(self):
+        self.scapy_factory.loseConnection()
+
+    def test_pcapdnet_installed(self):
+        assert txscapy.pcapdnet_installed() == True
+    
+    def test_send_packet_no_answer(self):
+        from scapy.all import IP, TCP
+        sender = txscapy.ScapySender()
+        self.scapy_factory.registerProtocol(sender)
+        packet = IP(dst='8.8.8.8')/TCP(dport=53)
+        sender.startSending([packet])
+        self.scapy_factory.super_socket.send.assert_called_with(packet)
+        assert len(sender.sent_packets) == 1
+    
+    @defer.inlineCallbacks
+    def test_send_packet_with_answer(self):
+        from scapy.all import IP, TCP
+        sender = txscapy.ScapySender()
+        self.scapy_factory.registerProtocol(sender)
+
+        packet_sent = IP(dst='8.8.8.8',src='127.0.0.1')/TCP(dport=53, sport=5300)
+        packet_received = IP(dst='127.0.0.1', src='8.8.8.8')/TCP(sport=53, dport=5300)
+        
+        d = sender.startSending([packet_sent])
+        self.scapy_factory.super_socket.send.assert_called_with(packet_sent)
+
+        sender.packetReceived(packet_received)
+
+        result = yield d
+        assert result[0][0][0] == packet_sent
+        assert result[0][0][1] == packet_received



_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits