forked from cosmicpython/code
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_views.py
More file actions
35 lines (28 loc) · 1.52 KB
/
test_views.py
File metadata and controls
35 lines (28 loc) · 1.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from datetime import date
from allocation import views
from allocation.domain import commands
from allocation.service_layer import messagebus, unit_of_work
today = date.today()
def test_allocations_view(sqlite_session_factory):
uow = unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory)
messagebus.handle(commands.CreateBatch("sku1batch", "sku1", 50, None), uow)
messagebus.handle(commands.CreateBatch("sku2batch", "sku2", 50, today), uow)
messagebus.handle(commands.Allocate("order1", "sku1", 20), uow)
messagebus.handle(commands.Allocate("order1", "sku2", 20), uow)
# add a spurious batch and order to make sure we're getting the right ones
messagebus.handle(commands.CreateBatch("sku1batch-later", "sku1", 50, today), uow)
messagebus.handle(commands.Allocate("otherorder", "sku1", 30), uow)
messagebus.handle(commands.Allocate("otherorder", "sku2", 10), uow)
assert views.allocations("order1", uow) == [
{"sku": "sku1", "batchref": "sku1batch"},
{"sku": "sku2", "batchref": "sku2batch"},
]
def test_deallocation(sqlite_session_factory):
uow = unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory)
messagebus.handle(commands.CreateBatch("b1", "sku1", 50, None), uow)
messagebus.handle(commands.CreateBatch("b2", "sku1", 50, today), uow)
messagebus.handle(commands.Allocate("o1", "sku1", 40), uow)
messagebus.handle(commands.ChangeBatchQuantity("b1", 10), uow)
assert views.allocations("o1", uow) == [
{"sku": "sku1", "batchref": "b2"},
]