63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Автотесты лабораторной 3 (GIL)."""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from conftest import close, load_solution # noqa: E402
|
|
from common.tasks import run_sequential, smoke_params # noqa: E402
|
|
|
|
|
|
def test_variant_number_set():
|
|
mod = load_solution("lab03_gil")
|
|
assert 1 <= getattr(mod, "VARIANT_NUMBER", 0) <= 20, (
|
|
"Укажите свой VARIANT_NUMBER (1..20) в solution.py")
|
|
|
|
|
|
def test_run_task_on_threads_correct():
|
|
mod = load_solution("lab03_gil")
|
|
from common.tasks import get_variant
|
|
name = get_variant(mod.VARIANT_NUMBER)["task_name"]
|
|
sp = smoke_params(name)
|
|
|
|
expected, _ = run_sequential(name, sp, parts=1)
|
|
got, seconds = mod.run_task_on_threads(name, sp, p=4)
|
|
|
|
assert close(got, expected), f"checksum {got} != {expected}"
|
|
assert seconds >= 0
|
|
|
|
|
|
def test_run_task_on_threads_uses_pool():
|
|
src = (Path(__file__).resolve().parents[1] / "lab03_gil" / "solution.py"
|
|
).read_text(encoding="utf-8")
|
|
assert "ThreadPoolExecutor" in src or "threading.Thread" in src, (
|
|
"нужны именно потоки (ThreadPoolExecutor или Thread)")
|
|
|
|
|
|
def test_measure_cpu_threads_keys():
|
|
mod = load_solution("lab03_gil")
|
|
from common.tasks import get_variant
|
|
name = get_variant(mod.VARIANT_NUMBER)["task_name"]
|
|
sp = smoke_params(name)
|
|
res = mod.measure_cpu_threads(name, sp, [1, 2, 4])
|
|
assert sorted(res.keys()) == [1, 2, 4], f"ключи {sorted(res.keys())}"
|
|
assert all(t > 0 for t in res.values())
|
|
|
|
|
|
def test_measure_gil_release_parallel():
|
|
"""hashlib на блоках 1 МБ обязан показать S(4) > 1.4."""
|
|
mod = load_solution("lab03_gil")
|
|
res = mod.measure_gil_release([1, 4], block_size=1 << 20, blocks_per_part=8)
|
|
s4 = res[1] / res[4]
|
|
assert s4 > 1.4, (
|
|
f"S(4)={s4:.2f} — hashlib не ускорился потоками. Проверьте размер "
|
|
"блоков (нужны ≥ 256 КБ) и что общий объём делится между потоками")
|
|
|
|
|
|
def test_gil_status():
|
|
mod = load_solution("lab03_gil")
|
|
ver, gil = mod.gil_status()
|
|
assert isinstance(ver, str) and ver
|
|
assert isinstance(gil, bool)
|