import threading
import time

stop_event = threading.Event()

def worker():
    while not stop_event.is_set():
        print("Работаю...")
        time.sleep(0.3)
    print("Останавливаюсь корректно.")

thread = threading.Thread(target=worker)
thread.start()

time.sleep(1.2)
stop_event.set()   # посылаем сигнал остановки
thread.join()
print("Все потоки завершены корректно.")

import pytest
from hypothesis import given, strategies as st


# 1) Простой юнит-тест
def test_add():
    assert add(2, 3) == 5


2) Фикстура для окружения (временный файл)
@pytest.fixture
def temp_file(tmp_path):
    file_path = tmp_path / "data.txt"
    file_path.write_text("42")
    return file_path


def test_read_data(temp_file):
    assert read_data(temp_file) == 42


3) Property-based тест (генерация случайных входных данных)
@given(st.integers(), st.integers())
def test_add_random(a, b):
    assert add(a, b) == a + b


Быстрый запуск только упавших тестов:
pytest --lf

 простой «мягкий» троттлинг передачи данных

import asyncio

CHUNK = 32_000

async def send_smooth(writer, data: bytes):
    for i in range(0, len(data), CHUNK):
        writer.write(data[i:i+CHUNK])
        await writer.drain()
        await asyncio.sleep(0)  # отдаём управление циклу

async def handle(reader, writer):
    payload = b"x" * 5_000_000
    await send_smooth(writer, payload)
    writer.close()
    await writer.wait_closed()

async def main():
    srv = await asyncio.start_server(handle, "0.0.0.0", 8888)
    async with srv:
        await srv.serve_forever()

asyncio.run(main())

import os, hashlib
from http.server import SimpleHTTPRequestHandler, HTTPServer

STATIC = "static"
etag = {f: hashlib.md5(open(os.path.join(STATIC, f),"rb").read()).hexdigest()
        for f in os.listdir(STATIC)}

class H(SimpleHTTPRequestHandler):
    def end_headers(self):
        name = self.path.lstrip("/")
        if name in etag:
            self.send_header("ETag", etag[name])
        super().end_headers()

HTTPServer(("0.0.0.0", 8000), H).serve_forever()