Основы NoSQL

Updated Jul 2026

Не все данные живут в реляционных базах данных. MongoDB хранит документы, Redis кеширует пары ключ-значение, а DynamoDB предоставляет бессерверное хранилище ключ-значение/документов. Как QA-инженер, вы должны знать, как тестировать эти системы — паттерны значительно отличаются от SQL.

Типы NoSQL баз данных

Тестирование MongoDB

MongoDB хранит данные в виде JSON-подобных документов (BSON). Документы в одной коллекции могут иметь разную структуру — что является одновременно преимуществом и вызовом для тестирования.

Валидация структуры документов

from pymongo import MongoClient
import pytest

@pytest.fixture(scope="session")
def mongo_db():
    client = MongoClient("mongodb://localhost:27017")
    db = client["testdb"]
    yield db
    client.close()

def test_user_document_structure(mongo_db):
    """Verify user documents have expected fields."""
    mongo_db.users.insert_one({
        "name": "Alice",
        "email": "alice@test.com",
        "tags": ["admin"],
        "profile": {"bio": "Test user", "avatar_url": None}
    })

    user = mongo_db.users.find_one({"email": "alice@test.com"})
    assert user is not None
    assert isinstance(user["tags"], list)
    assert "admin" in user["tags"]
    assert isinstance(user["profile"], dict)
    assert "bio" in user["profile"]

def test_schema_validation_enforced(mongo_db):
    """If schema validation is configured, invalid documents should be rejected."""
    # Set up schema validation
    mongo_db.command({
        "collMod": "strict_users",
        "validator": {
            "$jsonSchema": {
                "bsonType": "object",
                "required": ["name", "email"],
                "properties": {
                    "name": {"bsonType": "string"},
                    "email": {"bsonType": "string", "pattern": "^.+@.+$"}
                }
            }
        }
    })

    # Valid document should succeed
    mongo_db.strict_users.insert_one({"name": "Valid", "email": "valid@test.com"})

    # Invalid document should fail
    from pymongo.errors import WriteError
    with pytest.raises(WriteError):
        mongo_db.strict_users.insert_one({"name": "Invalid"})  # Missing email

Тестирование агрегации MongoDB

def test_aggregation_pipeline(mongo_db):
    """Test an aggregation pipeline that calculates user statistics."""
    # Insert test data
    mongo_db.orders.insert_many([
        {"user_id": "user-1", "total": 50.00, "status": "completed"},
        {"user_id": "user-1", "total": 75.00, "status": "completed"},
        {"user_id": "user-1", "total": 25.00, "status": "cancelled"},
        {"user_id": "user-2", "total": 100.00, "status": "completed"},
    ])

    # Run aggregation
    pipeline = [
        {"$match": {"status": "completed"}},
        {"$group": {
            "_id": "$user_id",
            "total_spent": {"$sum": "$total"},
            "order_count": {"$sum": 1}
        }},
        {"$sort": {"total_spent": -1}}
    ]
    results = list(mongo_db.orders.aggregate(pipeline))

    assert len(results) == 2
    assert results[0]["_id"] == "user-1"
    assert results[0]["total_spent"] == 125.00  # 50 + 75 (cancelled excluded)
    assert results[0]["order_count"] == 2

Тестирование Redis

Redis используется преимущественно как кеш, хранилище сессий или брокер сообщений. Тестирование Redis фокусируется на поведении TTL, инвалидации кеша и корректности структур данных.

Истечение TTL

import redis
import time
import pytest

@pytest.fixture
def redis_client():
    client = redis.Redis(host="localhost", port=6379, db=1)
    yield client
    client.flushdb()  # Clean up after tests

def test_cache_ttl_expiration(redis_client):
    """Cached data should expire after TTL."""
    redis_client.setex("session:abc", 2, "user-data")  # 2 second TTL

    # Immediately available
    assert redis_client.get("session:abc") is not None

    # Wait for expiration
    time.sleep(3)
    assert redis_client.get("session:abc") is None

def test_cache_ttl_value(redis_client):
    """Verify the TTL is set correctly."""
    redis_client.setex("session:def", 300, "data")
    ttl = redis_client.ttl("session:def")
    assert 298 <= ttl <= 300  # Allow small timing variance

Инвалидация кеша

def test_cache_invalidation_on_update(redis_client, api_client):
    """When a user is updated via API, their cache should be invalidated."""
    # Prime the cache
    api_client.get("/users/123")
    assert redis_client.get("user:123") is not None

    # Update the user
    api_client.put("/users/123", json={"name": "Updated Name"})

    # Cache should be invalidated
    assert redis_client.get("user:123") is None

    # Next GET should re-populate cache
    response = api_client.get("/users/123")
    assert response.json()["name"] == "Updated Name"
    assert redis_client.get("user:123") is not None

Структуры данных Redis

def test_redis_sorted_set_leaderboard(redis_client):
    """Test a leaderboard using Redis sorted set."""
    redis_client.zadd("leaderboard", {"alice": 100, "bob": 85, "charlie": 92})

    # Get top 2
    top = redis_client.zrevrange("leaderboard", 0, 1, withscores=True)
    assert top[0] == (b"alice", 100.0)
    assert top[1] == (b"charlie", 92.0)

    # Get rank
    rank = redis_client.zrevrank("leaderboard", "bob")
    assert rank == 2  # 0-indexed: alice=0, charlie=1, bob=2

def test_redis_list_as_queue(redis_client):
    """Test a job queue using Redis list."""
    redis_client.rpush("job_queue", "job-1", "job-2", "job-3")
    assert redis_client.llen("job_queue") == 3

    # Process (FIFO)
    job = redis_client.lpop("job_queue")
    assert job == b"job-1"
    assert redis_client.llen("job_queue") == 2

Тестирование DynamoDB

import boto3
from moto import mock_dynamodb  # Mock for testing

@mock_dynamodb
def test_dynamodb_crud():
    """Test CRUD operations on DynamoDB."""
    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")

    # Create table
    table = dynamodb.create_table(
        TableName="users",
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST"
    )

    # Insert
    table.put_item(Item={"id": "user-1", "name": "Alice", "email": "alice@test.com"})

    # Read
    response = table.get_item(Key={"id": "user-1"})
    assert response["Item"]["name"] == "Alice"

    # Query non-existent
    response = table.get_item(Key={"id": "nonexistent"})
    assert "Item" not in response

Практическое упражнение

1. Напишите MongoDB-тесты: вставьте документ, запросите его, проверьте структуру, протестируйте валидацию схемы
2. Напишите Redis-тесты: установите ключ с TTL, проверьте его существование, дождитесь истечения, проверьте отсутствие
3. Напишите тест для ограничения частоты на Redis: инкрементируйте счётчик, проверьте сброс после окна
4. Протестируйте инвалидацию кеша: обновите данные через API, проверьте, что кеш был очищен
5. Сравните паттерны тестирования между SQL-ограничениями и валидацией схемы MongoDB

Ключевые выводы

• Тестирование NoSQL фокусируется на других аспектах, чем SQL: гибкость схемы, TTL, поведение кеша
• MongoDB: тестируйте структуру документов, конвейеры агрегации и правила валидации схемы
• Redis: тестируйте истечение TTL, инвалидацию кеша и операции со структурами данных
• DynamoDB: тестируйте паттерны запросов, поведение ёмкости и итоговую согласованность
• Используйте библиотеки мокирования (moto для AWS, mongomock для MongoDB) для юнит-тестов