Weeks 4–6 — Threading & Multiprocessing · Asyncio · Testing with pytest

import threading, time

def fetch(url, results, idx):
    time.sleep(0.5)   # simulate network I/O
    results[idx] = f"data from {url}"

urls = ["http://a.com", "http://b.com", "http://c.com"]
results = [None] * len(urls)
threads = [threading.Thread(target=fetch, args=(u, results, i)) for i, u in enumerate(urls)]
for t in threads: t.start()
for t in threads: t.join()
print(results)

import threading
from queue import Queue

lock = threading.Lock()
counter = 0

def increment(n):
    global counter
    for _ in range(n):
        with lock:
            counter += 1

t1 = threading.Thread(target=increment, args=(10000,))
t2 = threading.Thread(target=increment, args=(10000,))
t1.start(); t2.start()
t1.join(); t2.join()
print(counter)   # always 20000 with lock

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

# I/O-bound: use ThreadPoolExecutor
def download(url):
    time.sleep(0.5)   # simulate
    return f"result:{url}"

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(download, f"http://site{i}.com") for i in range(8)]
    results = [f.result() for f in futures]
print(len(results), "downloads done")

# CPU-bound: use ProcessPoolExecutor
def is_prime(n):
    if n < 2: return False
    return all(n % i for i in range(2, int(n**0.5)+1))

with ProcessPoolExecutor() as executor:
    primes = list(executor.map(is_prime, range(1000)))

import asyncio

async def greet(name, delay):
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")

async def main():
    # Run concurrently
    await asyncio.gather(
        greet("Alice", 2),
        greet("Bob", 1),
        greet("Carol", 0.5),
    )
    # Prints Carol, Bob, Alice (in order of completion)

asyncio.run(main())

import asyncio

async def background_task():
    for i in range(100):
        print(f"  bg: step {i}")
        await asyncio.sleep(0.1)

async def main():
    task = asyncio.create_task(background_task())
    await asyncio.sleep(0.3)   # let it run for 0.3s
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled")

asyncio.run(main())

import asyncio, aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return resp.status, len(await resp.text())

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            print(f"FAIL {url}: {result}")
        else:
            status, size = result
            print(f"{status} {url} ({size} chars)")

# test_calculator.py
import pytest

def add(a, b): return a + b
def divide(a, b):
    if b == 0: raise ZeroDivisionError
    return a / b

def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

# Run: pytest test_calculator.py -v

import pytest
from bank import BankAccount

@pytest.fixture
def account():
    """Fresh account with $100 balance."""
    acct = BankAccount("test")
    acct.deposit(100)
    return acct

def test_deposit(account):
    account.deposit(50)
    assert account.balance == 150

def test_withdraw(account):
    account.withdraw(30)
    assert account.balance == 70

def test_overdraft(account):
    with pytest.raises(ValueError, match="Insufficient"):
        account.withdraw(200)

import pytest
from unittest.mock import patch, MagicMock

@pytest.mark.parametrize("a,b,expected", [
    (2, 3, 5), (-1, 1, 0), (0, 0, 0), (100, -50, 50)
])
def test_add_param(a, b, expected):
    assert add(a, b) == expected

def test_with_mock():
    with patch("requests.get") as mock_get:
        mock_get.return_value = MagicMock(status_code=200, json=lambda: {"ok": True})
        # code that calls requests.get will use the mock
        resp = requests.get("http://example.com")
        assert resp.status_code == 200