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
89 changes: 60 additions & 29 deletions tests/protocols/misc/test_pcapng_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import struct
import subprocess # nosec: B404
import sys
import sysconfig
import tempfile
import textwrap
import time
Expand Down Expand Up @@ -4744,14 +4745,30 @@ def test_the_captured_len_vector_parses_under_a_memory_cap(self) -> None:
failing, and a cap that the parent cannot lift is the only way to tell
"parsed" from "did not run out of memory yet".

The cap is 512 MiB over the address space the subprocess has already
mapped, not 512 MiB in all: on a 64-bit build mimalloc reserves address
space 1 GiB at a time, the first as the interpreter starts, so under it
an absolute cap is spent before the parse begins (#1731). A
free-threaded build always uses mimalloc; any other build that has it
runs the subprocess a second time under ``PYTHONMALLOC=mimalloc``, so a
leg with the GIL measures that too.

"""
if importlib.util.find_spec('resource') is None: # pragma: no cover
self.skipTest('the resource module is unavailable on this platform')

source = textwrap.dedent('''
import io, json, resource, struct, sys, warnings
sys.path.insert(0, sys.argv[1])
resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 ** 2,) * 2)

# /proc/self/statm opens with the address space mapped, in pages;
# without procfs the cap is absolute
try:
with open('/proc/self/statm', encoding='ascii') as statm:
mapped = int(statm.read().split()[0]) * resource.getpagesize()
except OSError:
mapped = 0
resource.setrlimit(resource.RLIMIT_AS, (mapped + 512 * 1024 ** 2,) * 2)

import pcapkit
from pcapkit.foundation.extraction import Extractor
Expand All @@ -4778,34 +4795,48 @@ class NamedBuffer(io.BufferedReader):
'frames': len(extractor.frame)}))
''')

with tempfile.TemporaryDirectory() as tempdir:
script = os.path.join(tempdir, 'capped.py')
with open(script, 'w', encoding='utf-8') as handle:
handle.write(source)
status = os.path.join(tempdir, 'status')
output = os.path.join(tempdir, 'stdout')

with open(output, 'w', encoding='utf-8') as stdout:
completed = subprocess.run( # nosec: B603
(sys.executable, script, str(ROOT)),
stdout=stdout, stderr=subprocess.DEVNULL, check=False, timeout=300,
)
# the exit code read back from a file rather than taken from a
# pipeline, where a wrapper's own status can be reported instead
with open(status, 'w', encoding='utf-8') as handle:
handle.write(str(completed.returncode))
with open(status, encoding='utf-8') as handle:
returncode = int(handle.read())
with open(output, encoding='utf-8') as handle:
payload = handle.read()

self.assertEqual(returncode, 0, payload)
result = json.loads(payload)
# the subprocess measured a tree under this repository, not an installed
# copy of pcapkit somewhere else
self.assertTrue(result['file'].startswith(str(ROOT)), result['file'])
self.assertEqual(result['input'], 8048)
self.assertEqual(result['frames'], 200)
# a build option rather than a version: 3.13 and later can be built
# --without-mimalloc, and then reject PYTHONMALLOC=mimalloc at start-up
allocators = [None]
if sysconfig.get_config_var('WITH_MIMALLOC'):
allocators.append('mimalloc')

for allocator in allocators:
env = dict(os.environ)
if allocator is not None:
env['PYTHONMALLOC'] = allocator
with self.subTest(allocator=allocator or 'default'), tempfile.TemporaryDirectory() as tempdir:
script = os.path.join(tempdir, 'capped.py')
with open(script, 'w', encoding='utf-8') as handle:
handle.write(source)
status = os.path.join(tempdir, 'status')
output = os.path.join(tempdir, 'stdout')
errors = os.path.join(tempdir, 'stderr')

with open(output, 'w', encoding='utf-8') as stdout, \
open(errors, 'w', encoding='utf-8') as stderr:
completed = subprocess.run( # nosec: B603
(sys.executable, script, str(ROOT)),
stdout=stdout, stderr=stderr, env=env, check=False, timeout=300,
)
# the exit code read back from a file rather than taken from a
# pipeline, where a wrapper's own status can be reported instead
with open(status, 'w', encoding='utf-8') as handle:
handle.write(str(completed.returncode))
with open(status, encoding='utf-8') as handle:
returncode = int(handle.read())
with open(output, encoding='utf-8') as handle:
payload = handle.read()
with open(errors, encoding='utf-8', errors='replace') as handle:
diagnostic = handle.read()[-2000:]

self.assertEqual(returncode, 0, payload + diagnostic)
result = json.loads(payload)
# the subprocess measured a tree under this repository, not an
# installed copy of pcapkit somewhere else
self.assertTrue(result['file'].startswith(str(ROOT)), result['file'])
self.assertEqual(result['input'], 8048)
self.assertEqual(result['frames'], 200)


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
Expand Down
25 changes: 21 additions & 4 deletions tests/protocols/test_trial_parse_warnings_1580_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -382,16 +382,33 @@ def test_held_frame_gone_is_reported_outside_pcapkit(self) -> None:
[('past the outermost frame', __file__)])

def test_hold_is_local_to_its_context(self) -> None:
"""A warning reported in another context is not held by this one's block."""
"""A warning reported in another context is not held by this one's block.

Each other context records for itself: under
:data:`sys.flags.context_aware_warnings`, on by default in a
free-threaded build, :func:`warnings.catch_warnings` records in a context
variable, which an empty context does not have (#1731).

"""
from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn

def record_in_an_empty_context() -> 'list[str]':
with warnings.catch_warnings(record=True) as inner:
warnings.simplefilter('always')
warn('an empty context', SchemaWarning)
return self._schema(inner)

with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
# copied before the block is entered: the recording, not the block
copied = contextvars.copy_context()
with hold_warnings() as held:
contextvars.Context().run(warn, 'another context', SchemaWarning)
self.assertEqual(self._schema(caught), ['another context'])
copied.run(warn, 'a copied context', SchemaWarning)
self.assertEqual(self._schema(caught), ['a copied context'])
self.assertEqual(contextvars.Context().run(record_in_an_empty_context),
['an empty context'])
held.discard = True
self.assertEqual(self._schema(caught), ['another context'])
self.assertEqual(self._schema(caught), ['a copied context'])

def test_hold_left_out_of_order(self) -> None:
"""A generator's block left out of order neither ends another block nor outlives itself."""
Expand Down
Loading