Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions pcapkit/corekit/protochain.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@

"""
import collections.abc
import copy
from typing import TYPE_CHECKING, overload

from pcapkit.utilities.compat import cached_property
Expand Down Expand Up @@ -260,6 +259,8 @@ def __add__(self, other: 'ProtoChain') -> 'ProtoChain':
Merged protocol chain.

"""
new = copy.copy(self)
new.__data__ += other.__data__
# build a fresh instance, so the cached ``protocols`` and
# ``aliases`` of ``self`` are not carried over
new = self.__class__.__new__(self.__class__)
new.__data__ = self.__data__ + other.__data__
return new
61 changes: 61 additions & 0 deletions tests/corekit/test_protochain_add_stale_cache_1099_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
# -*- coding: utf-8 -*-
"""GitHub issue #1099: :meth:`ProtoChain.__add__ <pcapkit.corekit.protochain.ProtoChain.__add__>`
must not carry the left operand's cached ``protocols``/``aliases`` into the result.

Both are :func:`~functools.cached_property` values stored on the instance, so a
result built by shallow-copying the left operand kept them from before the merge.
"""
from __future__ import annotations

import unittest

from pcapkit.corekit.protochain import ProtoChain
from pcapkit.protocols.internet.ipv4 import IPv4
from pcapkit.protocols.link.ethernet import Ethernet
from pcapkit.protocols.transport.tcp import TCP


class ProtoChainAddCacheTests(unittest.TestCase):
def _check(self, *, warm: bool) -> None:
left = ProtoChain(Ethernet)
right = ProtoChain(TCP, basis=ProtoChain(IPv4)) # TCP:IPv4
if warm:
self.assertEqual(left.protocols, (Ethernet,))
self.assertEqual(left.aliases, ('Ethernet',))
self.assertEqual(right.protocols, (TCP, IPv4))

merged = left + right

self.assertEqual(merged.protocols, (Ethernet, TCP, IPv4))
self.assertEqual(merged.aliases, ('Ethernet', 'TCP', 'IPv4'))
self.assertEqual(merged.chain, 'Ethernet:TCP:IPv4')
self.assertEqual(len(merged), 3)

# neither operand is mutated, whether or not its cache was warm
self.assertEqual(left.protocols, (Ethernet,))
self.assertEqual(left.aliases, ('Ethernet',))
self.assertEqual(left.chain, 'Ethernet')
self.assertEqual(right.chain, 'TCP:IPv4')

def test_add_after_reading_cached_properties(self) -> None:
self._check(warm=True)

def test_add_without_prior_read(self) -> None:
self._check(warm=False)

def test_inplace_add_rebinds_without_mutating_the_original(self) -> None:
original = ProtoChain(Ethernet)
self.assertEqual(original.protocols, (Ethernet,))

chain = original
chain += ProtoChain(IPv4)

self.assertIsNot(chain, original)
self.assertEqual(chain.protocols, (Ethernet, IPv4))
self.assertEqual(chain.aliases, ('Ethernet', 'IPv4'))
self.assertEqual(original.protocols, (Ethernet,))
self.assertEqual(original.chain, 'Ethernet')


if __name__ == '__main__':
unittest.main()
Loading