62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Автотесты лабораторной 4 (multiprocessing).
|
|
|
|
Важная особенность: тесты запускают solution.py как ПОДПРОЦЕСС с переменной
|
|
окружения LAB_SMOKE=1 — это одновременно проверяет и корректность, и
|
|
Windows-совместимость (наличие if __name__ == "__main__").
|
|
"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
LAB_DIR = "lab04_multiprocessing"
|
|
|
|
|
|
def run_smoke():
|
|
"""Запускает solution.py в подпроцессе в smoke-режиме, возвращает JSON."""
|
|
env = dict(os.environ)
|
|
env["LAB_SMOKE"] = "1"
|
|
env["PYTHONIOENCODING"] = "utf-8"
|
|
proc = subprocess.run(
|
|
[sys.executable, str(ROOT / LAB_DIR / "solution.py")],
|
|
capture_output=True, text=True, env=env, timeout=120,
|
|
cwd=str(ROOT))
|
|
if proc.returncode != 0:
|
|
raise AssertionError(
|
|
f"solution.py упал:\n--- stdout ---\n{proc.stdout}\n--- stderr ---\n{proc.stderr}")
|
|
# smoke-режим печатает одну строку JSON с префиксом LAB_SMOKE_JSON:
|
|
for line in proc.stdout.splitlines():
|
|
if line.startswith("LAB_SMOKE_JSON:"):
|
|
return json.loads(line[len("LAB_SMOKE_JSON:"):])
|
|
raise AssertionError(
|
|
f"в smoke-режиме не найден LAB_SMOKE_JSON. stdout:\n{proc.stdout}")
|
|
|
|
|
|
def test_smoke_processes_work():
|
|
"""Запуск на процессах из подпроцесса: checksum корректен, p>1 отработал."""
|
|
out = run_smoke()
|
|
assert out.get("ok") is True, f"smoke-отчёт: {out}"
|
|
assert out.get("result_ok") is True, (
|
|
"checksum параллельного запуска не совпал с последовательным")
|
|
assert out.get("p") and out["p"] >= 2, "smoke должен запускаться при p>=2"
|
|
|
|
|
|
def test_smoke_overhead_measured():
|
|
out = run_smoke()
|
|
ov = out.get("overhead", {})
|
|
assert len(ov) >= 2, "measure_overhead должна вернуть замеры минимум для 2 размеров"
|
|
assert all(t > 0 for t in ov.values())
|
|
|
|
|
|
def test_uses_dispatch_kernel():
|
|
src = (ROOT / LAB_DIR / "solution.py").read_text(encoding="utf-8")
|
|
assert "dispatch_kernel" in src, (
|
|
"для Pool нужен пиккелируемый dispatch_kernel из common.tasks")
|
|
assert 'if __name__ == "__main__"' in src, (
|
|
"обязательна защита if __name__ == '__main__' (Windows/spawn)")
|