RuleBasedStateMachine, инварианты, тестирование сложных систем
«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()Проблемы:
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) >= 0Hypothesis автоматически генерирует последовательности: push(5) → push(3) → pop() → push(1) → ...
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):
"""Проверка инварианта после каждой операции"""
passfrom 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.balanceclass 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 — правило пропускается.
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 позволяет передавать значения между правилами:
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]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| Аспект | Property-Based (@given) | Stateful (RuleBasedStateMachine) |
|---|---|---|
| Состояние | Нет или минимальное | Центральная концепция |
| Последовательности | Один набор данных | Последовательность операций |
| Инварианты | Проверяются один раз | Проверяются после каждой операции |
| Use case | Чистые функции | Системы с состоянием, API, БД |
Когда 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.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.