Changeset 4870

Timestamp:

06/02/09 11:15:41 (4 years ago)

Author:

getxsick

Message:

umpa.sniffing.from_file() splitted to 2 functions (from_file() and
from_file_loop()). Now API is more unified (the same is for sniff()
and sniff_loop()).
add failing test and fix for the bug (handling func arguments (callback_args))

Location:

branch/UMPA

Files:

• tests/unit/test_sniffing/test_sniffing_init.py (modified) (3 diffs)
• umpa/sniffing/__init__.py (modified) (6 diffs)

branch/UMPA/tests/unit/test_sniffing/test_sniffing_init.py

@@ -75 +75 @@
             if args[0] > args[1]:
                 raise UMPASniffingException("test")
+        
+        def cbk2(ts, pkt, *args):
+            print pkt
+            
+        th = SendPacket(umpa.Packet(IP(source_address="1.2.3.6"),
+                                    TCP(source_port=99)))
+        th.start()
+        umpa.sniffing.sniff_loop(1, filter="src 1.2.3.6", device='any',
+                                            callback=cbk, callback_args=[1,2])
+        th.join()
 
         th = SendPacket(umpa.Packet(IP(source_address="1.2.3.6"),
@@ -80 +90 @@
         th.start()
         umpa.sniffing.sniff_loop(1, filter="src 1.2.3.6", device='any',
-                                            callback=cbk, callback_args=[1,2])
+                                                            callback=cbk2)
         th.join()
 
@@ -93 +103 @@
     def test_from_file(self):
         py.test.skip("implement to_file first")
+
+    def test_from_file_loop(self):
+        py.test.skip("implement to_file first")

branch/UMPA/umpa/sniffing/__init__.py

@@ -106 +106 @@
 
 def sniff_loop(count=0, filter=None, device=None, timeout=0, snaplen=1024,
-                        promisc=True, callback=None, callback_args=None):
+                            promisc=True, callback=None, callback_args=None):
     """
     Sniff packets and call a callback function for each.
@@ -136 +136 @@
     """
 
+    if callback_args is None:
+        callback_args = []
+
     session = lpcap.open_pcap(device, snaplen, promisc, timeout)
     if filter:
@@ -148 +151 @@
     return sniff(1, device='any')
 
-def from_file(filename, count=0, filter=None, callback=None,
-                                                        callback_args=None):
+def from_file(filename, count=0, filter=None):
     """
     Load data from pcap file instead of sniffing online.
@@ -163 +165 @@
     @type filter: C{str}
     @param filter: BPF filter
-
-    @type callback: C{func}
-    @param callback: function with (timestamp, pkt, *callback_args) prototype
-
-    @type callback_args: C{list}
-    @param callback_args: additional arguments for callback function
     """
 
@@ -178 +174 @@
     if filter:
         f.setfilter(filter)
-    if callback is not None:
-        f.loop(count, callback, *callback_args)
-    elif count > 0:
+    if count > 0:
         packets = []
         for i in xrange(count):
@@ -189 +183 @@
     return packets
 
+def from_file_loop(filename, count=0, filter=None, callback=None,
+                                                callback_args=None):
+    """
+    Load data from pcap file instead of sniffing online.
+
+    Call callback for each or return list of packets.
+
+    @type filename: C{str}
+    @param filename: path to a file in pcap format
+
+    @type count: C{int}
+    @param count: number of sniffing packets; 0 means infinity (default: I{0})
+
+    @type filter: C{str}
+    @param filter: BPF filter
+
+    @type callback: C{func}
+    @param callback: function with (timestamp, pkt, *callback_args) prototype
+
+    @type callback_args: C{list}
+    @param callback_args: additional arguments for callback function
+    """
+
+    if callback_args is None:
+        callback_args = []
+
+    if os.path.isfile(filename):
+        f = lpcap.open_pcap(filename)
+    else:
+        raise UMPASniffingException("can't open file: %s" % filename)
+
+    if filter:
+        f.setfilter(filter)
+    f.loop(count, callback, *callback_args)
+
 def to_file():
     pass