14 промтов для написания unit-тестов и интеграционных тестов (pytest, Jest, unittest)

Напиши unit-тест на pytest для функции `add(a, b)`, которая возвращает сумму двух чисел. Включи тесты: 1) сложение положительных чисел, 2) сложение с нулём, 3) сложение отрицательных чисел, 4) граничные случаи (float, большие числа). Используй параметризацию.

import pytest

def add(a, b):
    return a + b

@pytest.mark.parametrize("a, b, expected", [
    (1, 2, 3),
    (0, 5, 5),
    (-1, -2, -3),
    (1.5, 2.5, 4.0),
    (10**10, 10**10, 2*10**10),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

Напиши тест на unittest для функции `fetch_user_data(user_id)`, которая делает HTTP GET запрос к внешнему API. Используй mock, чтобы имитировать ответ сервера. Проверь обработку успешного ответа (код 200) и ошибки (код 404).

import unittest
from unittest.mock import patch
import requests

def fetch_user_data(user_id):
    response = requests.get(f"https://api.example.com/users/{user_id}")
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 404:
        return None
    raise Exception("API error")

class TestFetchUser(unittest.TestCase):
    @patch("requests.get")
    def test_success(self, mock_get):
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"id": 1, "name": "Alice"}
        result = fetch_user_data(1)
        self.assertEqual(result["name"], "Alice")

    @patch("requests.get")
    def test_not_found(self, mock_get):
        mock_get.return_value.status_code = 404
        result = fetch_user_data(999)
        self.assertIsNone(result)

if __name__ == "__main__":
    unittest.main()

Напиши тест на pytest для функции `divide(a, b)`, которая выбрасывает ValueError при делении на ноль. Также проверь корректное деление.

import pytest

def divide(a, b):
    if b == 0:
        raise ValueError("Division by zero")
    return a / b

def test_divide_zero():
    with pytest.raises(ValueError, match="Division by zero"):
        divide(10, 0)

def test_divide_correct():
    assert divide(10, 2) == 5.0

Создай тесты на pytest для класса `BankAccount` с методами `deposit(amount)` и `withdraw(amount)`. Проверь: 1) депозит увеличивает баланс, 2) снятие уменьшает баланс, 3) снятие больше баланса вызывает исключение, 4) начальный баланс равен 0.

import pytest

class BankAccount:
    def __init__(self):
        self.balance = 0

    def deposit(self, amount):
        if amount <= 0:
            raise ValueError("Amount must be positive")
        self.balance += amount

    def withdraw(self, amount):
        if amount > self.balance:
            raise ValueError("Insufficient funds")
        self.balance -= amount

class TestBankAccount:
    def test_initial_balance(self):
        acc = BankAccount()
        assert acc.balance == 0

    def test_deposit(self):
        acc = BankAccount()
        acc.deposit(100)
        assert acc.balance == 100

    def test_withdraw(self):
        acc = BankAccount()
        acc.deposit(100)
        acc.withdraw(50)
        assert acc.balance == 50

    def test_withdraw_insufficient(self):
        acc = BankAccount()
        with pytest.raises(ValueError, match="Insufficient funds"):
            acc.withdraw(10)

Напиши тесты на Jest для функции `multiply(a, b)`, которая перемножает два числа. Включи тесты: 1) обычные числа, 2) умножение на ноль, 3) умножение на дробь, 4) отрицательные числа.

function multiply(a, b) {
    return a * b;
}

test('multiply positive numbers', () => {
    expect(multiply(2, 3)).toBe(6);
});

test('multiply by zero', () => {
    expect(multiply(5, 0)).toBe(0);
});

test('multiply with decimal', () => {
    expect(multiply(2.5, 2)).toBe(5);
});

test('multiply negative numbers', () => {
    expect(multiply(-2, -3)).toBe(6);
});

Напиши тест на pytest с фикстурой, которая создаёт временную базу данных SQLite в памяти. Проверь, что функция `get_user(db, user_id)` возвращает правильное имя.

import pytest
import sqlite3

def get_user(db, user_id):
    cursor = db.execute("SELECT name FROM users WHERE id=?", (user_id,))
    row = cursor.fetchone()
    return row[0] if row else None

@pytest.fixture
def db():
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users (id INT, name TEXT)")
    conn.execute("INSERT INTO users VALUES (1, 'Bob')")
    yield conn
    conn.close()

def test_get_user(db):
    result = get_user(db, 1)
    assert result == "Bob"

def test_get_user_not_found(db):
    result = get_user(db, 999)
    assert result is None

Напиши тест на pytest для асинхронной функции `fetch_page(url)`, которая возвращает содержимое страницы через aiohttp. Используй pytest-asyncio и mock для эмуляции ответа.

import pytest
from unittest.mock import AsyncMock
import aiohttp

async def fetch_page(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

@pytest.mark.asyncio
async def test_fetch_page(mocker):
    mock_get = AsyncMock()
    mock_get.return_value.__aenter__.return_value.text = AsyncMock(return_value="<html>Test</html>")
    mocker.patch("aiohttp.ClientSession.get", return_value=mock_get)
    result = await fetch_page("http://example.com")
    assert "Test" in result

Напиши интеграционный тест на pytest, который подключается к тестовой PostgreSQL (через testcontainers), выполняет операции вставки и выборки, и проверяет корректность данных. Используй библиотеку `psycopg2`.

import pytest
import psycopg2
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="module")
def postgres_db():
    with PostgresContainer("postgres:16") as postgres:
        conn = psycopg2.connect(
            host=postgres.get_container_host_ip(),
            port=postgres.get_exposed_port(5432),
            user=postgres.POSTGRES_USER,
            password=postgres.POSTGRES_PASSWORD,
            dbname=postgres.POSTGRES_DB
        )
        yield conn
        conn.close()

def test_insert_and_select(postgres_db):
    cursor = postgres_db.cursor()
    cursor.execute("CREATE TABLE test (id INT, data TEXT)")
    cursor.execute("INSERT INTO test VALUES (1, 'hello')")
    postgres_db.commit()
    cursor.execute("SELECT data FROM test WHERE id=1")
    result = cursor.fetchone()
    assert result[0] == "hello"

Напиши интеграционный тест на pytest с использованием библиотеки `requests`. Проверь, что POST запрос к эндпоинту `/api/users` создаёт пользователя и возвращает статус 201 и правильный JSON.

import pytest
import requests

BASE_URL = "http://test-server:8000"

def test_create_user():
    payload = {"name": "Alice", "email": "alice@example.com"}
    response = requests.post(f"{BASE_URL}/api/users", json=payload)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert "id" in data

Напиши интеграционный тест на Jest для функции `sendEmail(user)`, которая отправляет письмо через сервис SendGrid. Используй mock для `@sendgrid/mail`.

const sgMail = require('@sendgrid/mail');

function sendEmail(user) {
    const msg = {
        to: user.email,
        from: 'test@example.com',
        subject: 'Welcome',
        text: `Hello ${user.name}`,
    };
    return sgMail.send(msg);
}

jest.mock('@sendgrid/mail', () => ({
    send: jest.fn().mockResolvedValue([{ statusCode: 202 }]),
}));

test('sendEmail sends email', async () => {
    const user = { email: 'bob@test.com', name: 'Bob' };
    const result = await sendEmail(user);
    expect(sgMail.send).toHaveBeenCalledWith(expect.objectContaining({
        to: 'bob@test.com',
    }));
    expect(result[0].statusCode).toBe(202);
});

Напиши интеграционный тест на pytest, который проверяет миграцию данных из старой схемы в новую. Используй SQLite для имитации.

import pytest
import sqlite3

def migrate(old_conn, new_conn):
    old_cursor = old_conn.execute("SELECT id, name FROM users_old")
    for row in old_cursor:
        new_conn.execute("INSERT INTO users_new (user_id, full_name) VALUES (?, ?)", (row[0], row[1]))
    new_conn.commit()

@pytest.fixture
def old_db():
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users_old (id INT, name TEXT)")
    conn.execute("INSERT INTO users_old VALUES (1, 'Alice')")
    return conn

@pytest.fixture
def new_db():
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users_new (user_id INT, full_name TEXT)")
    return conn

def test_migrate(old_db, new_db):
    migrate(old_db, new_db)
    cursor = new_db.execute("SELECT * FROM users_new")
    rows = cursor.fetchall()
    assert len(rows) == 1
    assert rows[0] == (1, "Alice")

Напиши интеграционный тест на pytest, который создаёт временный файл, вызывает функцию `process_file(path)`, и проверяет результат.

import pytest
import tempfile
import os

def process_file(path):
    with open(path, 'r') as f:
        data = f.read()
    return data.upper()

def test_process_file():
    with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
        f.write("hello world")
        temp_path = f.name
    try:
        result = process_file(temp_path)
        assert result == "HELLO WORLD"
    finally:
        os.remove(temp_path)

Напиши интеграционный тест на pytest, который подключается к Redis (через testcontainers), записывает и читает значение.

import pytest
import redis
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="module")
def redis_client():
    with RedisContainer("redis:7") as redis_container:
        client = redis.Redis(
            host=redis_container.get_container_host_ip(),
            port=redis_container.get_exposed_port(6379),
            decode_responses=True
        )
        yield client

def test_cache_set_get(redis_client):
    redis_client.set("key", "value")
    assert redis_client.get("key") == "value"

Напиши интеграционный тест на pytest, который поднимает RabbitMQ через testcontainers, отправляет сообщение в очередь и проверяет его получение.

import pytest
import pika
from testcontainers.rabbitmq import RabbitMQContainer

@pytest.fixture(scope="module")
def rabbit_connection():
    with RabbitMQContainer("rabbitmq:3") as rabbitmq:
        connection = pika.BlockingConnection(
            pika.URLParameters(f"amqp://guest:guest@{rabbitmq.get_container_host_ip()}:{rabbitmq.get_exposed_port(5672)}/")
        )
        yield connection
        connection.close()

def test_send_receive(rabbit_connection):
    channel = rabbit_connection.channel()
    channel.queue_declare(queue="test_queue")
    channel.basic_publish(exchange='', routing_key="test_queue", body="Hello")
    method_frame, header_frame, body = channel.basic_get(queue="test_queue", auto_ack=True)
    assert body == b"Hello"