58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Автотесты лабораторной 2 (многопоточный код)."""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from conftest import load_solution, source_of # noqa: E402
|
|
|
|
|
|
def test_variant_number_set():
|
|
mod = load_solution("lab02_threads")
|
|
assert 1 <= getattr(mod, "VARIANT_NUMBER", 0) <= 20, (
|
|
"Укажите свой VARIANT_NUMBER (1..20) в solution.py")
|
|
|
|
|
|
def test_manual_threads_correct():
|
|
mod = load_solution("lab02_threads")
|
|
results, seconds = mod.run_io_threads_manual(8, 0.0, p=3)
|
|
assert [r[0] for r in results] == list(range(8)), (
|
|
"результаты должны быть упорядочены по item")
|
|
assert seconds >= 0
|
|
|
|
|
|
def test_manual_threads_parallel():
|
|
"""8 запросов по 0.1 c на 4 потоках должны занять ~0.2 c, не ~0.8 c."""
|
|
mod = load_solution("lab02_threads")
|
|
_, seconds = mod.run_io_threads_manual(8, 0.1, p=4)
|
|
assert seconds < 0.6, (
|
|
f"выполнено за {seconds:.2f} c — похоже, потоки работают по очереди "
|
|
"(start() до join()!)")
|
|
|
|
|
|
def test_pool_threads_correct():
|
|
mod = load_solution("lab02_threads")
|
|
results, seconds = mod.run_io_threads_pool(8, 0.0, p=3)
|
|
assert [r[0] for r in results] == list(range(8))
|
|
assert seconds >= 0
|
|
|
|
|
|
def test_pool_threads_parallel():
|
|
mod = load_solution("lab02_threads")
|
|
_, seconds = mod.run_io_threads_pool(8, 0.1, p=4)
|
|
assert seconds < 0.6, (
|
|
f"выполнено за {seconds:.2f} c — пул не даёт параллельности")
|
|
|
|
|
|
def test_uses_threading():
|
|
src = source_of("lab02_threads")
|
|
assert "threading" in src or "ThreadPoolExecutor" in src, (
|
|
"в решении должны использоваться потоки")
|
|
|
|
|
|
def test_manual_uses_thread_class():
|
|
src = source_of("lab02_threads")
|
|
assert "threading.Thread" in src, (
|
|
"run_io_threads_manual должна использовать threading.Thread")
|