Sometimes I found myself in a situation where it’s hard to understand what some components suppose to achieve. In these situations, I go to test where scenarios/contracts of this component should be.
But then I find some integrations tests with some entries in DB and expected mechanical results.
My strategy in this situation is to refactor test from integration test to BDD -like tests. Try to be as close to Gherkin scenarios using commands and events to observe this component (avoiding direct DB connection).
Here you have an example of such a situation. Before and after of such tests. After I tried to describe what I found out during refactoring.
Before / Original tests
class TestUpdateStock:
@fixture(autouse=True)
def setUp(self, container):
self.repository = container.get(MysqlProductRepository)
def test_apply_por_order(self, handler, por_command, dto, in_sync):
self.repository.save(dto)
event = handler.handle(por_command)
assert event.sku == dto.sku
assert event.affected == [
StockIdentifier(site=stock.site, platform=stock.platform)
for stock in dto.stocks
]
def test_apply_por_order_before_update(
self, handler, por_command_before_update, dto, in_sync):
self.repository.save(dto)
event = handler.handle(por_command_before_update)
assert event is None
def test_apply_in_sync_order(self, handler, in_sync_command, dto, in_sync):
self.repository.save(dto)
event = handler.handle(in_sync_command)
assert event.sku == dto.sku
assert event.affected == [
StockIdentifier(site=stock.site, platform=stock.platform)
for stock in dto.stocks
]
@fixture
def in_sync(self):
return StockDTO(
platform=Platform.Other,
site='DE',
quantity=Quantity(5),
)
@fixture
def dto(self, in_sync):
stock = StockDTO(
platform=Platform.Some,
site='PL',
quantity=Quantity(5),
)
dto = ProductDTOFactory(
stocks=[stock, in_sync], last_update=datetime.utcnow())
dto.point_of_reference = stock
return dto
@fixture
def in_sync_command(self, dto):
return ApplyOrder(
sku=dto.sku,
platform=dto.point_of_reference.platform,
site=dto.point_of_reference.site,
quantity_sold=Quantity(10),
when_sold=datetime.utcnow() + timedelta(minutes=10),
)
@fixture
def por_command(self, dto):
return ApplyOrder(
sku=dto.sku,
platform=dto.in_sync[0].platform,
site=dto.in_sync[0].site,
quantity_sold=Quantity(1),
when_sold=datetime.utcnow(),
)
@fixture
def por_command_before_update(self, dto):
return ApplyOrder(
sku=dto.sku,
platform=dto.point_of_reference.platform,
site=dto.point_of_reference.site,
quantity_sold=Quantity(1),
when_sold=datetime.utcnow() - timedelta(days=1),
)
After / BDD-like tests
class TestUpdateStock:
@fixture(autouse=True)
def setUp(self, handler):
self.stocks = handler
with given('Registered product with point of reference stock'):
self.point_of_reference = StockIdentifier()
self.sku = handler.handle(
RegisterProduct(stock=self.point_of_reference)
).sku
with given('Registered product stock in sync'):
self.in_sync = StockIdentifier()
self.stocks.handle(
RegisterStock(sku=self.sku, stock=self.in_sync)
)
def test_affects_only_in_sync_stocks_when_order_on_por(self):
old_quantity = self.quantity()
with given('Order on point of reference'):
order = self.order(self.point_of_reference)
result = self.stocks.handle(order)
assert result.sku == self.sku
with expect('Affects only stocks in sync'):
affected, = result.affected
assert affected.identifier == self.in_sync
assert affected.change == QuantityChange(
previous=old_quantity,
current=old_quantity - order.quantity_sold,
)
def test_ignore_order_made_before_seller_changed_por_stock(self):
with given('Not processed order'):
order = self.order(self.point_of_reference)
with when('Seller updates stocks'):
seller_updated_stock = UpdateStock(
sku=self.sku,
stock=self.point_of_reference,
when_updated=order.when_sold + timedelta(days=1),
)
assert self.stocks.handle(seller_updated_stock)
with then('order made before update is ignored'):
assert not self.stocks.handle(order)
def test_affects_all_stocks_when_order_made_on_in_sync_stock(self):
old_quantity = self.quantity()
with given('Order on stock in sync'):
order = self.order(self.in_sync)
result = self.stocks.handle(order)
assert result.sku == self.sku
with expect('Affects all stocks'):
affected = [a.identifier for a in result.affected]
assert self.point_of_reference in affected
assert self.in_sync in affected
assert all(
affected.change == QuantityChange(
previous=old_quantity,
current=old_quantity - order.quantity_sold,
)
for affected in result.affected
)
def test_apply_order_only_once_when_send_multiple_times(self):
with given('Processed point of reference order'):
por_order = self.order(self.point_of_reference)
assert self.stocks.handle(por_order)
with expect('Cannot be applied anymore'):
assert not self.stocks.handle(por_order)
assert not self.stocks.handle(por_order)
assert not self.stocks.handle(por_order)
with given('Processed stock in sync order'):
in_sync_order = self.order(self.in_sync)
assert self.stocks.handle(in_sync_order)
with expect('Cannot be applied anymore'):
assert not self.stocks.handle(in_sync_order)
assert not self.stocks.handle(in_sync_order)
assert not self.stocks.handle(in_sync_order)
def order(self, stock: StockIdentifier) -> ApplyOrder:
return ApplyOrder(sku=self.sku, stock=stock)
def quantity(self) -> Quantity:
return self.stocks.get_quantity(sku=self.sku, stock=self.por)