Added test for handling PromptFailed error.
@@ -5,8 +5,8 @@ from eventsourcing.application.process import Process, RepositoryWrapper
from eventsourcing.application.command import CommandProcess
from eventsourcing.domain.model.aggregate import AggregateRoot
from eventsourcing.domain.model.command import Command
-from eventsourcing.domain.model.events import assert_event_handlers_empty
-from eventsourcing.exceptions import CausalDependencyFailed
+from eventsourcing.domain.model.events import assert_event_handlers_empty, subscribe, unsubscribe
+from eventsourcing.exceptions import CausalDependencyFailed, PromptFailed
from eventsourcing.utils.topic import resolve_topic
from eventsourcing.utils.transcoding import json_loads
@@ -132,6 +132,39 @@ class TestProcess(TestCase):
downstream1.close()
downstream2.close()
+ def test_handle_prompt_failed(self):
+ process = Process(
+ 'test',
+ policy=example_policy,
+ persist_event_type=ExampleAggregate.Event,
+ setup_tables=True,
+ )
+
+ def raise_exception(_):
+ raise Exception()
+
+ def raise_prompt_failed(_):
+ raise PromptFailed()
+
+ subscribe(raise_exception)
+ try:
+ with self.assertRaises(PromptFailed):
+ process.publish_prompt()
+ finally:
+ unsubscribe(raise_exception)
+
+ subscribe(raise_prompt_failed)
+ try:
+ with self.assertRaises(PromptFailed):
+ process.publish_prompt()
+ finally:
+ unsubscribe(raise_prompt_failed)
+
+ try:
+ process.publish_prompt()
+ finally:
+ process.close()
+
def tearDown(self):
assert_event_handlers_empty()
