Python · Testing with pytest · Advanced

Integration Testing

5 tasks

Test multiple components working together against real services.

Unit vs integration vs end-to-end: when to use each

#
## The test pyramid Not all tests serve the same purpose. A healthy test suite uses three distinct layers: ``` ┌─────────────────┐ │ End-to-End │ few, slow, test the whole system ├─────────────────┤ │ Integration │ more, test real components together ├─────────────────┤ │ Unit │ many, fast, test one thing in isolation └─────────────────┘ ``` Most projects should have many unit tests, a moderate number of integration tests, and a small number of end-to-end tests. This is the classic shape because each layer has different costs and benefits. ## Unit tests: fast and isolated A unit test exercises a single function or class in complete isolation. Every external dependency — databases, HTTP APIs, the file system — is replaced with a mock or stub. ```python # Unit test: no real HTTP call def test_get_product_count_success(): with patch('myapp.requests.get') as mock_get: mock_get.return_value.json.return_value = {'count': 42} mock_get.return_value.status_code = 200 result = get_product_count() assert result == 42 ``` **Benefits:** milliseconds per test, no external dependencies, pinpoint failures. **Limitation:** mocks can drift from reality. If the API returns `{'total': 42}` instead of `{'count': 42}`, the unit test still passes — the mock was wrong. ## Integration tests: real components, real contracts An integration test exercises multiple real components together. For API testing, this means making actual HTTP requests to a real server, parsing real JSON, checking real status codes. ```python # Integration test: real HTTP call to the actual server def test_get_product_count(): resp = requests.get('https://apilearn.tukas.dev/api/products/') assert resp.status_code == 200 data = resp.json() assert 'results' in data result = len(data['results']) assert result >= 0 ``` **Benefits:** catches real integration bugs — schema mismatches, auth failures, unexpected redirects, actual response shapes. The test uses the same HTTP path your production code uses. **Costs:** slower (network round-trip), can be flaky (network or server issues), requires a real server, can leave data behind. ## End-to-end tests: the full system An end-to-end test drives the entire system through its real interface — typically a browser for web apps, or the full CLI/API surface. It tests that all layers (frontend, backend, database) work together. End-to-end tests are the most expensive: slow, brittle, hard to debug. Use them sparingly, for the most critical user journeys. ## What makes a test an integration test? A test is an integration test if it: - Makes real network calls (HTTP, gRPC, WebSocket) - Reads from or writes to a real database - Reads or writes real files on disk - Uses real external services (auth providers, payment gateways, queues) If the test uses mocks instead, it is a unit test regardless of how many functions it calls internally. ## The costs of integration tests **Slow:** a network round-trip takes 50–500 ms. A unit test takes 0.1 ms. One hundred integration tests take 5–50 seconds; one hundred unit tests take 0.01 seconds. **Flaky:** network failures, timeouts, rate limits, and server restarts all cause integration tests to fail for reasons unrelated to your code. A flaky test suite erodes trust in CI. **State pollution:** integration tests can create data that affects subsequent tests. A test that creates a user but does not clean it up can cause another test that checks "no users with this username" to fail. **Dependencies:** integration tests require a real server to be running. In CI, this means deploying a test environment, waiting for it to be healthy, then running tests. ## Test isolation strategies **Use unique identifiers:** when creating users or resources, generate unique values (timestamps, UUIDs) so tests do not collide with each other or with pre-existing data: ```python import time def test_register_and_login(): username = f'testuser_{int(time.time())}' ... ``` **Clean up after yourself:** if a test creates a resource, delete it at the end. Use fixture teardown for this: ```python @pytest.fixture def temp_user(): username = f'tmp_{int(time.time())}' # setup: create user resp = requests.post(BASE_URL + '/api/users/register/', json={...}) yield {'username': username, 'token': get_token(username)} # teardown: delete user requests.delete(BASE_URL + f'/api/users/{username}/', headers=auth_headers) ``` **Use a dedicated test account:** for tests that read data (not write), authenticate with a known test account rather than creating new users. The `testuser` / `TestUser2024!` account on `apilearn.tukas.dev` is suitable for this. **Session-scoped auth:** if many tests need authentication, log in once with a session-scoped fixture and share the token. This cuts login overhead from N round-trips to 1. ## Running integration tests separately from unit tests Mark integration tests and run them separately: ```ini # pytest.ini markers = integration: marks tests that make real network calls ``` ```python @pytest.mark.integration def test_products_api(): ... ``` ```bash pytest -m integration # only integration tests pytest -m "not integration" # only unit tests (fast CI) ``` This lets you run the fast unit suite on every commit and the slower integration suite nightly or on demand. ## Comparing the two approaches The same feature, implemented as unit test and integration test: ```python # The function under test import requests def get_product_count(base_url='https://apilearn.tukas.dev'): resp = requests.get(f'{base_url}/api/products/') resp.raise_for_status() return resp.json()['count'] # Unit test: fast, isolated, but mock can drift def test_get_product_count_unit(): from unittest.mock import patch with patch('requests.get') as mock_get: mock_get.return_value.status_code = 200 mock_get.return_value.json.return_value = {'count': 5} mock_get.return_value.raise_for_status.return_value = None assert get_product_count() == 5 # Integration test: slower, but tests the real API contract def test_get_product_count_integration(): import requests resp = requests.get('https://apilearn.tukas.dev/api/products/') assert resp.status_code == 200 data = resp.json() assert 'count' in data or 'results' in data # whichever the API returns ``` Notice that the unit test assumes the API returns `{'count': 5}`. If the real API returns `{'results': [...], 'count': 5}`, the unit test is testing a mock that matches reality — fine. But if the field is actually called `'total'`, the unit test passes while the integration test catches the mismatch.

Integration tests in practice: auth flows and shared sessions

#
## Project setup ```bash pip install requests pytest ``` ```ini # pytest.ini [pytest] markers = integration: marks tests that make real network calls ``` All examples use `https://apilearn.tukas.dev`. ## Example 1: Auth flow — login and access a protected endpoint ```python # test_auth_flow.py import requests BASE_URL = 'https://apilearn.tukas.dev' def test_login_and_profile(): # Step 1: authenticate auth_resp = requests.post(f'{BASE_URL}/api/auth/token/', json={ 'username': 'testuser', 'password': 'TestUser2024!', }) assert auth_resp.status_code == 200 token = auth_resp.json()['token'] assert token # non-empty string # Step 2: use token to access a protected endpoint headers = {'Authorization': f'Token {token}'} profile_resp = requests.get(f'{BASE_URL}/api/users/profile/', headers=headers) assert profile_resp.status_code == 200 profile = profile_resp.json() assert profile['username'] == 'testuser' ``` This is a genuine integration test: it makes two real HTTP requests, parses real JSON, and asserts on the real response. If the auth service changes the token field name, or the profile endpoint changes `username` to `user`, this test catches it immediately. ## Example 2: Session-scoped auth fixture Logging in for every test is slow and wasteful. A session-scoped fixture logs in once: ```python # conftest.py import pytest import requests BASE_URL = 'https://apilearn.tukas.dev' @pytest.fixture(scope='session') def base_url(): return BASE_URL @pytest.fixture(scope='session') def auth_headers(base_url): resp = requests.post(f'{base_url}/api/auth/token/', json={ 'username': 'testuser', 'password': 'TestUser2024!', }) resp.raise_for_status() token = resp.json()['token'] return {'Authorization': f'Token {token}'} ``` ```python # test_protected.py def test_profile_username(base_url, auth_headers): resp = requests.get(f'{base_url}/api/users/profile/', headers=auth_headers) assert resp.status_code == 200 assert resp.json()['username'] == 'testuser' def test_profile_has_email(base_url, auth_headers): resp = requests.get(f'{base_url}/api/users/profile/', headers=auth_headers) assert resp.status_code == 200 assert '@' in resp.json().get('email', '') def test_products_authenticated(base_url, auth_headers): resp = requests.get(f'{base_url}/api/products/', headers=auth_headers) assert resp.status_code == 200 ``` One HTTP login for the entire test session. All three tests share the same `auth_headers` dict. Session scope is appropriate here because the token does not change between tests. ## Example 3: Pagination as integration concern Pagination contracts (page size, result count, `next` URL) are exactly the kind of thing unit tests with mocks cannot verify: ```python # test_pagination.py def test_default_page_returns_results(base_url): resp = requests.get(f'{base_url}/api/products/') assert resp.status_code == 200 data = resp.json() assert 'results' in data assert len(data['results']) > 0 def test_page_size_respected(base_url): resp = requests.get(f'{base_url}/api/products/?page_size=2') assert resp.status_code == 200 results = resp.json()['results'] assert len(results) <= 2 def test_page_two_exists_when_enough_products(base_url): first = requests.get(f'{base_url}/api/products/?page=1&page_size=2') assert first.status_code == 200 data = first.json() if data.get('next'): second = requests.get(f'{base_url}/api/products/?page=2&page_size=2') assert second.status_code == 200 assert isinstance(second.json()['results'], list) ``` ## Example 4: Unit test vs integration test — side by side The same `get_product_names` function, tested both ways: ```python # app.py import requests def get_product_names(base_url='https://apilearn.tukas.dev'): resp = requests.get(f'{base_url}/api/products/') resp.raise_for_status() return [p['name'] for p in resp.json()['results']] ``` ```python # Unit test — fast, but mock can diverge from real API from unittest.mock import patch def test_get_product_names_unit(): with patch('requests.get') as mock_get: mock_get.return_value.status_code = 200 mock_get.return_value.raise_for_status.return_value = None mock_get.return_value.json.return_value = { 'results': [{'name': 'Widget'}, {'name': 'Gadget'}] } names = get_product_names() assert names == ['Widget', 'Gadget'] # This test passes even if the real API returns {'name': None} for some items. # Integration test — slower, but verifies the real contract def test_get_product_names_integration(): names = get_product_names() # We don't know exact values, so assert on structure: assert isinstance(names, list) assert len(names) > 0 assert all(isinstance(n, str) for n in names) assert all(len(n) > 0 for n in names) ``` The integration test cannot assert exact values (product names change) but verifies the contract: `get_product_names` returns a non-empty list of non-empty strings. This will fail if the API returns `null` for `name`, or if the structure changes — exactly the bugs that matter in production. ## Example 5: Registration flow with unique data When tests create data, use unique identifiers to avoid collisions: ```python # test_registration.py import time import requests BASE_URL = 'https://apilearn.tukas.dev' def test_register_and_login(): # Generate a unique username for this test run username = f'pytest_{int(time.time())}' # Step 1: register reg_resp = requests.post(f'{BASE_URL}/api/users/register/', json={ 'username': username, 'email': f'{username}@example.com', 'password': 'Secure123!', 'first_name': 'Test', 'last_name': 'User', }) assert reg_resp.status_code in (200, 201) # Step 2: login with new credentials auth_resp = requests.post(f'{BASE_URL}/api/auth/token/', json={ 'username': username, 'password': 'Secure123!', }) assert auth_resp.status_code == 200 token = auth_resp.json()['token'] # Step 3: verify profile profile_resp = requests.get(f'{BASE_URL}/api/users/profile/', headers={'Authorization': f'Token {token}'}) assert profile_resp.status_code == 200 assert profile_resp.json()['username'] == username ``` `f'pytest_{int(time.time())}'` generates a username like `pytest_1735689600` — unique per second. If tests run in parallel, append a random suffix too.

Integration testing reference card

#
## Test classification | Type | Real network? | Real DB? | Speed | Use for | |---|---|---|---|---| | Unit | No (mocked) | No | Fast | Logic, algorithms, transformations | | Integration | Yes | Yes | Slow | API contracts, auth flows, data pipelines | | End-to-end | Yes | Yes | Slowest | Full user journeys | ## Session-scoped auth fixture pattern ```python # conftest.py import pytest import requests BASE_URL = 'https://apilearn.tukas.dev' @pytest.fixture(scope='session') def base_url(): return BASE_URL @pytest.fixture(scope='session') def auth_headers(base_url): resp = requests.post(f'{base_url}/api/auth/token/', json={ 'username': 'testuser', 'password': 'TestUser2024!', }) resp.raise_for_status() return {'Authorization': f'Token {resp.json()["token"]}'} ``` ## Common API assertions ```python # Status assert resp.status_code == 200 assert resp.status_code in (200, 201) # JSON structure data = resp.json() assert 'results' in data assert isinstance(data['results'], list) # Pagination assert len(data['results']) <= page_size # Field presence assert all('slug' in p for p in data['results']) # Auth required (should redirect or 401 without token) resp = requests.get(url) # no headers assert resp.status_code == 401 ``` ## Unique test data (avoid collisions) ```python import time, uuid username = f'test_{int(time.time())}' # unique per second username = f'test_{uuid.uuid4().hex[:8]}' # unique per call ``` ## Marking and selecting integration tests ```ini # pytest.ini markers = integration: marks tests that require network access ``` ```python @pytest.mark.integration def test_something(): ... ``` ```bash pytest -m integration # integration only pytest -m "not integration" # unit only (fast) ``` ## Test isolation checklist - [ ] Use unique usernames/IDs when creating resources - [ ] Clean up created resources in fixture teardown - [ ] Use session scope for authentication (login once) - [ ] Assert on structure and constraints, not exact values that can change - [ ] Handle both 200 and 201 where either is valid
01

Registration and login integration flow

#

Write an integration test `test_register_and_login` that: 1. Generates a unique username using `f'pytest_{int(time.time())}'` 2. Registers the user via `POST /api/users/register/` with fields: `username`, `email`, `password`, `first_name`, `last_name` 3. Asserts the registration response is 200 or 201 4. Logs in via `POST /api/auth/token/` using the new credentials 5. GETs `/api/users/profile/` with the returned token 6. Asserts the profile's `username` matches the one registered Use `BASE_URL = 'https://apilearn.tukas.dev'`.

# test_register_flow.py
import time
import requests

BASE_URL = 'https://apilearn.tukas.dev'


def test_register_and_login():
    username = f'pytest_{int(time.time())}'

    # Step 1: register
    reg = requests.post(f'{BASE_URL}/api/users/register/', json={
        'username': username,
        'email': f'{username}@example.com',
        'password': ...,
        'first_name': 'Test',
        'last_name': 'User',
    })
    assert reg.status_code in (200, 201)

    # Step 2: login
    auth = requests.post(f'{BASE_URL}/api/auth/token/', json={
        'username': username,
        'password': ...,
    })
    assert auth.status_code == 200
    token = ...

    # Step 3: get profile
    profile = requests.get(
        f'{BASE_URL}/api/users/profile/',
        headers={'Authorization': f'Token {token}'},
    )
    assert profile.status_code == 200
    assert ...
Solution
# test_register_flow.py
import time
import requests

BASE_URL = 'https://apilearn.tukas.dev'


def test_register_and_login():
    username = f'pytest_{int(time.time())}'
    password = 'Secure123!'

    reg = requests.post(f'{BASE_URL}/api/users/register/', json={
        'username': username,
        'email': f'{username}@example.com',
        'password': password,
        'first_name': 'Test',
        'last_name': 'User',
    })
    assert reg.status_code in (200, 201)

    auth = requests.post(f'{BASE_URL}/api/auth/token/', json={
        'username': username,
        'password': password,
    })
    assert auth.status_code == 200
    token = auth.json()['token']

    profile = requests.get(
        f'{BASE_URL}/api/users/profile/',
        headers={'Authorization': f'Token {token}'},
    )
    assert profile.status_code == 200
    assert profile.json()['username'] == username
02

Test pagination constraints

#

Write a session-scoped `base_url` fixture returning `'https://apilearn.tukas.dev'`, then write three tests: 1. `test_products_returns_results` — GET `/api/products/`, assert status 200, `'results'` key exists, result is a list with at least 1 item 2. `test_page_size_two` — GET `/api/products/?page_size=2`, assert the results list has **at most** 2 items 3. `test_second_page` — GET `/api/products/?page=2&page_size=2`, assert status 200 (page 2 may be empty but should not 404)

# conftest.py
import pytest


@pytest.fixture(scope='session')
def base_url():
    return 'https://apilearn.tukas.dev'


# test_pagination.py
import requests


def test_products_returns_results(base_url):
    resp = requests.get(f'{base_url}/api/products/')
    ...


def test_page_size_two(base_url):
    resp = requests.get(f'{base_url}/api/products/?page_size=2')
    ...


def test_second_page(base_url):
    resp = requests.get(f'{base_url}/api/products/?page=2&page_size=2')
    ...
Solution
# conftest.py
import pytest


@pytest.fixture(scope='session')
def base_url():
    return 'https://apilearn.tukas.dev'


# test_pagination.py
import requests


def test_products_returns_results(base_url):
    resp = requests.get(f'{base_url}/api/products/')
    assert resp.status_code == 200
    data = resp.json()
    assert 'results' in data
    assert isinstance(data['results'], list)
    assert len(data['results']) > 0


def test_page_size_two(base_url):
    resp = requests.get(f'{base_url}/api/products/?page_size=2')
    assert resp.status_code == 200
    results = resp.json()['results']
    assert len(results) <= 2


def test_second_page(base_url):
    resp = requests.get(f'{base_url}/api/products/?page=2&page_size=2')
    assert resp.status_code == 200
03

Session-scoped authentication for multiple tests

#

Write a `conftest.py` with two session-scoped fixtures: - `base_url` — returns `'https://apilearn.tukas.dev'` - `auth_headers` — logs in as `testuser` / `TestUser2024!` and returns `{'Authorization': 'Token ...'}` Write three tests that each use `auth_headers`: 1. `test_profile_username` — assert profile response has `username == 'testuser'` 2. `test_products_with_auth` — assert GET `/api/products/` returns 200 (authenticated) 3. `test_echo_with_auth` — assert GET `/api/echo/` returns 200

# conftest.py
import pytest
import requests


@pytest.fixture(scope='session')
def base_url():
    return 'https://apilearn.tukas.dev'


@pytest.fixture(scope='session')
def auth_headers(base_url):
    resp = requests.post(f'{base_url}/api/auth/token/', json={
        'username': 'testuser',
        'password': ...,
    })
    resp.raise_for_status()
    token = ...
    return {'Authorization': f'Token {token}'}


# test_authenticated.py
import requests


def test_profile_username(base_url, auth_headers):
    ...


def test_products_with_auth(base_url, auth_headers):
    ...


def test_echo_with_auth(base_url, auth_headers):
    ...
Solution
# conftest.py
import pytest
import requests


@pytest.fixture(scope='session')
def base_url():
    return 'https://apilearn.tukas.dev'


@pytest.fixture(scope='session')
def auth_headers(base_url):
    resp = requests.post(f'{base_url}/api/auth/token/', json={
        'username': 'testuser',
        'password': 'TestUser2024!',
    })
    resp.raise_for_status()
    token = resp.json()['token']
    return {'Authorization': f'Token {token}'}


# test_authenticated.py
import requests


def test_profile_username(base_url, auth_headers):
    resp = requests.get(f'{base_url}/api/users/profile/', headers=auth_headers)
    assert resp.status_code == 200
    assert resp.json()['username'] == 'testuser'


def test_products_with_auth(base_url, auth_headers):
    resp = requests.get(f'{base_url}/api/products/', headers=auth_headers)
    assert resp.status_code == 200


def test_echo_with_auth(base_url, auth_headers):
    resp = requests.get(f'{base_url}/api/echo/', headers=auth_headers)
    assert resp.status_code == 200
04

The same feature: mocked vs real HTTP

#

Implement `get_product_count(base_url)` — a function that GETs `{base_url}/api/products/` and returns the integer value of the `count` field in the response JSON. Write two tests for it: 1. `test_get_product_count_unit` — use `unittest.mock.patch` to mock `requests.get`, make it return a fake JSON response `{'count': 7, 'results': []}`, and assert `get_product_count(...)` returns `7`. The test should **not** make a real network call. 2. `test_get_product_count_integration` — call `get_product_count('https://apilearn.tukas.dev')` for real, assert the result is an integer >= 0. Add a comment in each test explaining what it can and cannot catch.

# test_count_comparison.py
import requests
from unittest.mock import patch


def get_product_count(base_url):
    resp = requests.get(f'{base_url}/api/products/')
    resp.raise_for_status()
    return resp.json()['count']


def test_get_product_count_unit():
    # This test: fast, no network.
    # Cannot catch: field renamed in real API, wrong URL, auth issues.
    with patch('requests.get') as mock_get:
        mock_get.return_value.status_code = 200
        mock_get.return_value.raise_for_status.return_value = None
        mock_get.return_value.json.return_value = ...
        result = get_product_count('http://fake')
    assert result == ...


def test_get_product_count_integration():
    # This test: slower, real network.
    # Cannot catch: logic bugs inside get_product_count (covered by unit test).
    result = get_product_count('https://apilearn.tukas.dev')
    assert ...
Solution
# test_count_comparison.py
import requests
from unittest.mock import patch


def get_product_count(base_url):
    resp = requests.get(f'{base_url}/api/products/')
    resp.raise_for_status()
    return resp.json()['count']


def test_get_product_count_unit():
    # This test: fast, no network.
    # Cannot catch: field renamed in real API, wrong URL, auth issues.
    with patch('requests.get') as mock_get:
        mock_get.return_value.status_code = 200
        mock_get.return_value.raise_for_status.return_value = None
        mock_get.return_value.json.return_value = {'count': 7, 'results': []}
        result = get_product_count('http://fake')
    assert result == 7


def test_get_product_count_integration():
    # This test: slower, real network.
    # Cannot catch: logic bugs inside get_product_count (covered by unit test).
    result = get_product_count('https://apilearn.tukas.dev')
    assert isinstance(result, int)
    assert result >= 0
05

Session-scoped fixture used across three independent tests

#

Build a session-scoped fixture `product_list` that: - GETs `https://apilearn.tukas.dev/api/products/` - Asserts status 200 - Returns the parsed `results` list Write three independent tests that each receive `product_list` as a fixture parameter: 1. `test_products_not_empty` — assert the list has at least 1 item 2. `test_products_have_slug` — assert every product has a non-empty `slug` field 3. `test_products_have_name` — assert every product has a non-empty `name` field The HTTP call should happen **once** regardless of how many tests use the fixture.

# conftest.py
import pytest
import requests


@pytest.fixture(scope='session')
def product_list():
    resp = requests.get('https://apilearn.tukas.dev/api/products/')
    assert resp.status_code == 200
    return resp.json()['results']


# test_product_list.py


def test_products_not_empty(product_list):
    assert len(product_list) > 0


def test_products_have_slug(product_list):
    ...


def test_products_have_name(product_list):
    ...
Solution
# conftest.py
import pytest
import requests


@pytest.fixture(scope='session')
def product_list():
    resp = requests.get('https://apilearn.tukas.dev/api/products/')
    assert resp.status_code == 200
    return resp.json()['results']


# test_product_list.py


def test_products_not_empty(product_list):
    assert len(product_list) > 0


def test_products_have_slug(product_list):
    assert all(isinstance(p.get('slug'), str) and len(p['slug']) > 0
               for p in product_list)


def test_products_have_name(product_list):
    assert all(isinstance(p.get('name'), str) and len(p['name']) > 0
               for p in product_list)