Stateful Testing и машины состояний

«Stateful testing превращает хаотичное тестирование систем с состоянием в научный метод.»

#Что такое stateful testing?

Stateful testing — подход к тестированию систем с состоянием, где тесты представляют собой последовательности операций. Hypothesis автоматически генерирует различные последовательности и проверяет инварианты после каждой операции.

#Проблема традиционных тестов

# Традиционный подход — вы придумываете последовательность
def test_stack_manual():
    stack = Stack()
    stack.push(1)
    stack.push(2)
    assert stack.pop() == 2
    assert stack.pop() == 1
    assert stack.is_empty()

Проблемы:

Вы придумываете только несколько сценариев
Легко пропустить краевые случаи
Не тестируются сложные последовательности

#Stateful подход

from hypothesis.stateful import RuleBasedStateMachine, rule, invariant

class StackMachine(RuleBasedStateMachine):
    def __init__(self):
        super().__init__()
        self.stack = []
    
    @rule(item=st.integers())
    def push(self, item):
        self.stack.append(item)
    
    @rule()
    def pop(self):
        if self.stack:
            return self.stack.pop()
    
    @invariant()
    def stack_size_non_negative(self):
        assert len(self.stack) >= 0

Hypothesis автоматически генерирует последовательности: push(5) → push(3) → pop() → push(1) → ...

#RuleBasedStateMachine

#Базовая структура

from hypothesis.stateful import (
    RuleBasedStateMachine,
    rule,
    invariant,
    precondition,
    initialize
)
from hypothesis import strategies as st

class MySystem(RuleBasedStateMachine):
    """
    Машина состояний для тестирования системы.
    """
    
    @initialize()
    def setup(self):
        """Инициализация состояния перед тестом"""
        self.state = {}
    
    @rule(param=st.integers())
    def operation(self, param):
        """Операция, изменяющая состояние"""
        pass
    
    @invariant()
    def check_invariant(self):
        """Проверка инварианта после каждой операции"""
        pass

#Пример: тестирование банковского счёта

from decimal import Decimal

class BankAccount(RuleBasedStateMachine):
    def __init__(self):
        super().__init__()
        self.balance = Decimal('0')
        self.transaction_history = []
    
    @rule(amount=st.decimals(min_value=Decimal('0.01'), max_value=Decimal('10000')))
    def deposit(self, amount):
        """Внесение средств"""
        self.balance += amount
        self.transaction_history.append(('deposit', amount))
    
    @rule(amount=st.decimals(min_value=Decimal('0.01'), max_value=Decimal('10000')))
    def withdraw(self, amount):
        """Снятие средств"""
        if self.balance >= amount:
            self.balance -= amount
            self.transaction_history.append(('withdraw', amount))
    
    @invariant()
    def balance_never_negative(self):
        """Баланс никогда не отрицателен"""
        assert self.balance >= 0
    
    @invariant()
    def history_matches_balance(self):
        """История транзакций соответствует балансу"""
        calculated = sum(
            amount if op == 'deposit' else -amount
            for op, amount in self.transaction_history
        )
        assert calculated == self.balance

#Precondition: условия применения правил

class BoundedStack(RuleBasedStateMachine):
    def __init__(self):
        super().__init__()
        self.stack = []
        self.capacity = 10
    
    @rule(item=st.integers())
    @precondition(lambda self: len(self.stack) < self.capacity)
    def push(self, item):
        self.stack.append(item)
    
    @rule()
    @precondition(lambda self: len(self.stack) > 0)
    def pop(self):
        return self.stack.pop()
    
    @invariant()
    def size_within_bounds(self):
        assert 0 <= len(self.stack) <= self.capacity

Важно: precondition проверяется перед применением правила. Если False — правило пропускается.

#Initialize: настройка начального состояния

class ConfigurableSystem(RuleBasedStateMachine):
    @initialize(config=st.builds(dict, 
        max_connections=st.integers(min_value=1, max_value=100),
        timeout=st.floats(min_value=0.1, max_value=60)
    ))
    def setup(self, config):
        self.config = config
        self.connections = 0
        self.requests = []
    
    @rule()
    @precondition(lambda self: self.connections < self.config['max_connections'])
    def add_connection(self):
        self.connections += 1
    
    @invariant()
    def connections_within_limit(self):
        assert self.connections <= self.config['max_connections']

#Bundle: обмен данными между правилами

Bundle позволяет передавать значения между правилами:

from hypothesis.stateful import Bundle

class DatabaseMachine(RuleBasedStateMachine):
    def __init__(self):
        super().__init__()
        self.tables = {}
        self.records = {}
    
    # Bundle для хранения ID записей
    record_ids = Bundle('record_ids')
    
    @rule(table_name=st.text(min_size=1, max_size=10))
    def create_table(self, table_name):
        if table_name not in self.tables:
            self.tables[table_name] = []
            self.records[table_name] = {}
    
    @rule(
        table_name=st.sampled_from(list(self.tables.keys())) if self.tables else st.just('default'),
        data=st.dictionaries(st.text(), st.integers())
    )
    @precondition(lambda self: len(self.tables) > 0)
    @initialize(target=record_ids)
    def insert_record(self, table_name, data):
        record_id = len(self.records[table_name])
        self.records[table_name][record_id] = data
        return (table_name, record_id)
    
    @rule(record_id=record_ids)
    def get_record(self, record_id):
        table_name, id_ = record_id
        assert id_ in self.records[table_name]

#Production-пример: тестирование очереди задач

from typing import Optional, List
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import heapq

class TaskStatus(Enum):
    PENDING = 'pending'
    RUNNING = 'running'
    COMPLETED = 'completed'
    FAILED = 'failed'

@dataclass(order=True)
class Task:
    priority: int
    id: int = field(compare=False)
    status: TaskStatus = field(default=TaskStatus.PENDING, compare=False)
    result: Optional[str] = field(default=None, compare=False)

class TaskQueueMachine(RuleBasedStateMachine):
    def __init__(self):
        super().__init__()
        self.queue: List[Task] = []
        self.completed: List[Task] = []
        self.failed: List[Task] = []
        self.next_id = 0
        self.running_task: Optional[Task] = None
    
    @rule(priority=st.integers(min_value=1, max_value=10))
    def add_task(self, priority):
        task = Task(priority=priority, id=self.next_id)
        self.next_id += 1
        heapq.heappush(self.queue, task)
    
    @rule()
    @precondition(lambda self: len(self.queue) > 0 and self.running_task is None)
    def start_task(self):
        task = heapq.heappop(self.queue)
        task.status = TaskStatus.RUNNING
        self.running_task = task
    
    @rule(result=st.text())
    @precondition(lambda self: self.running_task is not None)
    def complete_task(self, result):
        self.running_task.status = TaskStatus.COMPLETED
        self.running_task.result = result
        self.completed.append(self.running_task)
        self.running_task = None
    
    @rule(error=st.text())
    @precondition(lambda self: self.running_task is not None)
    def fail_task(self, error):
        self.running_task.status = TaskStatus.FAILED
        self.running_task.result = error
        self.failed.append(self.running_task)
        self.running_task = None
    
    @invariant()
    def no_running_when_empty(self):
        """Если очередь пуста и нет running задачи, то всё ок"""
        if len(self.queue) == 0 and self.running_task is None:
            assert all(t.status in [TaskStatus.COMPLETED, TaskStatus.FAILED] 
                      for t in self.completed + self.failed)
    
    @invariant()
    def max_one_running(self):
        """Максимум одна задача в работе"""
        running_count = 1 if self.running_task else 0
        assert running_count <= 1
    
    @invariant()
    def completed_have_results(self):
        """Завершённые задачи имеют результат"""
        for task in self.completed:
            assert task.status == TaskStatus.COMPLETED
            assert task.result is not None

#Stateful vs Property-Based

Аспект	Property-Based (@given)	Stateful (RuleBasedStateMachine)
Состояние	Нет или минимальное	Центральная концепция
Последовательности	Один набор данных	Последовательность операций
Инварианты	Проверяются один раз	Проверяются после каждой операции
Use case	Чистые функции	Системы с состоянием, API, БД

#Отладка failing sequences

Когда stateful тест падает, Hypothesis показывает последовательность:

Falsifying example:
state = TaskQueueMachine()
state.add_task(priority=5)
state.add_task(priority=3)
state.start_task()
state.fail_task(error='timeout')
state.start_task()
# AssertionError: ...

Это позволяет воспроизвести ошибку шаг за шагом.

#Заключение

Stateful testing — мощнейший инструмент для тестирования систем с состоянием. Он автоматически находит сложные баги в последовательностях операций, которые невозможно предвидеть вручную.

Следующая тема: Тестирование API — интеграция с FastAPI и Django, REST и GraphQL endpoints.

Аспект

Property-Based (@given)

Stateful (RuleBasedStateMachine)

Состояние

Нет или минимальное

Центральная концепция

Последовательности

Один набор данных

Последовательность операций

Инварианты

Проверяются один раз

Проверяются после каждой операции

Use case

Чистые функции

Системы с состоянием, API, БД