Application 2 - Cargo shipping¶
This example follows the original Cargo Shipping example that figures in the DDD book, as worked up into a running application by the DDD Sample project:
“One of the most requested aids to coming up to speed on DDD has been a running example application. Starting from a simple set of functions and a model based on the cargo example used in Eric Evans’ book, we have built a running application with which to demonstrate a practical implementation of the building block patterns as well as illustrate the impact of aggregates and bounded contexts.”
The original example was not event-sourced and was coded in Java. The example below is an event-sourced version of the original coded in Python.
Application¶
The application object BookingService
allows new cargo to be booked, cargo details
to be presented, the destination of cargo to be changed, choices of possible routes
for cargo to be presented, a route to be assigned, and for cargo handling events
to be registered.
The Booking
application defines and registers custom transcodings for the
custom value objects that are defined and used in the domain model.
from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional, cast
from uuid import UUID
from eventsourcing.application import Application
from eventsourcing.examples.cargoshipping.domainmodel import (
REGISTERED_ROUTES,
Cargo,
HandlingActivity,
Itinerary,
Leg,
Location,
)
from eventsourcing.persistence import Transcoder, Transcoding
class LocationAsName(Transcoding):
type = Location
name = "location"
def encode(self, obj: Location) -> str:
return obj.name
def decode(self, data: str) -> Location:
assert isinstance(data, str)
return Location[data]
class HandlingActivityAsName(Transcoding):
type = HandlingActivity
name = "handling_activity"
def encode(self, obj: HandlingActivity) -> str:
return obj.name
def decode(self, data: str) -> HandlingActivity:
assert isinstance(data, str)
return HandlingActivity[data]
class ItineraryAsDict(Transcoding):
type = Itinerary
name = "itinerary"
def encode(self, obj: Itinerary) -> Dict[str, Any]:
return obj.__dict__
def decode(self, data: Dict[str, Any]) -> Itinerary:
assert isinstance(data, dict)
return Itinerary(**data)
class LegAsDict(Transcoding):
type = Leg
name = "leg"
def encode(self, obj: Leg) -> Dict[str, Any]:
return obj.__dict__
def decode(self, data: Dict[str, Any]) -> Leg:
assert isinstance(data, dict)
return Leg(**data)
class BookingApplication(Application):
def register_transcodings(self, transcoder: Transcoder) -> None:
super(BookingApplication, self).register_transcodings(transcoder)
transcoder.register(LocationAsName())
transcoder.register(HandlingActivityAsName())
transcoder.register(ItineraryAsDict())
transcoder.register(LegAsDict())
def book_new_cargo(
self,
origin: Location,
destination: Location,
arrival_deadline: datetime,
) -> UUID:
cargo = Cargo.new_booking(origin, destination, arrival_deadline)
self.save(cargo)
return cargo.id
def change_destination(self, tracking_id: UUID, destination: Location) -> None:
cargo = self.get_cargo(tracking_id)
cargo.change_destination(destination)
self.save(cargo)
def request_possible_routes_for_cargo(self, tracking_id: UUID) -> List[Itinerary]:
cargo = self.get_cargo(tracking_id)
from_location = (cargo.last_known_location or cargo.origin).value
to_location = cargo.destination.value
try:
possible_routes = REGISTERED_ROUTES[(from_location, to_location)]
except KeyError:
raise Exception(
"Can't find routes from {} to {}".format(from_location, to_location)
)
return possible_routes
def assign_route(self, tracking_id: UUID, itinerary: Itinerary) -> None:
cargo = self.get_cargo(tracking_id)
cargo.assign_route(itinerary)
self.save(cargo)
def register_handling_event(
self,
tracking_id: UUID,
voyage_number: Optional[str],
location: Location,
handing_activity: HandlingActivity,
) -> None:
cargo = self.get_cargo(tracking_id)
cargo.register_handling_event(
tracking_id,
voyage_number,
location,
handing_activity,
)
self.save(cargo)
def get_cargo(self, tracking_id: UUID) -> Cargo:
return cast(Cargo, self.repository.get(tracking_id))
Domain model¶
The aggregate Cargo
allows new cargo bookings to be made, the destination
of the cargo to be changed, a route to be assigned, and for handling events
to be registered. It is defined in the more verbose style, using explicit
definitions of aggregate events, an explicit “create” method (new_booking()
),
and command methods that explicitly trigger events by calling trigger_event()
.
An aggregate projector function is implemented on the aggregate object using
@simpledispatchmethod
, with an event-specific method registered to handle
each type of aggregate event.
Custom value objects such as Location
and Itinerary
, are defined as
part of the domain model, and used in the Cargo
aggregate events and methods.
For the purpose of simplicity in this example, a fixed collection of routes between
locations are also defined, but in practice these would be editable and could be
also modelled as event-sourced aggregates.
from __future__ import annotations
from datetime import datetime, timedelta
from enum import Enum
from typing import Dict, List, Optional, Tuple, Union, cast
from uuid import UUID, uuid4
from eventsourcing.dispatch import singledispatchmethod
from eventsourcing.domain import Aggregate
class Location(Enum):
"""
Locations in the world.
"""
HAMBURG = "HAMBURG"
HONGKONG = "HONGKONG"
NEWYORK = "NEWYORK"
STOCKHOLM = "STOCKHOLM"
TOKYO = "TOKYO"
NLRTM = "NLRTM"
USDAL = "USDAL"
AUMEL = "AUMEL"
class Leg(object):
"""
Leg of an itinerary.
"""
def __init__(
self,
origin: str,
destination: str,
voyage_number: str,
):
self.origin: str = origin
self.destination: str = destination
self.voyage_number: str = voyage_number
class Itinerary(object):
"""
An itinerary along which cargo is shipped.
"""
def __init__(
self,
origin: str,
destination: str,
legs: Tuple[Leg, ...],
):
self.origin = origin
self.destination = destination
self.legs = legs
class HandlingActivity(Enum):
RECEIVE = "RECEIVE"
LOAD = "LOAD"
UNLOAD = "UNLOAD"
CLAIM = "CLAIM"
# Custom static types.
LegDetails = Dict[str, str]
ItineraryDetails = Dict[str, Union[str, List[LegDetails]]]
NextExpectedActivity = Optional[Tuple[HandlingActivity, Location, str]]
# Some routes from one location to another.
REGISTERED_ROUTES = {
("HONGKONG", "STOCKHOLM"): [
Itinerary(
origin="HONGKONG",
destination="STOCKHOLM",
legs=(
Leg(
origin="HONGKONG",
destination="NEWYORK",
voyage_number="V1",
),
Leg(
origin="NEWYORK",
destination="STOCKHOLM",
voyage_number="V2",
),
),
)
],
("TOKYO", "STOCKHOLM"): [
Itinerary(
origin="TOKYO",
destination="STOCKHOLM",
legs=(
Leg(
origin="TOKYO",
destination="HAMBURG",
voyage_number="V3",
),
Leg(
origin="HAMBURG",
destination="STOCKHOLM",
voyage_number="V4",
),
),
)
],
}
class Cargo(Aggregate):
"""
The Cargo aggregate is an event-sourced domain model aggregate that
specifies the routing from origin to destination, and can track what
happens to the cargo after it has been booked.
"""
def __init__(
self,
origin: Location,
destination: Location,
arrival_deadline: datetime,
):
self._origin: Location = origin
self._destination: Location = destination
self._arrival_deadline: datetime = arrival_deadline
self._transport_status: str = "NOT_RECEIVED"
self._routing_status: str = "NOT_ROUTED"
self._is_misdirected: bool = False
self._estimated_time_of_arrival: Optional[datetime] = None
self._next_expected_activity: NextExpectedActivity = None
self._route: Optional[Itinerary] = None
self._last_known_location: Optional[Location] = None
self._current_voyage_number: Optional[str] = None
@property
def origin(self) -> Location:
return self._origin
@property
def destination(self) -> Location:
return self._destination
@property
def arrival_deadline(self) -> datetime:
return self._arrival_deadline
@property
def transport_status(self) -> str:
return self._transport_status
@property
def routing_status(self) -> str:
return self._routing_status
@property
def is_misdirected(self) -> bool:
return self._is_misdirected
@property
def estimated_time_of_arrival(
self,
) -> Optional[datetime]:
return self._estimated_time_of_arrival
@property
def next_expected_activity(self) -> NextExpectedActivity:
return self._next_expected_activity
@property
def route(self) -> Optional[Itinerary]:
return self._route
@property
def last_known_location(self) -> Optional[Location]:
return self._last_known_location
@property
def current_voyage_number(self) -> Optional[str]:
return self._current_voyage_number
@classmethod
def new_booking(
cls,
origin: Location,
destination: Location,
arrival_deadline: datetime,
) -> "Cargo":
return cls._create(
event_class=cls.BookingStarted,
id=uuid4(),
origin=origin,
destination=destination,
arrival_deadline=arrival_deadline,
)
class BookingStarted(Aggregate.Created):
origin: Location
destination: Location
arrival_deadline: datetime
class Event(Aggregate.Event):
def apply(self, aggregate: Aggregate) -> None:
cast(Cargo, aggregate).when(self)
@singledispatchmethod
def when(self, event: Event) -> None:
"""
Default method to apply an aggregate event to the aggregate object.
"""
def change_destination(self, destination: Location) -> None:
self.trigger_event(
self.DestinationChanged,
destination=destination,
)
class DestinationChanged(Event):
destination: Location
@when.register
def _(self, event: Cargo.DestinationChanged) -> None:
self._destination = event.destination
def assign_route(self, itinerary: Itinerary) -> None:
self.trigger_event(self.RouteAssigned, route=itinerary)
class RouteAssigned(Event):
route: Itinerary
@when.register
def _(self, event: Cargo.RouteAssigned) -> None:
self._route = event.route
self._routing_status = "ROUTED"
self._estimated_time_of_arrival = Cargo.Event.create_timestamp() + timedelta(
weeks=1
)
self._next_expected_activity = (HandlingActivity.RECEIVE, self.origin, "")
self._is_misdirected = False
def register_handling_event(
self,
tracking_id: UUID,
voyage_number: Optional[str],
location: Location,
handling_activity: HandlingActivity,
) -> None:
self.trigger_event(
self.HandlingEventRegistered,
tracking_id=tracking_id,
voyage_number=voyage_number,
location=location,
handling_activity=handling_activity,
)
class HandlingEventRegistered(Event):
tracking_id: UUID
voyage_number: str
location: Location
handling_activity: str
@when.register
def _(self, event: Cargo.HandlingEventRegistered) -> None:
assert self.route is not None
if event.handling_activity == HandlingActivity.RECEIVE:
self._transport_status = "IN_PORT"
self._last_known_location = event.location
self._next_expected_activity = (
HandlingActivity.LOAD,
event.location,
self.route.legs[0].voyage_number,
)
elif event.handling_activity == HandlingActivity.LOAD:
self._transport_status = "ONBOARD_CARRIER"
self._current_voyage_number = event.voyage_number
for leg in self.route.legs:
if leg.origin == event.location.value:
if leg.voyage_number == event.voyage_number:
self._next_expected_activity = (
HandlingActivity.UNLOAD,
Location[leg.destination],
event.voyage_number,
)
break
else:
raise Exception(
"Can't find leg with origin={} and "
"voyage_number={}".format(
event.location,
event.voyage_number,
)
)
elif event.handling_activity == HandlingActivity.UNLOAD:
self._current_voyage_number = None
self._last_known_location = event.location
self._transport_status = "IN_PORT"
if event.location == self.destination:
self._next_expected_activity = (
HandlingActivity.CLAIM,
event.location,
"",
)
elif event.location.value in [leg.destination for leg in self.route.legs]:
for i, leg in enumerate(self.route.legs):
if leg.voyage_number == event.voyage_number:
next_leg: Leg = self.route.legs[i + 1]
assert Location[next_leg.origin] == event.location
self._next_expected_activity = (
HandlingActivity.LOAD,
event.location,
next_leg.voyage_number,
)
break
else:
self._is_misdirected = True
self._next_expected_activity = None
elif event.handling_activity == HandlingActivity.CLAIM:
self._next_expected_activity = None
self._transport_status = "CLAIMED"
else:
raise Exception(
"Unsupported handling event: {}".format(event.handling_activity)
)
Interface¶
The interface object BookingService
repeats the application methods, allowing
new cargo to be booked, cargo details to be presented, the destination of cargo to
be changed, choices of possible routes for cargo to be presented, a route to be
assigned, and for cargo handling events to be registered.
It allows clients (e.g. a test case, or Web interface) to deal with simple object
types that can be easily serialised and deserialised. It interacts with the
application using the custom value objects defined in the domain model. For
the purposes of testing, we need to simulate the user selecting a preferred
itinerary from a list, which we do by picking the first in the list of presented
options using the select_preferred_itinerary()
function.
from __future__ import annotations
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Union
from uuid import UUID
from eventsourcing.examples.cargoshipping.application import BookingApplication
from eventsourcing.examples.cargoshipping.domainmodel import (
HandlingActivity,
Itinerary,
ItineraryDetails,
LegDetails,
Location,
)
NextExpectedActivityDetails = Optional[Tuple[str, ...]]
CargoDetails = Dict[
str, Optional[Union[str, bool, datetime, NextExpectedActivityDetails]]
]
class BookingService(object):
"""
Presents an application interface that uses
simple types of object (str, bool, datetime).
"""
def __init__(self, app: BookingApplication):
self.app = app
def book_new_cargo(
self,
origin: str,
destination: str,
arrival_deadline: datetime,
) -> str:
tracking_id = self.app.book_new_cargo(
Location[origin],
Location[destination],
arrival_deadline,
)
return str(tracking_id)
def get_cargo_details(self, tracking_id: str) -> CargoDetails:
cargo = self.app.get_cargo(UUID(tracking_id))
# Present 'next_expected_activity'.
next_expected_activity: NextExpectedActivityDetails
if cargo.next_expected_activity is None:
next_expected_activity = None
elif len(cargo.next_expected_activity) == 2:
next_expected_activity = (
cargo.next_expected_activity[0].value,
cargo.next_expected_activity[1].value,
)
elif len(cargo.next_expected_activity) == 3:
next_expected_activity = (
cargo.next_expected_activity[0].value,
cargo.next_expected_activity[1].value,
cargo.next_expected_activity[2],
)
else:
raise Exception(
"Invalid next expected activity: {}".format(
cargo.next_expected_activity
)
)
# Present 'last_known_location'.
if cargo.last_known_location is None:
last_known_location = None
else:
last_known_location = cargo.last_known_location.value
# Present the cargo details.
return {
"id": str(cargo.id),
"origin": cargo.origin.value,
"destination": cargo.destination.value,
"arrival_deadline": cargo.arrival_deadline,
"transport_status": cargo.transport_status,
"routing_status": cargo.routing_status,
"is_misdirected": cargo.is_misdirected,
"estimated_time_of_arrival": cargo.estimated_time_of_arrival,
"next_expected_activity": next_expected_activity,
"last_known_location": last_known_location,
"current_voyage_number": cargo.current_voyage_number,
}
def change_destination(self, tracking_id: str, destination: str) -> None:
self.app.change_destination(UUID(tracking_id), Location[destination])
def request_possible_routes_for_cargo(
self, tracking_id: str
) -> List[ItineraryDetails]:
routes = self.app.request_possible_routes_for_cargo(UUID(tracking_id))
return [self.dict_from_itinerary(route) for route in routes]
def dict_from_itinerary(self, itinerary: Itinerary) -> ItineraryDetails:
legs_details = []
for leg in itinerary.legs:
leg_details: LegDetails = {
"origin": leg.origin,
"destination": leg.destination,
"voyage_number": leg.voyage_number,
}
legs_details.append(leg_details)
route_details: ItineraryDetails = {
"origin": itinerary.origin,
"destination": itinerary.destination,
"legs": legs_details,
}
return route_details
def assign_route(
self,
tracking_id: str,
route_details: ItineraryDetails,
) -> None:
routes = self.app.request_possible_routes_for_cargo(UUID(tracking_id))
for route in routes:
if route_details == self.dict_from_itinerary(route):
self.app.assign_route(UUID(tracking_id), route)
def register_handling_event(
self,
tracking_id: str,
voyage_number: Optional[str],
location: str,
handling_activity: str,
) -> None:
self.app.register_handling_event(
UUID(tracking_id),
voyage_number,
Location[location],
HandlingActivity[handling_activity],
)
# Stub function that picks an itinerary from a list of possible itineraries.
def select_preferred_itinerary(
itineraries: List[ItineraryDetails],
) -> ItineraryDetails:
return itineraries[0]
Test case¶
Following the sample project, the test case has two test methods. One test shows an administrator booking a new cargo, viewing the current state of the cargo, and changing the destination. The other test goes further by assigning a route to a cargo booking, tracking the cargo handling events as it is shipped around the world, recovering by assigning a new route after the cargo was unloaded in the wrong place, until finally the cargo is claimed at its correct destination.
from __future__ import annotations
import unittest
from datetime import timedelta
from eventsourcing.examples.cargoshipping.application import BookingApplication
from eventsourcing.examples.cargoshipping.domainmodel import Cargo
from eventsourcing.examples.cargoshipping.interface import (
BookingService,
select_preferred_itinerary,
)
class TestBookingService(unittest.TestCase):
def setUp(self) -> None:
self.service = BookingService(BookingApplication())
def test_admin_can_book_new_cargo(self) -> None:
arrival_deadline = Cargo.Event.create_timestamp() + timedelta(weeks=3)
cargo_id = self.service.book_new_cargo(
origin="NLRTM",
destination="USDAL",
arrival_deadline=arrival_deadline,
)
cargo_details = self.service.get_cargo_details(cargo_id)
self.assertTrue(cargo_details["id"])
self.assertEqual(cargo_details["origin"], "NLRTM")
self.assertEqual(cargo_details["destination"], "USDAL")
self.service.change_destination(cargo_id, destination="AUMEL")
cargo_details = self.service.get_cargo_details(cargo_id)
self.assertEqual(cargo_details["destination"], "AUMEL")
self.assertEqual(
cargo_details["arrival_deadline"],
arrival_deadline,
)
def test_scenario_cargo_from_hongkong_to_stockholm(
self,
) -> None:
# Test setup: A cargo should be shipped from
# Hongkong to Stockholm, and it should arrive
# in no more than two weeks.
origin = "HONGKONG"
destination = "STOCKHOLM"
arrival_deadline = Cargo.Event.create_timestamp() + timedelta(weeks=2)
# Use case 1: booking.
# A new cargo is booked, and the unique tracking
# id is assigned to the cargo.
tracking_id = self.service.book_new_cargo(origin, destination, arrival_deadline)
# The tracking id can be used to lookup the cargo
# in the repository.
# Important: The cargo, and thus the domain model,
# is responsible for determining the status of the
# cargo, whether it is on the right track or not
# and so on. This is core domain logic. Tracking
# the cargo basically amounts to presenting
# information extracted from the cargo aggregate
# in a suitable way.
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(
cargo_details["transport_status"],
"NOT_RECEIVED",
)
self.assertEqual(cargo_details["routing_status"], "NOT_ROUTED")
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertEqual(
cargo_details["estimated_time_of_arrival"],
None,
)
self.assertEqual(cargo_details["next_expected_activity"], None)
# Use case 2: routing.
#
# A number of possible routes for this cargo is
# requested and may be presented to the customer
# in some way for him/her to choose from.
# Selection could be affected by things like price
# and time of delivery, but this test simply uses
# an arbitrary selection to mimic that process.
itineraries = self.service.request_possible_routes_for_cargo(tracking_id)
route_details = select_preferred_itinerary(itineraries)
# The cargo is then assigned to the selected
# route, described by an itinerary.
self.service.assign_route(tracking_id, route_details)
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(
cargo_details["transport_status"],
"NOT_RECEIVED",
)
self.assertEqual(cargo_details["routing_status"], "ROUTED")
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertTrue(cargo_details["estimated_time_of_arrival"])
self.assertEqual(
cargo_details["next_expected_activity"],
("RECEIVE", "HONGKONG", ""),
)
# Use case 3: handling
# A handling event registration attempt will be
# formed from parsing the data coming in as a
# handling report either via the web service
# interface or as an uploaded CSV file. The
# handling event factory tries to create a
# HandlingEvent from the attempt, and if the
# factory decides that this is a plausible
# handling event, it is stored. If the attempt
# is invalid, for example if no cargo exists for
# the specified tracking id, the attempt is
# rejected.
#
# Handling begins: cargo is received in Hongkong.
self.service.register_handling_event(tracking_id, None, "HONGKONG", "RECEIVE")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["transport_status"], "IN_PORT")
self.assertEqual(
cargo_details["last_known_location"],
"HONGKONG",
)
self.assertEqual(
cargo_details["next_expected_activity"],
("LOAD", "HONGKONG", "V1"),
)
# Load onto voyage V1.
self.service.register_handling_event(tracking_id, "V1", "HONGKONG", "LOAD")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], "V1")
self.assertEqual(
cargo_details["last_known_location"],
"HONGKONG",
)
self.assertEqual(
cargo_details["transport_status"],
"ONBOARD_CARRIER",
)
self.assertEqual(
cargo_details["next_expected_activity"],
("UNLOAD", "NEWYORK", "V1"),
)
# Incorrectly unload in Tokyo.
self.service.register_handling_event(tracking_id, "V1", "TOKYO", "UNLOAD")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], None)
self.assertEqual(cargo_details["last_known_location"], "TOKYO")
self.assertEqual(cargo_details["transport_status"], "IN_PORT")
self.assertEqual(cargo_details["is_misdirected"], True)
self.assertEqual(cargo_details["next_expected_activity"], None)
# Reroute.
itineraries = self.service.request_possible_routes_for_cargo(tracking_id)
route_details = select_preferred_itinerary(itineraries)
self.service.assign_route(tracking_id, route_details)
# Load in Tokyo.
self.service.register_handling_event(tracking_id, "V3", "TOKYO", "LOAD")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], "V3")
self.assertEqual(cargo_details["last_known_location"], "TOKYO")
self.assertEqual(
cargo_details["transport_status"],
"ONBOARD_CARRIER",
)
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertEqual(
cargo_details["next_expected_activity"],
("UNLOAD", "HAMBURG", "V3"),
)
# Unload in Hamburg.
self.service.register_handling_event(tracking_id, "V3", "HAMBURG", "UNLOAD")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], None)
self.assertEqual(cargo_details["last_known_location"], "HAMBURG")
self.assertEqual(cargo_details["transport_status"], "IN_PORT")
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertEqual(
cargo_details["next_expected_activity"],
("LOAD", "HAMBURG", "V4"),
)
# Load in Hamburg
self.service.register_handling_event(tracking_id, "V4", "HAMBURG", "LOAD")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], "V4")
self.assertEqual(cargo_details["last_known_location"], "HAMBURG")
self.assertEqual(
cargo_details["transport_status"],
"ONBOARD_CARRIER",
)
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertEqual(
cargo_details["next_expected_activity"],
("UNLOAD", "STOCKHOLM", "V4"),
)
# Unload in Stockholm
self.service.register_handling_event(tracking_id, "V4", "STOCKHOLM", "UNLOAD")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], None)
self.assertEqual(
cargo_details["last_known_location"],
"STOCKHOLM",
)
self.assertEqual(cargo_details["transport_status"], "IN_PORT")
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertEqual(
cargo_details["next_expected_activity"],
("CLAIM", "STOCKHOLM", ""),
)
# Finally, cargo is claimed in Stockholm.
self.service.register_handling_event(tracking_id, None, "STOCKHOLM", "CLAIM")
cargo_details = self.service.get_cargo_details(tracking_id)
self.assertEqual(cargo_details["current_voyage_number"], None)
self.assertEqual(
cargo_details["last_known_location"],
"STOCKHOLM",
)
self.assertEqual(cargo_details["transport_status"], "CLAIMED")
self.assertEqual(cargo_details["is_misdirected"], False)
self.assertEqual(cargo_details["next_expected_activity"], None)