|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""Calling ``pack()`` leaves the instance it is called on as it was. C.f. #1769. |
| 3 | +
|
| 4 | +:meth:`ProtocolBase.pack <pcapkit.protocols.protocol.ProtocolBase.pack>` builds |
| 5 | +a fresh header from its keywords alone. It used to assign that header to the |
| 6 | +instance as well, so a bare ``UDP(srcport=1234, dstport=53).pack()`` left |
| 7 | +``schema.srcport`` at ``0`` while ``info.srcport`` was still ``1234``, and |
| 8 | +:meth:`TCP.make <pcapkit.protocols.transport.tcp.TCP.make>` overwrote |
| 9 | +:attr:`~pcapkit.protocols.transport.tcp.TCP.connection` on the way. The |
| 10 | +``pack`` overrides of ``Frame`` and ``PCAPNG`` did the same, and HTTP/2's |
| 11 | +dropped the header altogether. |
| 12 | +
|
| 13 | +Every value is built here, so the module is in the unit tier. |
| 14 | +
|
| 15 | +""" |
| 16 | +from __future__ import annotations |
| 17 | + |
| 18 | +import importlib |
| 19 | +import importlib.util |
| 20 | +import unittest |
| 21 | +import warnings |
| 22 | +from typing import TYPE_CHECKING |
| 23 | + |
| 24 | +from tests._support import reimport_once_per_class |
| 25 | + |
| 26 | +if TYPE_CHECKING: |
| 27 | + from typing import Any |
| 28 | + |
| 29 | + from pcapkit.protocols.protocol import ProtocolBase |
| 30 | + |
| 31 | +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') |
| 32 | +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) |
| 33 | + |
| 34 | + |
| 35 | +class _KeepsState(unittest.TestCase): |
| 36 | + """Shared assertions.""" |
| 37 | + |
| 38 | + def setUp(self) -> 'None': |
| 39 | + reimport_once_per_class(self) |
| 40 | + |
| 41 | + def assertPackKeepsState(self, proto: 'ProtocolBase', **kwargs: 'Any') -> 'bytes': # noqa: N802 |
| 42 | + """Call ``proto.pack(**kwargs)`` and check nothing on ``proto`` moved. |
| 43 | +
|
| 44 | + ``schema`` is a cached property, so it is read only *after* the call: |
| 45 | + read before, it would hide a replaced header behind its cache. |
| 46 | +
|
| 47 | + Returns: |
| 48 | + What ``pack`` returned. |
| 49 | +
|
| 50 | + """ |
| 51 | + header, info, data = proto.__header__, proto.info, proto.data |
| 52 | + with warnings.catch_warnings(): |
| 53 | + warnings.simplefilter('ignore') |
| 54 | + packed = proto.pack(**kwargs) |
| 55 | + self.assertIs(proto.__header__, header) |
| 56 | + self.assertIs(proto.schema, header) |
| 57 | + self.assertIs(proto.info, info) |
| 58 | + self.assertEqual(proto.data, data) |
| 59 | + return packed |
| 60 | + |
| 61 | + |
| 62 | +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') |
| 63 | +class TestLayers(_KeepsState): |
| 64 | + """A link, an internet and two transport protocols, constructed and parsed.""" |
| 65 | + |
| 66 | + CASES = ( |
| 67 | + ('pcapkit.protocols.link.ethernet', 'Ethernet', |
| 68 | + {'src': 'aa:bb:cc:dd:ee:ff', 'dst': '11:22:33:44:55:66'}), |
| 69 | + ('pcapkit.protocols.internet.ipv4', 'IPv4', {'src': '1.2.3.4', 'dst': '5.6.7.8', 'ttl': 7}), |
| 70 | + ('pcapkit.protocols.transport.udp', 'UDP', {'srcport': 1234, 'dstport': 53}), |
| 71 | + ('pcapkit.protocols.transport.tcp', 'TCP', {'srcport': 80, 'dstport': 443, 'syn': True, 'ack': True}), |
| 72 | + ) |
| 73 | + |
| 74 | + def test_bare_pack_returns_a_default_header(self) -> 'None': |
| 75 | + for module, name, kwargs in self.CASES: |
| 76 | + with self.subTest(protocol=name): |
| 77 | + cls = getattr(importlib.import_module(module), name) |
| 78 | + proto = cls(**kwargs) |
| 79 | + self.assertEqual(self.assertPackKeepsState(proto), cls().data) |
| 80 | + # NOTE: the schema still describes the octets the instance holds. |
| 81 | + self.assertEqual(bytes(proto.schema), proto.data) |
| 82 | + |
| 83 | + def test_pack_with_keywords_keeps_state(self) -> 'None': |
| 84 | + for module, name, kwargs in self.CASES: |
| 85 | + with self.subTest(protocol=name): |
| 86 | + cls = getattr(importlib.import_module(module), name) |
| 87 | + proto = cls() |
| 88 | + self.assertEqual(self.assertPackKeepsState(proto, **kwargs), cls(**kwargs).data) |
| 89 | + self.assertEqual(bytes(proto.schema), cls().data) |
| 90 | + |
| 91 | + def test_parsed_instance_keeps_state(self) -> 'None': |
| 92 | + for module, name, kwargs in self.CASES: |
| 93 | + with self.subTest(protocol=name): |
| 94 | + cls = getattr(importlib.import_module(module), name) |
| 95 | + octets = cls(**kwargs).data |
| 96 | + proto = cls(octets, len(octets)) |
| 97 | + self.assertEqual(self.assertPackKeepsState(proto), cls().data) |
| 98 | + self.assertEqual(bytes(proto.schema), octets) |
| 99 | + |
| 100 | + def test_issue_example(self) -> 'None': |
| 101 | + from pcapkit.protocols.transport.udp import UDP |
| 102 | + |
| 103 | + udp = UDP(srcport=1234, dstport=53) |
| 104 | + self.assertEqual(udp.data, bytes.fromhex('04d2003500080000')) |
| 105 | + self.assertEqual(self.assertPackKeepsState(udp), bytes.fromhex('0000000000080000')) |
| 106 | + self.assertEqual(udp.schema.srcport, 1234) |
| 107 | + self.assertEqual(udp.info.srcport, 1234) |
| 108 | + |
| 109 | + def test_tcp_connection_flags(self) -> 'None': |
| 110 | + from pcapkit.const.tcp.flags import Flags |
| 111 | + from pcapkit.protocols.transport.tcp import TCP |
| 112 | + |
| 113 | + tcp = TCP(srcport=80, dstport=443, syn=True, ack=True) |
| 114 | + self.assertEqual(tcp.connection, Flags.SYN | Flags.ACK) |
| 115 | + self.assertPackKeepsState(tcp) |
| 116 | + self.assertEqual(tcp.connection, Flags.SYN | Flags.ACK) |
| 117 | + |
| 118 | + |
| 119 | +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') |
| 120 | +class TestOverrides(_KeepsState): |
| 121 | + """The ``pack`` overrides that built their own header.""" |
| 122 | + |
| 123 | + def test_frame(self) -> 'None': |
| 124 | + from pcapkit.protocols.misc.pcap.frame import Frame |
| 125 | + from pcapkit.protocols.misc.pcap.header import Header |
| 126 | + |
| 127 | + header = Header().info |
| 128 | + frame = Frame(num=1, header=header, ts_sec=1, ts_usec=2, packet=b'abcd') |
| 129 | + packed = self.assertPackKeepsState(frame, num=1, header=header, ts_sec=3, ts_usec=4, packet=b'xy') |
| 130 | + self.assertEqual(packed, Frame(num=1, header=header, ts_sec=3, ts_usec=4, packet=b'xy').data) |
| 131 | + self.assertEqual(frame.schema.ts_sec, 1) |
| 132 | + |
| 133 | + def test_pcapng(self) -> 'None': |
| 134 | + from pcapkit.const.pcapng.block_type import BlockType |
| 135 | + from pcapkit.const.pcapng.option_type import OptionType |
| 136 | + from pcapkit.const.reg.linktype import LinkType |
| 137 | + from pcapkit.foundation.engines.pcapng import Context |
| 138 | + from pcapkit.protocols.misc.pcapng import PCAPNG |
| 139 | + |
| 140 | + section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, block={'byteorder': 'big'}) |
| 141 | + self.assertEqual(section.byteorder, 'big') |
| 142 | + self.assertPackKeepsState(section, num=0, sct=1, ctx=None, |
| 143 | + type=BlockType.Section_Header_Block, block={'byteorder': 'little'}) |
| 144 | + self.assertEqual(section.byteorder, 'big') |
| 145 | + |
| 146 | + context = Context(section.info) |
| 147 | + kwargs = {'num': 1, 'sct': 1, 'ctx': context, 'type': BlockType.Interface_Description_Block, |
| 148 | + 'block': {'linktype': LinkType.ETHERNET, 'snaplen': 0x40000, |
| 149 | + 'options': [(OptionType.if_name, {'name': 'eth0'})]}} |
| 150 | + interface = PCAPNG(**kwargs) |
| 151 | + counts = dict(interface._opt) # pylint: disable=protected-access |
| 152 | + self.assertEqual(self.assertPackKeepsState(interface, **kwargs), interface.data) |
| 153 | + self.assertEqual(dict(interface._opt), counts) # pylint: disable=protected-access |
| 154 | + # NOTE: the bare make defaults to another block type. |
| 155 | + packed = self.assertPackKeepsState(interface, num=1, sct=1, ctx=context) |
| 156 | + self.assertNotEqual(packed, interface.data) |
| 157 | + self.assertEqual(interface.block, BlockType.Interface_Description_Block) |
| 158 | + |
| 159 | + def test_httpv2(self) -> 'None': |
| 160 | + from pcapkit.protocols.application.httpv2 import HTTP as HTTPv2 |
| 161 | + |
| 162 | + kwargs = {'type': 0x00, 'sid': 1, 'frame': {'data': b'abc'}} |
| 163 | + frame = HTTPv2(**kwargs) |
| 164 | + self.assertIsNotNone(frame.__header__) |
| 165 | + self.assertEqual(self.assertPackKeepsState(frame, **kwargs), frame.data) |
| 166 | + |
| 167 | + |
| 168 | +if __name__ == '__main__': |
| 169 | + unittest.main() |
0 commit comments