Remove 'mock' dependency (#4480)

2024-07-28 20:01:39 +02:00 · 2024-07-28 20:01:39 +02:00 · 8ac13e7c15
parent 26303ffd4e
commit 8ac13e7c15
18 changed files with 72 additions and 77 deletions
--- a/scapy/utils.py
+++ b/scapy/utils.py
@ -981,14 +981,10 @@ class ContextManagerCaptureOutput(object):
    def __init__(self):
        # type: () -> None
        self.result_export_object = ""
-        try:
-            import mock  # noqa: F401
-        except Exception:
-            raise ImportError("The mock module needs to be installed !")

    def __enter__(self):
        # type: () -> ContextManagerCaptureOutput
-        import mock
+        from unittest import mock

        def write(s, decorator=self):
            # type: (str, ContextManagerCaptureOutput) -> None
--- a/test/answering_machines.uts
+++ b/test/answering_machines.uts
@ -8,7 +8,7 @@
 + Answering Machines

 = Generic answering machine mocker
-import mock
+from unittest import mock
@mock.patch("scapy.ansmachine.sniff")
 def test_am(cls_name, packet_query, check_reply, mock_sniff, **kargs):
    packet_query = packet_query.__class__(bytes(packet_query))
@ -236,7 +236,7 @@ assert res[DHCP6_Solicit]
 a.print_reply(req, res)

 = WiFi_am
-import mock
+from unittest import mock
@mock.patch("scapy.layers.dot11.sniff")
 def test_WiFi_am(packet_query, check_reply, mock_sniff, **kargs):
    def sniff(*args,**kargs):
--- a/test/bpf.uts
+++ b/test/bpf.uts
@ -114,7 +114,7 @@ s.close()
 = L2bpfListenSocket - read failure
 ~ needs_root

-import mock
+from unittest import mock

@mock.patch("scapy.arch.bpf.supersocket.os.read")
 def _test_osread(osread):
--- a/test/contrib/dtp.uts
+++ b/test/contrib/dtp.uts
@ -16,7 +16,7 @@ assert pkt[DTP].tlvlist[3].status == b'\x03'

 = Test negotiate_trunk

-import mock
+from unittest import mock

 def test_pkt(pkt):
    pkt = Ether(raw(pkt))
--- a/test/linux.uts
+++ b/test/linux.uts
@ -66,7 +66,7 @@ else:
 = catch loopback device missing
 ~ linux needs_root

-from mock import patch
+from unittest.mock import patch

 # can't remove the lo device (or its address without causing trouble) - use some pseudo dummy instead

@ -77,7 +77,7 @@ with patch('scapy.arch.linux.conf.loopback_name', 'scapy_lo_x'):
 ~ linux needs_root

 import os, socket
-from mock import patch
+from unittest.mock import patch

 try:
    exit_status = os.system("ip link add name scapy_lo type dummy")
@ -116,7 +116,7 @@ finally:

 conf.ifaces._add_fake_iface("scapy0", 'e2:39:91:79:19:10')

-from mock import patch
+from unittest.mock import patch
 conf.route6.routes =  [('fe80::', 64, '::', 'scapy0', ['fe80::e039:91ff:fe79:1910'], 256)]
 conf.route6.ipv6_ifaces = set(['scapy0'])
 bck_conf_iface = conf.iface
@ -274,7 +274,7 @@ except Exception:
 = Routing table, interface with no names
 ~ linux

-from mock import patch
+from unittest.mock import patch

@patch("scapy.arch.linux.ioctl")
 def test_read_routes(mock_ioctl):
@ -293,7 +293,8 @@ test_read_routes()

 from scapy.arch.linux import L3PacketSocket

-import mock, socket
+import socket
+from unittest import mock

@mock.patch("scapy.arch.linux.socket.socket.sendto")
 def test_L3PacketSocket_sendto_python3(mock_sendto):
--- a/test/pipetool.uts
+++ b/test/pipetool.uts
@ -228,7 +228,7 @@ p.wait_and_stop()

 = Test SniffSource

-import mock
+from unittest import mock
 fd = ObjectPipe("sniffsource")
 fd.write("test")

@ -295,7 +295,7 @@ else:

 = Test exhausted AutoSource and SniffSource

-import mock
+from unittest import mock
 from scapy.error import Scapy_Exception

 def _fail():
@ -323,7 +323,7 @@ except:
 q = ObjectPipe("wiresharksink")
 pkt = Ether(dst="aa:aa:aa:aa:aa:aa", src="bb:bb:bb:bb:bb:bb")/IP(dst="127.0.0.1", src="127.0.0.1")/ICMP()

-import mock
+from unittest import mock
 with mock.patch("scapy.scapypipes.subprocess.Popen", return_value=Bunch(stdin=q)) as popen:
    sink = WiresharkSink()
    sink.start()
@ -345,7 +345,7 @@ linktype = scapy.data.DLT_EN3MB
 q = ObjectPipe("wiresharksink_linktype")
 pkt = Ether(dst="aa:aa:aa:aa:aa:aa", src="bb:bb:bb:bb:bb:bb")/IP(dst="127.0.0.1", src="127.0.0.1")/ICMP()

-import mock
+from unittest import mock
 with mock.patch("scapy.scapypipes.subprocess.Popen", return_value=Bunch(stdin=q)) as popen:
    sink = WiresharkSink(linktype=linktype)
    sink.start()
@ -363,7 +363,7 @@ linktype = scapy.data.DLT_EN3MB
 q = ObjectPipe("wiresharksink_args")
 pkt = Ether(dst="aa:aa:aa:aa:aa:aa", src="bb:bb:bb:bb:bb:bb")/IP(dst="127.0.0.1", src="127.0.0.1")/ICMP()

-import mock
+from unittest import mock
 with mock.patch("scapy.scapypipes.subprocess.Popen", return_value=Bunch(stdin=q)) as popen:
    sink = WiresharkSink(args=['-c', '1'])
    sink.start()
@ -404,7 +404,7 @@ os.unlink(os.path.join(dname, "t2.pcap.gz"))
 = Test InjectSink and Inject3Sink
 ~ needs_root

-import mock
+from unittest import mock

 a = IP(dst="192.168.0.1")/ICMP()
 msgs = []
--- a/test/regression.uts
+++ b/test/regression.uts
@ -46,7 +46,7 @@ assert _version_checker(FakeModule3, (2, 4, 2))

 = Check Scapy version

-import mock
+from unittest import mock

 import scapy
 from scapy import _parse_tag, _version_from_git_describe
@ -280,7 +280,7 @@ assert p == "127.0.0.1"

 = Interface related functions

-import mock
+from unittest import mock

 conf.iface

@ -321,7 +321,7 @@ assert conf.iface == old
 assert isinstance(conf.iface, NetworkInterface)
 assert conf.iface.is_valid()

-import mock
+from unittest import mock
@mock.patch("scapy.interfaces.conf.route.routes", [])
@mock.patch("scapy.interfaces.conf.ifaces.values")
 def _test_get_working_if(rou):
@ -670,7 +670,8 @@ assert len(conf.temp_files) == 0
 = Emulate interact()
 ~ interact

-import mock, sys
+import sys
+from unittest import mock
 from scapy.main import interact

 from scapy.main import DEFAULT_PRESTART_FILE, DEFAULT_PRESTART, _read_config_file
@ -712,7 +713,7 @@ interact_emulator(extra_args=["-d"])  # Extended
 ~ interact

 import sys
-import mock
+from unittest import mock

 from scapy.main import DEFAULT_PRESTART_FILE, DEFAULT_PRESTART, _read_config_file
 _read_config_file(DEFAULT_PRESTART_FILE, _locals=globals(), default=DEFAULT_PRESTART)
@ -756,7 +757,7 @@ assert called
 = Test explore() with GUI mode
 ~ command

-import mock
+from unittest import mock

 def test_explore_gui(is_layer, layer):
    prompt_toolkit_mocked_module = Bunch(
@ -847,7 +848,7 @@ assert lhex((28,7)) == "(0x1c, 0x7)"
 assert lhex([28,7]) == "[0x1c, 0x7]"

 = Test restart function
-import mock
+from unittest import mock
 conf.interactive = True

 try:
@ -969,7 +970,7 @@ zerofree_randstring(4) in [b"\xd2\x12\xe4\x5b", b'\xd3\x8b\x13\x12']
 assert strand(b"AC", b"BC") == b'@C'

 = Test export_object and import_object functions
-import mock
+from unittest import mock
 def test_export_import_object():
    with ContextManagerCaptureOutput() as cmco:
        export_object(2807)
@ -1157,7 +1158,7 @@ conf.netcache

 = Test pyx detection functions

-from mock import patch
+from unittest.mock import patch

 def _r(*args, **kwargs):
    raise OSError
@ -1168,7 +1169,7 @@ with patch("scapy.libs.test_pyx.subprocess.check_call", _r):

 = Test matplotlib detection functions

-from mock import MagicMock, patch
+from unittest.mock import MagicMock, patch

 bck_scapy_libs_matplot = sys.modules.get("scapy.libs.matplot", None)
 if bck_scapy_libs_matplot:
@ -2049,7 +2050,7 @@ send(fuzz(ARP()))
 = Test SuperSocket.select
 ~ select

-import mock
+from unittest import mock

@mock.patch("scapy.supersocket.select")
 def _test_select(select):
@ -2444,7 +2445,7 @@ assert len(sniff(offline=[IP()/UDP(), IP()/TCP()], lfilter=lambda x: TCP in x))

 = Check offline sniff() without a tcpdump binary
 ~ tcpdump
-import mock
+from unittest import mock

 conf_prog_tcpdump = conf.prog.tcpdump
 conf.prog.tcpdump = "tcpdump_fake"
@ -2628,7 +2629,7 @@ assert b'127.0.0.1 > 127.0.0.1:' in data[2]

 * Non existing tcpdump binary

-import mock
+from unittest import mock

 conf_prog_tcpdump = conf.prog.tcpdump
 conf.prog.tcpdump = "tcpdump_fake"
@ -2836,7 +2837,7 @@ os.remove(filename)

 = Check wrpcap() with different packets types

-import mock
+from unittest import mock
 import os
 import tempfile

@ -2910,7 +2911,7 @@ assert pkterf[1][Ether].src == "00:0f:53:3f:ca:c0"
 = Truncated netstat -rn output on OS X
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@ -2968,7 +2969,7 @@ test_osx_netstat_truncated()
 = macOS 10.13
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@ -3027,7 +3028,7 @@ test_osx_10_13_ipv4()
 = macOS 10.15
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@ -3082,7 +3083,7 @@ test_osx_10_15_ipv4()
 = OpenBSD 6.3
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@ -3152,7 +3153,7 @@ assert locked[3] == "bge0"
 = Solaris 11.1
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

 # Mocked Solaris 11.1 parsing behavior
@ -3284,7 +3285,7 @@ assert results_dict == expected
 = Preliminary definitions
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

 def valid_output_read_routes6(routes):
@ -3314,7 +3315,7 @@ def check_mandatory_ipv6_routes(routes6):
 = Mac OS X 10.9.5
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@ -3362,7 +3363,7 @@ test_osx_10_9_5()
 = Mac OS X 10.9.5 with global IPv6 connectivity
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@ -3415,7 +3416,7 @@ test_osx_10_9_5_global()
 = Mac OS X 10.10.4
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@ -3458,7 +3459,7 @@ test_osx_10_10_4()
 = FreeBSD 10.2
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@ -3504,7 +3505,7 @@ test_freebsd_10_2()
 = FreeBSD 13.0
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO
 
@mock.patch("scapy.arch.get_if_addr")
@ -3550,7 +3551,7 @@ test_freebsd_13()
 = OpenBSD 5.5
 ~ mock_read_routes_bsd

-import mock
+from unittest import mock
 from io import StringIO

@mock.patch("scapy.arch.unix.OPENBSD")
@ -4415,7 +4416,7 @@ assert sum(1 for oid in conf.mib) > 100
 = MIB - graph
 ~ mib

-import mock
+from unittest import mock

@mock.patch("scapy.asn1.mib.do_graph")
 def get_mib_graph(do_graph):
@ -4890,7 +4891,7 @@ assert all(bytes(a[0]) == bytes(b[0]) for a, b in zip(unp, srl))

 = plot()

-import mock
+from unittest import mock
 import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
@ -4906,7 +4907,7 @@ test_plot()

 = diffplot()

-import mock
+from unittest import mock
 import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
@ -4922,7 +4923,7 @@ test_diffplot()

 = multiplot()

-import mock
+from unittest import mock
 import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
@ -5031,7 +5032,7 @@ test_nzpadding()

 = conversations()

-import mock
+from unittest import mock
@mock.patch("scapy.plist.do_graph")
 def test_conversations(mock_do_graph):
    def fake_do_graph(graph, **kwargs):
@ -5059,7 +5060,7 @@ assert len(pl.sessions().keys()) == 5

 = afterglow()

-import mock
+from unittest import mock
@mock.patch("scapy.plist.do_graph")
 def test_afterglow(mock_do_graph):
    def fake_do_graph(graph, **kwargs):
@ -5151,7 +5152,7 @@ del os.environ["SCAPY_VERSION"]
 assert scapy._version() == version
 os.unlink(version_filename)

-import mock
+from unittest import mock
 with mock.patch("scapy._version_from_git_archive") as archive:
    archive.return_value = "4.4.4"
    assert scapy._version() == "4.4.4"
@ -5189,7 +5190,7 @@ assert os.path.isdir(dname)
 = test fragleak functions
 ~ netaccess linux fragleak

-import mock
+from unittest import mock

@mock.patch("scapy.layers.inet.conf.L3socket")
@mock.patch("scapy.layers.inet.select.select")
--- a/test/scapy/layers/inet.uts
+++ b/test/scapy/layers/inet.uts
@ -590,7 +590,7 @@ assert isinstance(pkt, DNS) and isinstance(pkt.payload, NoPayload)
 = Layer binding with show()
 * getmacbyip must only be called when building

-import mock
+from unittest import mock

 def _err(*_):
    raise ValueError
@ -763,7 +763,7 @@ def test_summary():

 test_summary()

-import mock
+from unittest import mock
 import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
@ -802,7 +802,7 @@ assert "192.168.0.254" not in [p[IP].src for p in new_pl]
 = IPv4 - reporting
 ~ netaccess

-import mock
+from unittest import mock

@mock.patch("scapy.layers.inet.sr")
 def test_report_ports(mock_sr):
--- a/test/scapy/layers/inet6.uts
+++ b/test/scapy/layers/inet6.uts
@ -1970,7 +1970,7 @@ r6.ifdel("scapy0")

 = IPv6 - utils

-import mock
+from unittest import mock
@mock.patch("scapy.layers.inet6.get_if_hwaddr")
@mock.patch("scapy.layers.inet6.srp1")
 def test_neighsol(mock_srp1, mock_get_if_hwaddr):
@ -2069,7 +2069,7 @@ assert a.answers(q)

 = Define test utilities

-import mock
+from unittest import mock

@mock.patch("scapy.layers.inet6.sniff")
@mock.patch("scapy.layers.inet6.sendp")
--- a/test/scapy/layers/kerberos.uts
+++ b/test/scapy/layers/kerberos.uts
@ -899,7 +899,7 @@ t.open_file(TICKETER_TEMPFILE)
 = Ticketer++ - Get ticket 0, change it, resign it and set it back

 # mock the random to get consistency
-import mock
+from unittest import mock

 def fake_random(x):
    # wow, impressive entropy
@ -1211,7 +1211,7 @@ ts.pausec == 0x9a4db

 = [MS-KILE] RC4 GSS_WrapEx (RFC4757) test vectors (sect 4.5)

-import mock
+from unittest import mock
 from scapy.libs.rfc3961 import Key, EncryptionType

 ssp = KerberosSSP()
@ -1252,7 +1252,7 @@ assert _msgs[0].data.hex() == "112233445566778899aabbccddeeff"
 = Create randomness-mock context manager

 # mock the random to get consistency
-import mock
+from unittest import mock
 from datetime import datetime

 def fake_urandom(x):
@ -1586,7 +1586,7 @@ assert ddata == data

 = GSS_WrapEx/GSS_UnwrapEx: client sends wrapped payload with confidentiality

-import mock
+from unittest import mock
 from scapy.libs.rfc3961 import Key, EncryptionType

 # Data
--- a/test/scapy/layers/l2.uts
+++ b/test/scapy/layers/l2.uts
@ -76,7 +76,7 @@ def sendp_spoof(x, *args, **kwargs):
    assert x[1].psrc == "192.168.0.1"
    assert x[1].pdst == "192.168.0.2"

-import mock
+from unittest import mock
 with mock.patch('scapy.layers.l2.srp', side_effect=srp_spoof), \
     mock.patch('scapy.layers.l2.srploop', side_effect=srploop_spoof), \
     mock.patch('scapy.layers.l2.sendp', side_effect=sendp_spoof):
@ -123,7 +123,7 @@ def srploop_spoof(x, *args, **kwargs):
 def sendp_spoof(x, *args, **kwargs):
    pass

-import mock
+from unittest import mock
 with mock.patch('scapy.layers.l2.srp', side_effect=srp_spoof), \
     mock.patch('scapy.layers.l2.srploop', side_effect=srploop_spoof), \
     mock.patch('scapy.layers.l2.sendp', side_effect=sendp_spoof):
--- a/test/scapy/layers/msnrpc.uts
+++ b/test/scapy/layers/msnrpc.uts
@ -26,7 +26,7 @@ assert ComputeSessionKeyStrongKey(SharedSecret, ClientChallenge, ServerChallenge

 = [MS-NRPC] test vectors - sect 4.3

-import mock
+from unittest import mock
 from scapy.layers.msrpce.msnrpc import NetlogonSSP

 # Input
@ -55,7 +55,7 @@ assert bytes(sig)[:len(FullNetlogonSignatureHeader)] == FullNetlogonSignatureHea

 = [MS-NRPC] test vectors - sect 4.3.1

-import mock
+from unittest import mock
 from scapy.layers.msrpce.msnrpc import NetlogonSSP

 # Input
@ -253,7 +253,7 @@ assert bytes(pkt) == bytes(auth_resp)
 = [NetlogonSSP] - Create randomness-mock context manager

 # mock the random to get consistency
-import mock
+from unittest import mock

 def fake_urandom(x):
    # wow, impressive entropy
--- a/test/scapy/layers/ntlm.uts
+++ b/test/scapy/layers/ntlm.uts
@ -110,7 +110,7 @@ assert bytes(sig) == b'\x01\x00\x00\x00\x7f\xb3\x8e\xc5\xc5]Iv\x00\x00\x00\x00'
 = Create randomness-mock context manager

 # mock the random to get consistency
-import mock
+from unittest import mock

 def fake_urandom(x):
    # wow, impressive entropy
--- a/test/scapy/layers/tls/tls.uts
+++ b/test/scapy/layers/tls/tls.uts
@ -1345,7 +1345,7 @@ assert not TLSHelloRequest().tls_session_update(None)

 = Cryptography module is unavailable
 ~ mock
-import mock
+from unittest import mock

@mock.patch("scapy.layers.tls.crypto.suites.get_algs_from_ciphersuite_name")
 def test_tls_without_cryptography(get_algs_from_ciphersuite_name_mock):
--- a/test/sendsniff.uts
+++ b/test/sendsniff.uts
@ -276,7 +276,7 @@ with VEthPair('a_0', 'a_1') as veth_0:

 = Create a tap interface

-import mock
+from unittest import mock
 import struct
 import subprocess
 from threading import Thread
--- a/test/tools/isotpscanner.uts
+++ b/test/tools/isotpscanner.uts
@ -11,7 +11,7 @@ with open(scapy_path("test/contrib/automotive/interface_mockup.py")) as f:

 ISOTPSocket = ISOTPSoftSocket

-from mock import patch
+from unittest.mock import patch

 + Usage tests

--- a/test/windows.uts
+++ b/test/windows.uts
@ -6,7 +6,7 @@

 = Imports

-import mock
+from unittest import mock

 ############
 ############
--- a/tox.ini
+++ b/tox.ini
@ -28,9 +28,7 @@ passenv =
    OPENSSL_CONF
    # Used by scapy
    SCAPY_USE_LIBPCAP
-deps = mock
-       # cryptography requirements
-       setuptools>=18.5
+deps =
       ipython
       cryptography
       coverage[toml]
@ -105,7 +103,6 @@ description = "Check Scapy compliance against static typing"
 skip_install = true
 deps = mypy==1.7.0
       typing
-       types-mock
 commands = python .config/mypy/mypy_check.py linux
           python .config/mypy/mypy_check.py win32