Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
SessionStarted(),
ActionExecuted(ACTION_LISTEN_NAME),
UserUttered(
f"/greet{json.dumps(slot_1)}",
{"name": "greet", "confidence": 1.0},
[{"entity": entity, "start": 6, "end": 22, "value": "Core"}],
),
SlotSet(entity, slot_1[entity]),
ActionExecuted("utter_greet"),
BotUttered("hey there Core!"),
ActionExecuted(ACTION_LISTEN_NAME),
ActionExecuted(ACTION_SESSION_START_NAME),
SessionStarted(),
# the initial SlotSet is reapplied after the SessionStarted sequence
SlotSet(entity, slot_1[entity]),
ActionExecuted(ACTION_LISTEN_NAME),
UserUttered(
f"/greet{json.dumps(slot_2)}",
{"name": "greet", "confidence": 1.0},
[
{
"entity": entity,
"start": 6,
"end": 42,
"value": "post-session start hello",
}
],
),
SlotSet(entity, slot_2[entity]),
ActionExecuted(ACTION_LISTEN_NAME),
]
def test_common_action_prefix():
this = [
ActionExecuted("action_listen"),
ActionExecuted("greet"),
UserUttered("hey"),
ActionExecuted("amazing"),
# until this point they are the same
SlotSet("my_slot", "a"),
ActionExecuted("a"),
ActionExecuted("after_a"),
]
other = [
ActionExecuted("action_listen"),
ActionExecuted("greet"),
UserUttered("hey"),
ActionExecuted("amazing"),
# until this point they are the same
SlotSet("my_slot", "b"),
ActionExecuted("b"),
ActionExecuted("after_b"),
]
num_common = visualization._length_of_common_action_prefix(this, other)
assert num_common == 3
async def test_action_session_start_without_slots(
default_channel: CollectingOutputChannel,
template_nlg: TemplatedNaturalLanguageGenerator,
template_sender_tracker: DialogueStateTracker,
default_domain: Domain,
):
events = await ActionSessionStart().run(
default_channel, template_nlg, template_sender_tracker, default_domain
)
assert events == [SessionStarted(), ActionExecuted(ACTION_LISTEN_NAME)]
async def test_successful_rephrasing(
self, default_channel, default_nlg, default_domain
):
events = [
ActionExecuted(ACTION_LISTEN_NAME),
user_uttered("greet", 0.2),
ActionExecuted(ACTION_DEFAULT_ASK_AFFIRMATION_NAME),
ActionExecuted(ACTION_LISTEN_NAME),
user_uttered("deny", 1),
ActionExecuted(ACTION_DEFAULT_ASK_REPHRASE_NAME),
ActionExecuted(ACTION_LISTEN_NAME),
user_uttered("bye", 1),
]
tracker = await self._get_tracker_after_reverts(
events, default_channel, default_nlg, default_domain
)
assert "bye" == tracker.latest_message.parse_data["intent"]["name"]
assert tracker.export_stories() == "## sender\n* bye\n"
def test_events_metadata():
# It should be possible to attach arbitrary metadata to any event and then
# retrieve it after getting the tracker dict representation.
events = [
ActionExecuted("one", metadata={"one": 1}),
user_uttered("two", 1, metadata={"two": 2}),
ActionExecuted(ACTION_LISTEN_NAME, metadata={"three": 3}),
]
events = get_tracker(events).current_state(EventVerbosity.ALL)["events"]
assert events[0]["metadata"] == {"one": 1}
assert events[1]["metadata"] == {"two": 2}
assert events[2]["metadata"] == {"three": 3}
def test_session_start_is_not_serialised(default_domain: Domain):
tracker = DialogueStateTracker("default", default_domain.slots)
# the retrieved tracker should be empty
assert len(tracker.events) == 0
# add SlotSet event
tracker.update(SlotSet("slot", "value"))
# add the two SessionStarted events and a user event
tracker.update(ActionExecuted(ACTION_SESSION_START_NAME))
tracker.update(SessionStarted())
tracker.update(UserUttered("say something"))
# make sure session start is not serialised
story = Story.from_events(tracker.events, "some-story01")
expected = """## some-story01
- slot{"slot": "value"}
* say something
"""
assert story.as_story_string(flat=True) == expected
def test_common_action_prefix():
this = [
ActionExecuted("action_listen"),
ActionExecuted("greet"),
UserUttered("hey"),
ActionExecuted("amazing"),
# until this point they are the same
SlotSet("my_slot", "a"),
ActionExecuted("a"),
ActionExecuted("after_a"),
]
other = [
ActionExecuted("action_listen"),
ActionExecuted("greet"),
UserUttered("hey"),
ActionExecuted("amazing"),
# until this point they are the same
SlotSet("my_slot", "b"),
ActionExecuted("b"),
async def test_reminder_cancelled_multi_user(
default_channel: CollectingOutputChannel, default_processor: MessageProcessor
):
sender_ids = [uuid.uuid4().hex, uuid.uuid4().hex]
trackers = []
for sender_id in sender_ids:
tracker = default_processor.tracker_store.get_or_create_tracker(sender_id)
tracker.update(UserUttered("test"))
tracker.update(ActionExecuted("action_reminder_reminder"))
tracker.update(
ReminderScheduled(
"greet", datetime.datetime.now(), kill_on_user_message=True
)
)
trackers.append(tracker)
# cancel all reminders (one) for the first user
trackers[0].update(ReminderCancelled())
for tracker in trackers:
default_processor.tracker_store.save(tracker)
await default_processor._schedule_reminders(
tracker.events, tracker, default_channel, default_processor.nlg
)
# check that the jobs were added
warnings.warn(
"MappingPolicy tried to predict unknown "
f"action '{action}'. Make sure all mapped actions are "
"listed in the domain."
)
else:
prediction[idx] = 1
if any(prediction):
logger.debug(
"The predicted intent '{}' is mapped to "
" action '{}' in the domain."
"".format(intent, action)
)
elif tracker.latest_action_name == action and action is not None:
latest_action = tracker.get_last_event_for(ActionExecuted)
assert latest_action.action_name == action
if latest_action.policy and latest_action.policy.endswith(
type(self).__name__
):
# this ensures that we only predict listen, if we predicted
# the mapped action
logger.debug(
"The mapped action, '{}', for this intent, '{}', was "
"executed last so MappingPolicy is returning to "
"action_listen.".format(action, intent)
)
idx = domain.index_for_action(ACTION_LISTEN_NAME)
prediction[idx] = 1
else:
logger.debug(
result += self._bot_string(s)
else:
# form is active
if self.story_string_helper.form_rejected:
if (
self.story_string_helper.form_prefix_string
and self.story_string_helper.form_validation
and s.action_name == self.story_string_helper.active_form
):
# if there is something in `form_prefix_string`,
# add action_listen before it,
# because this form user input will be ignored by core
# and therefore action_listen will not be automatically
# added during reading the stories
result += self._bot_string(
ActionExecuted(ACTION_LISTEN_NAME)
)
result += self.story_string_helper.form_prefix_string
elif self.story_string_helper.no_form_prefix_string:
result += self.story_string_helper.no_form_prefix_string
# form rejected, add story string without form prefix
result += self._bot_string(s)
else:
# form succeeded, so add story string with form prefix
result += self.story_string_helper.form_prefix_string
result += self._bot_string(s, FORM_PREFIX)
# remove all stored story strings
self._reset_stored_strings()
if s.action_name == self.story_string_helper.active_form:
# form was successfully executed