|
| 1 | +"""Fresh-process integration coverage for the native allocator entrypoint default.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import json |
| 6 | +import os |
| 7 | +import subprocess |
| 8 | +import sys |
| 9 | + |
| 10 | +import pytest |
| 11 | + |
| 12 | +from tests.integration.src.sqlbuild.cli.commands.main._test_types import ( |
| 13 | + NativeAllocatorEntryTestCase, |
| 14 | +) |
| 15 | + |
| 16 | +_PURGE_DELAY_VARIABLE: str = "MIMALLOC_PURGE_DELAY" |
| 17 | +_SCRIPT: str = """ |
| 18 | +import json |
| 19 | +import os |
| 20 | +import sys |
| 21 | +from sqlbuild.cli.entry.main.entry import main |
| 22 | +
|
| 23 | +loaded_at_import = "sqlbuild._native" in sys.modules |
| 24 | +exit_code = main(["--version"]) |
| 25 | +print(json.dumps({ |
| 26 | + "loaded_at_import": loaded_at_import, |
| 27 | + "loaded_after_main": "sqlbuild._native" in sys.modules, |
| 28 | + "purge_delay": os.environ.get("MIMALLOC_PURGE_DELAY"), |
| 29 | + "exit_code": exit_code, |
| 30 | +})) |
| 31 | +""" |
| 32 | + |
| 33 | + |
| 34 | +@pytest.mark.parametrize( |
| 35 | + "test_case", |
| 36 | + [ |
| 37 | + NativeAllocatorEntryTestCase( |
| 38 | + description="unset purge delay defaults to ten milliseconds", |
| 39 | + inherited_environment=(), |
| 40 | + expected_purge_delay="10", |
| 41 | + ), |
| 42 | + NativeAllocatorEntryTestCase( |
| 43 | + description="user purge delay is preserved", |
| 44 | + inherited_environment=((_PURGE_DELAY_VARIABLE, "250"),), |
| 45 | + expected_purge_delay="250", |
| 46 | + ), |
| 47 | + ], |
| 48 | + ids=lambda case: case.description, |
| 49 | +) |
| 50 | +def test_given_fresh_process_when_running_cli_entry_then_purge_delay_is_set_before_native_load( |
| 51 | + test_case: NativeAllocatorEntryTestCase, |
| 52 | +) -> None: |
| 53 | + environment: dict[str, str] = os.environ.copy() |
| 54 | + environment.pop(_PURGE_DELAY_VARIABLE, None) |
| 55 | + environment.update(test_case.inherited_environment) |
| 56 | + |
| 57 | + result: subprocess.CompletedProcess[str] = subprocess.run( |
| 58 | + [sys.executable, "-c", _SCRIPT], |
| 59 | + check=True, |
| 60 | + capture_output=True, |
| 61 | + env=environment, |
| 62 | + text=True, |
| 63 | + ) |
| 64 | + payload: dict[str, object] = json.loads(result.stdout.strip().splitlines()[-1]) |
| 65 | + |
| 66 | + assert payload == { |
| 67 | + "loaded_at_import": False, |
| 68 | + "loaded_after_main": True, |
| 69 | + "purge_delay": test_case.expected_purge_delay, |
| 70 | + "exit_code": 0, |
| 71 | + } |
| 72 | + |
| 73 | + |
| 74 | +if __name__ == "__main__": |
| 75 | + pytest.main([__file__, "-vv"]) |
0 commit comments