pyshark/tests/test_cap_operations.py at ad4602305b349d429cfd326317b7bd970fdc0977 · KimiNewt/pyshark · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import time
from asyncio import TimeoutError
from multiprocessing import Process
from multiprocessing import Queue
from multiprocessing.queues import Empty
from unittest import mock

import pytest

import pyshark
from pyshark.packet.packet_summary import PacketSummary
from tests.conftest import example_pcap_path


def test_packet_callback_called_for_each_packet(lazy_simple_capture):
    # Test cap has 24 packets
    mock_callback = mock.Mock()
    lazy_simple_capture.apply_on_packets(mock_callback)
    assert mock_callback.call_count == 24


def test_async_packet_callback_called_for_each_packet(lazy_simple_capture):
    # Test cap has 24 packets
    mock_callback = mock.AsyncMock()
    lazy_simple_capture.apply_on_packets(mock_callback)
    assert mock_callback.call_count == 24
    mock_callback.assert_awaited()


def test_apply_on_packet_stops_on_timeout(lazy_simple_capture):
    def wait(pkt):
        time.sleep(5)
    with pytest.raises(TimeoutError):
        lazy_simple_capture.apply_on_packets(wait, timeout=1)


def test_lazy_loading_of_packets_on_getitem(lazy_simple_capture):
    # Seventh packet is ICMP
    assert 'ICMP' in lazy_simple_capture[6]


def test_lazy_loading_of_packet_does_not_recreate_packets(lazy_simple_capture):
    # Seventh packet is ICMP
    icmp_packet_id = id(lazy_simple_capture[6])
    # load some more
    lazy_simple_capture[8]
    assert icmp_packet_id == id(lazy_simple_capture[6])


def test_filling_cap_in_increments(lazy_simple_capture):
    lazy_simple_capture.load_packets(1)
    assert len(lazy_simple_capture) == 1
    lazy_simple_capture.load_packets(2)
    assert len(lazy_simple_capture) == 3


def test_getting_packet_summary(simple_summary_capture):
    assert isinstance(simple_summary_capture[0], PacketSummary)

    # Since we cannot check the exact fields since they're dependent on wireshark configuration,
    # we'll at least make sure some data is in.
    assert simple_summary_capture[0]._fields


def _iterate_capture_object(example_pcap_path, q):
    cap_obj = pyshark.FileCapture(example_pcap_path, debug=True, only_summaries=True)
    cap_obj.display_filter = "frame.len == 1"
    for _ in cap_obj:
        pass
    q.put(True)


def test_iterate_empty_psml_capture(example_pcap_path):
    q = Queue()
    p = Process(target=_iterate_capture_object,
                args=(example_pcap_path, q))
    p.start()
    p.join(2)
    try:
        no_hang = q.get_nowait()
    except Empty:
        no_hang = False
    if p.is_alive():
        p.terminate()
    assert no_hang  # False here