Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# a couple of event instances that we can use for testing
test_events = [
Event.from_parameters(
{
"event": UserUttered.type_name,
"text": "/goodbye",
"parse_data": {
"intent": {"confidence": 1.0, "name": "greet"},
"entities": [],
},
}
),
BotUttered("Welcome!", {"test": True}),
SlotSet("cuisine", 34),
SlotSet("cuisine", "34"),
SlotSet("location", None),
SlotSet("location", [34, "34", None]),
]
@pytest.fixture
def rasa_app_without_api(rasa_server_without_api: Sanic) -> SanicTestClient:
return get_test_client(rasa_server_without_api)
@pytest.fixture
def rasa_app(rasa_server: Sanic) -> SanicTestClient:
return get_test_client(rasa_server)
@pytest.fixture
def rasa_app_nlu(rasa_nlu_server: Sanic) -> SanicTestClient:
@pytest.mark.parametrize(
"one_event,another_event",
[
(
UserUttered("/greet", {"name": "greet", "confidence": 1.0}, []),
UserUttered("/goodbye", {"name": "goodbye", "confidence": 1.0}, []),
),
(SlotSet("my_slot", "value"), SlotSet("my__other_slot", "value")),
(Restarted(), None),
(AllSlotsReset(), None),
(ConversationPaused(), None),
(ConversationResumed(), None),
(StoryExported(), None),
(ActionReverted(), None),
(UserUtteranceReverted(), None),
(ActionExecuted("my_action"), ActionExecuted("my_other_action")),
(FollowupAction("my_action"), FollowupAction("my_other_action")),
(
BotUttered("my_text", {"my_data": 1}),
BotUttered("my_other_test", {"my_other_data": 1}),
),
(
AgentUttered("my_text", "my_data"),
AgentUttered("my_other_test", "my_other_data"),
def test_dynamo_tracker_floats():
conversation_id = uuid.uuid4().hex
tracker_store = DynamoTrackerStore(domain)
tracker = tracker_store.get_or_create_tracker(
conversation_id, append_action_listen=False
)
# save `slot` event with known `float`-type timestamp
timestamp = 13423.23434623
tracker.update(SlotSet("key", "val", timestamp=timestamp))
tracker_store.save(tracker)
# retrieve tracker and the event timestamp is retrieved as a `float`
tracker = tracker_store.get_or_create_tracker(conversation_id)
retrieved_timestamp = tracker.events[0].timestamp
assert isinstance(retrieved_timestamp, float)
assert retrieved_timestamp == timestamp
@pytest.mark.parametrize("one_event,another_event", [
(UserUttered("/greet", {"name": "greet", "confidence": 1.0}, []),
UserUttered("/goodbye", {"name": "goodbye", "confidence": 1.0}, [])),
(SlotSet("my_slot", "value"),
SlotSet("my__other_slot", "value")),
(Restarted(),
None),
(AllSlotsReset(),
None),
(ConversationPaused(),
None),
(ConversationResumed(),
None),
(StoryExported(),
None),
SlotSet("my_slot", "value"),
SlotSet("another-slot", "value2"),
ActionExecuted(action_name=ACTION_LISTEN_NAME),
],
),
(
SessionConfig(123, False),
[SessionStarted(), ActionExecuted(action_name=ACTION_LISTEN_NAME)],
),
],
)
async def test_action_session_start_with_slots(
default_channel: CollectingOutputChannel,
template_nlg: TemplatedNaturalLanguageGenerator,
template_sender_tracker: DialogueStateTracker,
default_domain: Domain,
session_config: SessionConfig,
SessionConfig(123, False),
[SessionStarted(), ActionExecuted(action_name=ACTION_LISTEN_NAME)],
),
],
)
async def test_action_session_start_with_slots(
default_channel: CollectingOutputChannel,
template_nlg: TemplatedNaturalLanguageGenerator,
template_sender_tracker: DialogueStateTracker,
default_domain: Domain,
session_config: SessionConfig,
expected_events: List[Event],
):
# set a few slots on tracker
slot_set_event_1 = SlotSet("my_slot", "value")
slot_set_event_2 = SlotSet("another-slot", "value2")
for event in [slot_set_event_1, slot_set_event_2]:
template_sender_tracker.update(event)
default_domain.session_config = session_config
events = await ActionSessionStart().run(
default_channel, template_nlg, template_sender_tracker, default_domain
)
assert events == expected_events
# make sure that the list of events has ascending timestamps
assert sorted(events, key=lambda x: x.timestamp) == events
assert template_sender_tracker.followup_action == ACTION_LISTEN_NAME
test_events = [
Event.from_parameters(
{
"event": UserUttered.type_name,
"text": "/goodbye",
"parse_data": {
"intent": {"confidence": 1.0, "name": "greet"},
"entities": [],
},
}
),
BotUttered("Welcome!", {"test": True}),
SlotSet("cuisine", 34),
SlotSet("cuisine", "34"),
SlotSet("location", None),
SlotSet("location", [34, "34", None]),
]
@pytest.fixture
def rasa_app_without_api(rasa_server_without_api: Sanic) -> SanicTestClient:
return get_test_client(rasa_server_without_api)
@pytest.fixture
def rasa_app(rasa_server: Sanic) -> SanicTestClient:
return get_test_client(rasa_server)
@pytest.fixture
def rasa_app_nlu(rasa_nlu_server: Sanic) -> SanicTestClient:
return get_test_client(rasa_nlu_server)
def _merge_slots(
self, entities: Optional[List[Dict[Text, Any]]] = None
) -> List[SlotSet]:
"""Take a list of entities and create tracker slot set events.
If an entity type matches a slots name, the entities value is set
as the slots value by creating a ``SlotSet`` event.
"""
entities = entities if entities else self.latest_message.entities
new_slots = [
SlotSet(e["entity"], e["value"])
for e in entities
if e["entity"] in self.slots.keys()
]
return new_slots
def _merge_slots(
self, entities: Optional[List[Dict[Text, Any]]] = None
) -> List[SlotSet]:
"""Take a list of entities and create tracker slot set events.
If an entity type matches a slots name, the entities value is set
as the slots value by creating a ``SlotSet`` event.
"""
entities = entities if entities else self.latest_message.entities
new_slots = [
SlotSet(e["entity"], e["value"])
for e in entities
if e["entity"] in self.slots.keys()
]
return new_slots