Testing with pytest¶

from acme_data_pipeline.pipeline import normalise_event, EventSchema
import pytest

# ── Basic unit test ───────────────────────────────────────────────
def test_normalise_strips_whitespace():
    raw = {"user_id": "  u123  ", "event": "click", "value": 42}
    result = normalise_event(raw)
    assert result["user_id"] == "u123"    # pytest rewrites assertion on failure

def test_normalise_rejects_negative_value():
    with pytest.raises(ValueError, match="value must be non-negative"):
        normalise_event({"user_id": "u1", "event": "click", "value": -1})

def test_event_schema_serialises_to_dict():
    event = EventSchema(user_id="u1", event="click", value=10)
    d = event.model_dump()
    assert d["user_id"] == "u1"
    assert d["value"] == 10
    assert "timestamp" in d    # schema auto-adds timestamp

# ── Grouping related tests in a class (no inheritance needed) ────
class TestNormaliseEdgeCases:
    def test_zero_value_is_valid(self):
        result = normalise_event({"user_id": "u1", "event": "view", "value": 0})
        assert result["value"] == 0

    def test_missing_user_id_raises(self):
        with pytest.raises(KeyError):
            normalise_event({"event": "click", "value": 5})

import pytest
import sqlite3
from acme_data_pipeline.db import create_schema, UserRepository

# scope="session" → created once for the entire test run (expensive resources)
# scope="function" → default: fresh copy for each test (safe isolation)
# scope="module"   → one per test file

@pytest.fixture(scope="function")
def db_conn():
    """In-memory SQLite database — isolated per test, no cleanup needed."""
    conn = sqlite3.connect(":memory:")
    create_schema(conn)
    yield conn               # tests run here; everything after yield is teardown
    conn.close()

@pytest.fixture
def user_repo(db_conn):
    """Depends on db_conn — pytest resolves the dependency graph automatically."""
    return UserRepository(db_conn)

@pytest.fixture
def sample_users():
    return [
        {"user_id": "u1", "email": "ada@acme.io",    "tier": "enterprise"},
        {"user_id": "u2", "email": "grace@acme.io",  "tier": "starter"},
        {"user_id": "u3", "email": "alan@acme.io",   "tier": "enterprise"},
    ]

# ── Using fixtures in tests ──────────────────────────────────────
def test_insert_and_retrieve_user(user_repo, sample_users):
    user_repo.bulk_insert(sample_users[:1])
    user = user_repo.get("u1")
    assert user["email"] == "ada@acme.io"

def test_enterprise_users_count(user_repo, sample_users):
    user_repo.bulk_insert(sample_users)
    count = user_repo.count_by_tier("enterprise")
    assert count == 2

import pytest
from acme_data_pipeline.pipeline import normalise_event

# Each tuple becomes a separate test case with its own PASS/FAIL line
@pytest.mark.parametrize("raw_id, expected", [
    ("  u123  ",  "u123"),     # leading/trailing whitespace
    ("U123",      "u123"),     # uppercase normalised to lower
    ("u-123",     "u-123"),    # hyphens preserved
    ("",          None),       # empty string returns None (edge case)
])
def test_user_id_normalisation(raw_id, expected):
    raw = {"user_id": raw_id, "event": "click", "value": 0}
    if expected is None:
        with pytest.raises(ValueError):
            normalise_event(raw)
    else:
        result = normalise_event(raw)
        assert result["user_id"] == expected

# Parametrize over multiple columns
@pytest.mark.parametrize("value, should_raise", [
    (0,    False),
    (100,  False),
    (-1,   True),
    (-999, True),
])
def test_value_validation(value, should_raise):
    raw = {"user_id": "u1", "event": "click", "value": value}
    if should_raise:
        with pytest.raises(ValueError):
            normalise_event(raw)
    else:
        normalise_event(raw)   # should not raise

from unittest.mock import patch, MagicMock
import pytest
from acme_data_pipeline.ingestor import fetch_events, IngestorError

# Patch the httpx.get that fetch_events calls internally
# The path "acme_data_pipeline.ingestor.httpx" targets the module where it is USED
@patch("acme_data_pipeline.ingestor.httpx")
def test_fetch_events_happy_path(mock_httpx):
    # Configure the mock response
    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_response.json.return_value = [
        {"user_id": "u1", "event": "click", "value": 5}
    ]
    mock_httpx.get.return_value = mock_response

    events = fetch_events(api_url="https://api.example.com/events")

    assert len(events) == 1
    assert events[0]["user_id"] == "u1"
    mock_httpx.get.assert_called_once_with(
        "https://api.example.com/events",
        timeout=10.0,
        headers={"Authorization": "Bearer TOKEN"},
    )

@patch("acme_data_pipeline.ingestor.httpx")
def test_fetch_events_raises_on_500(mock_httpx):
    mock_response = MagicMock()
    mock_response.status_code = 500
    mock_response.raise_for_status.side_effect = Exception("500 Server Error")
    mock_httpx.get.return_value = mock_response

    with pytest.raises(IngestorError, match="upstream API failed"):
        fetch_events(api_url="https://api.example.com/events")