rasahq--rasa/tests/shared/core/test_domain.py

import copy
import json
import re
import textwrap
from pathlib import Path
import random
from typing import Dict, List, Text, Any, Union, Set, Optional

import pytest
from pytest import WarningsRecorder

from rasa.shared.exceptions import YamlSyntaxException, YamlException
import rasa.shared.utils.io
from rasa.shared.constants import (
    DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES,
    LATEST_TRAINING_DATA_FORMAT_VERSION,
    IGNORED_INTENTS,
)
from rasa.core import training
from rasa.core.featurizers.tracker_featurizers import MaxHistoryTrackerFeaturizer
from rasa.shared.core.slots import InvalidSlotTypeException, TextSlot
from rasa.shared.core.constants import (
    DEFAULT_INTENTS,
    SLOT_LISTED_ITEMS,
    SLOT_LAST_OBJECT,
    SLOT_LAST_OBJECT_TYPE,
    DEFAULT_KNOWLEDGE_BASE_ACTION,
    ENTITY_LABEL_SEPARATOR,
    DEFAULT_ACTION_NAMES,
)
from rasa.shared.core.domain import (
    InvalidDomain,
    SessionConfig,
    EntityProperties,
    ENTITY_ROLES_KEY,
    USED_ENTITIES_KEY,
    USE_ENTITIES_KEY,
    IGNORE_ENTITIES_KEY,
    State,
    Domain,
    KEY_FORMS,
    KEY_E2E_ACTIONS,
    KEY_INTENTS,
    KEY_ENTITIES,
    KEY_SLOTS,
)
from rasa.shared.core.trackers import DialogueStateTracker
from rasa.shared.core.events import ActionExecuted, SlotSet, UserUttered
from rasa.shared.utils.validation import YamlValidationException
from rasa.utils.common import EXPECTED_WARNINGS


def test_slots_states_before_user_utterance(domain: Domain):
    featurizer = MaxHistoryTrackerFeaturizer()
    tracker = DialogueStateTracker.from_events(
        "bla",
        evts=[
            SlotSet(domain.slots[0].name, "some_value"),
            ActionExecuted("utter_default"),
        ],
        slots=domain.slots,
    )
    trackers_as_states, _ = featurizer.training_states_and_labels([tracker], domain)
    expected_states = [[{"slots": {"name": (1.0,)}}]]
    assert trackers_as_states == expected_states


def test_create_train_data_no_history(domain: Domain, stories_path: Text):
    featurizer = MaxHistoryTrackerFeaturizer(max_history=1)
    training_trackers = training.load_data(stories_path, domain, augmentation_factor=0)

    assert len(training_trackers) == 4
    (decoded, _) = featurizer.training_states_and_labels(training_trackers, domain)

    # decoded needs to be sorted
    hashed = []
    for states in decoded:
        hashed.append(json.dumps(states, sort_keys=True))
    hashed = sorted(hashed, reverse=True)

    assert hashed == [
        "[{}]",
        '[{"prev_action": {"action_name": "utter_greet"}, "user": {"intent": "greet"}}]',
        '[{"prev_action": {"action_name": "utter_greet"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}]',
        '[{"prev_action": {"action_name": "utter_goodbye"}, "user": {"intent": "goodbye"}}]',
        '[{"prev_action": {"action_name": "utter_default"}, "user": {"intent": "default"}}]',
        '[{"prev_action": {"action_name": "utter_default"}, "slots": {"name": [1.0]}, "user": {"intent": "default"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "greet"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "goodbye"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "default"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"intent": "default"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}]',
    ]


def test_create_train_data_with_history(domain: Domain, stories_path: Text):
    featurizer = MaxHistoryTrackerFeaturizer(max_history=4)
    training_trackers = training.load_data(stories_path, domain, augmentation_factor=0)
    assert len(training_trackers) == 4
    (decoded, _) = featurizer.training_states_and_labels(training_trackers, domain)

    # decoded needs to be sorted
    hashed = []
    for states in decoded:
        hashed.append(json.dumps(states, sort_keys=True))
    hashed = sorted(hashed)

    assert hashed == [
        '[{"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}, {"prev_action": {"action_name": "utter_greet"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}, {"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"intent": "default"}}, {"prev_action": {"action_name": "utter_default"}, "slots": {"name": [1.0]}, "user": {"intent": "default"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "default"}}, {"prev_action": {"action_name": "utter_default"}, "user": {"intent": "default"}}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "goodbye"}}, {"prev_action": {"action_name": "utter_goodbye"}, "user": {"intent": "goodbye"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "greet"}}, {"prev_action": {"action_name": "utter_greet"}, "user": {"intent": "greet"}}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "default"}}, {"prev_action": {"action_name": "utter_default"}, "user": {"intent": "default"}}]',
        '[{"prev_action": {"action_name": "utter_greet"}, "user": {"intent": "greet"}}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "default"}}, {"prev_action": {"action_name": "utter_default"}, "user": {"intent": "default"}}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "goodbye"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}, {"prev_action": {"action_name": "utter_greet"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}, {"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"intent": "default"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}, {"prev_action": {"action_name": "utter_greet"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "slots": {"name": [1.0]}, "user": {"entities": ["name"], "intent": "greet"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "goodbye"}}, {"prev_action": {"action_name": "utter_goodbye"}, "user": {"intent": "goodbye"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "goodbye"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "greet"}}, {"prev_action": {"action_name": "utter_greet"}, "user": {"intent": "greet"}}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "default"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "greet"}}, {"prev_action": {"action_name": "utter_greet"}, "user": {"intent": "greet"}}]',
        '[{}, {"prev_action": {"action_name": "action_listen"}, "user": {"intent": "greet"}}]',
        "[{}]",
    ]


def check_for_too_many_entities_and_remove_them(state: State) -> State:
    # we ignore entities where there are > 1 of them:
    # entities come from dictionary keys; as a result, they are stored
    # in different order in the tuple which makes the test unstable
    if (
        state.get("user")
        and state.get("user", {}).get("entities")
        and len(state.get("user").get("entities")) > 1
    ):
        state.get("user")["entities"] = ()
    return state


def test_create_train_data_unfeaturized_entities():
    domain_file = "data/test_domains/default_unfeaturized_entities.yml"
    stories_file = "data/test_yaml_stories/stories_unfeaturized_entities.yml"
    domain = Domain.load(domain_file)
    featurizer = MaxHistoryTrackerFeaturizer(max_history=1)
    training_trackers = training.load_data(stories_file, domain, augmentation_factor=0)

    assert len(training_trackers) == 2
    (decoded, _) = featurizer.training_states_and_labels(training_trackers, domain)

    # decoded needs to be sorted
    hashed = []
    for states in decoded:
        new_states = [
            check_for_too_many_entities_and_remove_them(state) for state in states
        ]

        hashed.append(json.dumps(new_states, sort_keys=True))
    hashed = sorted(hashed, reverse=True)

    assert hashed == [
        "[{}]",
        '[{"prev_action": {"action_name": "utter_greet"}, "user": {"intent": "greet"}}]',
        '[{"prev_action": {"action_name": "utter_greet"}, "user": {"entities": ["name"], "intent": "greet"}}]',
        '[{"prev_action": {"action_name": "utter_goodbye"}, "user": {"intent": "goodbye"}}]',
        '[{"prev_action": {"action_name": "utter_default"}, "user": {"intent": "why"}}]',
        '[{"prev_action": {"action_name": "utter_default"}, "user": {"intent": "thank"}}]',
        '[{"prev_action": {"action_name": "utter_default"}, "user": {"entities": [], "intent": "default"}}]',
        '[{"prev_action": {"action_name": "utter_default"}, "user": {"entities": [], "intent": "ask"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "why"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "thank"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "greet"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "goodbye"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"entities": [], "intent": "default"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"entities": [], "intent": "ask"}}]',
        '[{"prev_action": {"action_name": "action_listen"}, "user": {"entities": ["name"], "intent": "greet"}}]',
    ]


def test_domain_from_template(domain: Domain):
    assert not domain.is_empty()
    assert len(domain.intents) == 10 + len(DEFAULT_INTENTS)
    assert len(domain.action_names_or_texts) == 18


def test_avoid_action_repetition(domain: Domain):
    domain = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        actions:
        - utter_greet
        responses:
            utter_greet:
            - text: "hi"
        """
    )

    assert len(domain.action_names_or_texts) == len(DEFAULT_ACTION_NAMES) + 1


def test_custom_slot_type(tmpdir: Path):
    domain_path = str(tmpdir / "domain.yml")
    rasa.shared.utils.io.write_text_file(
        """
       slots:
         custom:
           type: tests.core.conftest.CustomSlot
           mappings:
           - type: from_text

       responses:
         utter_greet:
           - text: hey there! """,
        domain_path,
    )
    Domain.load(domain_path)


@pytest.mark.parametrize(
    "domain_unkown_slot_type",
    [
        """
    slots:
        custom:
         type: tests.core.conftest.Unknown
         mappings:
         - type: from_text

    responses:
        utter_greet:
         - text: hey there!""",
        """
    slots:
        custom:
         type: blubblubblub
         mappings:
         - type: from_text

    responses:
        utter_greet:
         - text: hey there!""",
    ],
)
def test_domain_fails_on_unknown_custom_slot_type(tmpdir, domain_unkown_slot_type):
    domain_path = str(tmpdir / "domain.yml")
    rasa.shared.utils.io.write_text_file(domain_unkown_slot_type, domain_path)
    with pytest.raises(InvalidSlotTypeException):
        Domain.load(domain_path)


def test_custom_slot_type_with_custom_key():
    domain = Domain.load("data/test_domains/custom_slot_domain.yml")

    assert domain.slots[0].limit == 1000


@pytest.mark.parametrize(
    "domain_invalid_type_for_slot_key",
    [
        """slots:
            limit:
                type: text
                influence_conversation: yes
                mappings:
                - type: from_entity
                  entity: limit""",
        """slots:
            limit:
                type: text
                values: notalist
                mappings:
                - type: from_entity
                  entity: limit""",
        """slots:
            limit:
                type: text
                min_value: notanumber
                mappings:
                - type: from_entity
                  entity: limit""",
        """slots:
            limit:
                type: text
                max_value: notanumber
                mappings:
                - type: from_entity
                  entity: limit""",
    ],
)
def test_domain_fails_on_invalid_type_for_known_slot_key(
    tmpdir, domain_invalid_type_for_slot_key
):
    domain_path = str(tmpdir / "domain.yml")
    rasa.shared.utils.io.write_text_file(domain_invalid_type_for_slot_key, domain_path)
    with pytest.raises(YamlValidationException):
        Domain.load(domain_path)


def test_domain_to_dict():
    test_yaml = textwrap.dedent(
        f"""
    actions:
    - action_save_world
    config:
      store_entities_as_slots: true
    entities: []
    forms:
      some_form:
       required_slots: []
    intents: []
    responses:
      utter_greet:
      - text: hey there!
    session_config:
      carry_over_slots_to_new_session: true
      session_expiration_time: 60
    {KEY_E2E_ACTIONS}:
    - Hello, dear user
    - what's up
    slots:
      some_slot:
        type: categorical
        values:
        - high
        - low
        mappings:
        - type: from_text"""
    )

    domain_as_dict = Domain.from_yaml(test_yaml).as_dict()

    assert domain_as_dict == {
        "version": LATEST_TRAINING_DATA_FORMAT_VERSION,
        "actions": ["action_save_world"],
        "config": {"store_entities_as_slots": True},
        KEY_E2E_ACTIONS: ["Hello, dear user", "what's up"],
        "forms": {"some_form": {"required_slots": []}},
        "responses": {"utter_greet": [{"text": "hey there!"}]},
        "session_config": {
            "carry_over_slots_to_new_session": True,
            "session_expiration_time": 60,
        },
        "slots": {
            "some_slot": {
                "values": ["high", "low"],
                "mappings": [{"type": "from_text"}],
                "type": "categorical",
            }
        },
    }


def test_domain_to_yaml():
    test_yaml = f"""
version: '{LATEST_TRAINING_DATA_FORMAT_VERSION}'
actions:
- action_save_world
config:
  store_entities_as_slots: true
e2e_actions: []
entities: []
forms: {{}}
intents: []
responses:
  utter_greet:
  - text: hey there!
session_config:
  carry_over_slots_to_new_session: true
  session_expiration_time: {DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES}
slots: {{}}
"""

    domain = Domain.from_yaml(test_yaml)
    actual_yaml = domain.as_yaml()

    expected_yaml = f"""
version: '{LATEST_TRAINING_DATA_FORMAT_VERSION}'
actions:
- action_save_world
config:
  store_entities_as_slots: true
responses:
  utter_greet:
  - text: hey there!
session_config:
  carry_over_slots_to_new_session: true
  session_expiration_time: {DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES}
"""

    actual = rasa.shared.utils.io.read_yaml(actual_yaml)
    expected = rasa.shared.utils.io.read_yaml(expected_yaml)
    assert actual == expected


def test_merge_yaml_domains():
    test_yaml_1 = f"""config:
  store_entities_as_slots: true
entities: []
intents: []
slots: {{}}
responses:
  utter_greet:
  - text: hey there!
{KEY_E2E_ACTIONS}:
- Hi"""

    test_yaml_2 = f"""config:
  store_entities_as_slots: false
session_config:
    session_expiration_time: 20
    carry_over_slots: true
entities:
- cuisine
intents:
- greet
slots:
  cuisine:
    type: text
    mappings:
    - type: from_text
{KEY_E2E_ACTIONS}:
- Bye
responses:
  utter_goodbye:
  - text: bye!
  utter_greet:
  - text: hey you!"""

    domain_1 = Domain.from_yaml(test_yaml_1)
    domain_2 = Domain.from_yaml(test_yaml_2)

    domain = domain_1.merge(domain_2)

    # single attribute should be taken from domain_1
    assert domain.store_entities_as_slots
    # conflicts should be taken from domain_1
    assert domain.responses == {
        "utter_greet": [{"text": "hey there!"}],
        "utter_goodbye": [{"text": "bye!"}],
    }
    # lists should be deduplicated and merged
    assert domain.intents == sorted(["greet", *DEFAULT_INTENTS])
    assert domain.entities == ["cuisine"]
    assert isinstance(domain.slots[0], TextSlot)
    assert domain.slots[0].name == "cuisine"
    assert sorted(domain.user_actions) == sorted(["utter_greet", "utter_goodbye"])
    assert domain.session_config == SessionConfig(20, True)

    domain = domain_1.merge(domain_2, override=True)
    # single attribute should be taken from domain_2
    assert not domain.store_entities_as_slots
    # conflicts should take value from domain_2
    assert domain.responses == {
        "utter_greet": [{"text": "hey you!"}],
        "utter_goodbye": [{"text": "bye!"}],
    }
    assert domain.session_config == SessionConfig(20, True)
    assert domain.action_texts == ["Bye", "Hi"]


@pytest.mark.parametrize("default_intent", DEFAULT_INTENTS)
def test_merge_yaml_domains_with_default_intents(default_intent: Text):
    test_yaml_1 = """intents: []"""

    # this domain contains an overridden default intent
    test_yaml_2 = f"""intents:
- greet
- {default_intent}"""

    domain_1 = Domain.from_yaml(test_yaml_1)
    domain_2 = Domain.from_yaml(test_yaml_2)

    domain = domain_1.merge(domain_2)

    # check that the default intents were merged correctly
    assert default_intent in domain.intents
    assert domain.intents == sorted(["greet", *DEFAULT_INTENTS])

    # ensure that the default intent is contain the domain's dictionary dump
    assert default_intent in domain.as_dict()[KEY_INTENTS]


def test_merge_session_config_if_first_is_not_default():
    yaml1 = """
session_config:
    session_expiration_time: 20
    carry_over_slots: true"""

    yaml2 = """
 session_config:
    session_expiration_time: 40
    carry_over_slots: true
    """

    domain1 = Domain.from_yaml(yaml1)
    domain2 = Domain.from_yaml(yaml2)

    merged = domain1.merge(domain2)
    assert merged.session_config == SessionConfig(20, True)

    merged = domain1.merge(domain2, override=True)
    assert merged.session_config == SessionConfig(40, True)


def test_merge_with_empty_domain():
    domain = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        config:
          store_entities_as_slots: false
        session_config:
            session_expiration_time: 20
            carry_over_slots: true
        entities:
        - cuisine
        intents:
        - greet
        slots:
          cuisine:
            type: text
            mappings:
            - type: from_text
        responses:
          utter_goodbye:
          - text: bye!
          utter_greet:
          - text: hey you!
        """
    )
    empty_domain = Domain.empty()
    merged = empty_domain.merge(domain, override=True)
    assert merged.as_dict() == domain.as_dict()


@pytest.mark.parametrize("other", [Domain.empty(), None])
def test_merge_with_empty_other_domain(other: Optional[Domain]):
    domain = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        config:
          store_entities_as_slots: false
        session_config:
            session_expiration_time: 20
            carry_over_slots: true
        entities:
        - cuisine
        intents:
        - greet
        slots:
          cuisine:
            type: text
            mappings:
            - type: from_text
        responses:
          utter_goodbye:
          - text: bye!
          utter_greet:
          - text: hey you!
        """
    )

    merged = domain.merge(other, override=True)

    assert merged.as_dict() == domain.as_dict()


def test_merge_domain_with_forms():
    test_yaml_1 = """
    slots:
      slot1:
        type: text
        mappings:
        - type: from_text
    forms:
      my_form:
        required_slots:
            - slot1

      my_form2:
        required_slots: []
    """

    test_yaml_2 = """
    slots:
      slot1:
        type: text
        mappings:
        - type: from_text
    forms:
      my_form3:
        required_slots:
          - slot1
    """

    domain_1 = Domain.from_yaml(test_yaml_1)
    domain_2 = Domain.from_yaml(test_yaml_2)

    domain = domain_1.merge(domain_2)

    expected_number_of_forms = 3
    assert len(domain.form_names) == expected_number_of_forms
    assert len(domain.forms) == expected_number_of_forms


@pytest.mark.parametrize(
    "intents, entity_properties, intent_properties",
    [
        (
            ["greet", "goodbye"],
            {
                "entities": ["entity", "other", "third"],
                "roles": {"entity": ["role-1", "role-2"]},
                "groups": {},
                "default_ignored_entities": [],
            },
            {
                "greet": {
                    USED_ENTITIES_KEY: [
                        "entity",
                        f"entity{ENTITY_LABEL_SEPARATOR}role-1",
                        f"entity{ENTITY_LABEL_SEPARATOR}role-2",
                        "other",
                        "third",
                    ]
                },
                "goodbye": {
                    USED_ENTITIES_KEY: [
                        "entity",
                        f"entity{ENTITY_LABEL_SEPARATOR}role-1",
                        f"entity{ENTITY_LABEL_SEPARATOR}role-2",
                        "other",
                        "third",
                    ]
                },
            },
        ),
        (
            [{"greet": {USE_ENTITIES_KEY: []}}, "goodbye"],
            {
                "entities": ["entity", "other", "third"],
                "roles": {},
                "groups": {"other": ["1", "2"]},
                "default_ignored_entities": [],
            },
            {
                "greet": {USED_ENTITIES_KEY: []},
                "goodbye": {
                    USED_ENTITIES_KEY: [
                        "entity",
                        "other",
                        f"other{ENTITY_LABEL_SEPARATOR}1",
                        f"other{ENTITY_LABEL_SEPARATOR}2",
                        "third",
                    ]
                },
            },
        ),
        (
            [
                {
                    "greet": {
                        "triggers": "utter_goodbye",
                        USE_ENTITIES_KEY: ["entity"],
                        IGNORE_ENTITIES_KEY: ["other"],
                    }
                },
                "goodbye",
            ],
            {
                "entities": ["entity", "other", "third", "unused"],
                "roles": {"entity": ["role"], "other": ["role"]},
                "groups": {},
                "default_ignored_entities": ["unused"],
            },
            {
                "greet": {
                    "triggers": "utter_goodbye",
                    USED_ENTITIES_KEY: [
                        "entity",
                        f"entity{ENTITY_LABEL_SEPARATOR}role",
                    ],
                },
                "goodbye": {
                    USED_ENTITIES_KEY: [
                        "entity",
                        f"entity{ENTITY_LABEL_SEPARATOR}role",
                        "other",
                        f"other{ENTITY_LABEL_SEPARATOR}role",
                        "third",
                    ]
                },
            },
        ),
        (
            [
                {"greet": {"triggers": "utter_goodbye", USE_ENTITIES_KEY: None}},
                {"goodbye": {USE_ENTITIES_KEY: [], IGNORE_ENTITIES_KEY: []}},
            ],
            {
                "entities": ["entity", "other", "third"],
                "roles": {},
                "groups": {},
                "default_ignored_entities": [],
            },
            {
                "greet": {USED_ENTITIES_KEY: [], "triggers": "utter_goodbye"},
                "goodbye": {USED_ENTITIES_KEY: []},
            },
        ),
        (
            [
                "greet",
                "goodbye",
                {"chitchat": {"is_retrieval_intent": True, "use_entities": None}},
            ],
            {
                "entities": ["entity", "other", "third"],
                "roles": {},
                "groups": {},
                "default_ignored_entities": [],
            },
            {
                "greet": {USED_ENTITIES_KEY: ["entity", "other", "third"]},
                "goodbye": {USED_ENTITIES_KEY: ["entity", "other", "third"]},
                "chitchat": {USED_ENTITIES_KEY: [], "is_retrieval_intent": True},
            },
        ),
    ],
)
def test_collect_intent_properties(
    intents: Union[Set[Text], List[Union[Text, Dict[Text, Any]]]],
    entity_properties: Dict[Text, Union[List[Text], Dict[Text, List[Text]]]],
    intent_properties: Dict[Text, Dict[Text, Union[bool, List]]],
):
    entity_properties = EntityProperties(**entity_properties)
    Domain._add_default_intents(intent_properties, entity_properties)

    assert (
        Domain.collect_intent_properties(intents, entity_properties)
        == intent_properties
    )


@pytest.mark.parametrize(
    "entities, entity_properties",
    [
        (
            ["plain_entity", {"ignored_entity": {"influence_conversation": False}}],
            {
                "entities": ["plain_entity", "ignored_entity"],
                "roles": {},
                "groups": {},
                "default_ignored_entities": ["ignored_entity"],
            },
        ),
    ],
)
def test_collect_entity_properties(
    entities: List[Union[Text, Dict[Text, Any]]],
    entity_properties: Dict[Text, Union[List[Text], Dict[Text, List[Text]]]],
):
    expected_entity_properties = EntityProperties(**entity_properties)
    assert Domain.collect_entity_properties(entities) == expected_entity_properties


def test_load_domain_from_directory_tree():
    domain_path = "data/test_domains/test_domain_from_directory_tree"

    actual = Domain.load(domain_path)
    expected_intents = [
        "utter_root",
        "utter_root2",
        "utter_skill_1",
        "utter_skill_2",
        "utter_subskill",
    ]
    expected_entities = ["ball", "chess", "monopoly", "cluedo", "pandemic"]
    expected_responses = {
        "utter_greet": [{"text": "Hey! How are you?"}],
        "utter_cheer_up": [
            {
                "text": "Here is something to cheer you up:",
                "image": "https://i.imgur.com/nGF1K8f.jpg",
            }
        ],
    }
    assert set(expected_intents).issubset(set(actual.intents))
    assert set(expected_entities) == (set(actual.entities))
    assert set(expected_responses) == (set(actual.responses))


def test_domain_from_multiple_files():
    domain_path = "data/test_domains/test_domain_from_multiple_files"
    domain = Domain.load(domain_path)

    expected_intents = [
        "affirm",
        "are_you_there",
        "back",
        "bot_challenge",
        "cure_network",
        "delay",
        "deny",
        "device_selection_scaffold",
        "drum_clocks",
        "drum_lampshades",
        "drum_robot",
        "drum_robot_chocolate",
        "drum_soups",
        "drum_wallets",
        "endless_love",
        "exchange_wallet",
        "finish_humble_selection",
        "finish_selection",
        "finish_selection_line",
        "greeting",
        "humble_selection",
        "humble_selection_scaffold",
        "main_menu",
        "nlu_fallback",
        "open_wallet",
        "out_of_scope",
        "profanity",
        "restart",
        "run_finish",
        "run_finish_recent",
        "run_finished",
        "selection_troubleshooting",
        "self_selection",
        "session_start",
        "thanks",
        "unsure_selection_scaffold",
        "view_offers",
    ]
    expected_entities = [
        "caramel_robot",
        "chocolate_robot",
        "other_robot",
        "pistachio_robot",
        "rum_and_raisin_robot",
        "strawberry_robot",
        "vanilla_robot",
    ]
    expected_actions = [
        "action_increase_15",
        "action_prevent_20",
        "action_utter_cure_standard",
        "action_utter_drum_menu",
        "action_utter_main_menu",
        "action_utter_previous_message",
        "action_utter_robot_menu",
        "action_utter_smalltalk_greeting",
        "utter_anythingelse_menu",
        "utter_bot_challenge",
        "utter_cure_specific",
        "utter_cure_standard",
        "utter_drumclocks",
        "utter_drumlampshades",
        "utter_drumsoups",
        "utter_drumwallets",
        "utter_finish_humble_selection",
        "utter_finish_selection",
        "utter_finish_selection_line",
        "utter_greengrey_wallet",
        "utter_horn_selection_scaffold",
        "utter_humble_selection",
        "utter_humble_selection_scaffold",
        "utter_im_here",
        "utter_non_standard",
        "utter_open_wallet_options",
        "utter_profanity",
        "utter_run_finish",
        "utter_run_finish_recent",
        "utter_run_finished",
        "utter_selection_issues",
        "utter_smalltalk_greeting",
        "utter_std_drum_menu",
        "utter_thanks_response",
        "utter_tmo_love",
        "utter_amazement",
        "utter_default",
        "utter_goodbye",
        "utter_greet",
    ]
    expected_forms = {
        "robot_form": {"required_slots": ["propose_simulation", "display_cure_method"]}
    }
    expected_responses = {
        "utter_greet": [{"text": "hey there!"}],
        "utter_goodbye": [{"text": "goodbye :("}],
        "utter_default": [{"text": "default message"}],
        "utter_amazement": [{"text": "awesomness!"}],
    }
    expected_slots = [
        "activate_double_simulation",
        "activate_simulation",
        "display_cure_method",
        "display_drum_cure_horns",
        "display_method_artwork",
        "drumAllClocks",
        "drumAllLampshades",
        "drumAllSoups",
        "drumChocolateWallets",
        "drumClockAdapters",
        "drumClockCovers",
        "drumClocksChocolate",
        "drumClocksStrawberry",
        "drumMindspace",
        "drumOtherWallets",
        "drumSnareWallets",
        "drumSoupChocolate",
        "drumSoupStrawberry",
        "drumStrawberryWallets",
        "greenOrGrey",
        "humbleSelection",
        "humbleSelectionManagement",
        "humbleSelectionStatus",
        "offers",
        "requested_slot",
        "session_started_metadata",
    ]

    domain_slots = []

    for slot in domain.slots:
        domain_slots.append(slot.name)

    assert expected_intents == domain.intents
    assert expected_entities == sorted(domain.entities)
    assert sorted(expected_actions) == sorted(domain.user_actions)
    assert expected_responses == domain.responses
    assert expected_forms == domain.forms
    assert domain.session_config.session_expiration_time == 360
    assert expected_slots == sorted(domain_slots)


def test_domain_warnings(domain: Domain):
    warning_types = [
        "action_warnings",
        "intent_warnings",
        "entity_warnings",
        "slot_warnings",
    ]

    actions = ["action_1", "action_2"]
    intents = ["intent_1", "intent_2"]
    entities = ["entity_1", "entity_2"]
    slots = ["slot_1", "slot_2"]
    domain_warnings = domain.domain_warnings(
        intents=intents, entities=entities, actions=actions, slots=slots
    )

    # elements not found in domain should be in `in_training_data` diff
    for _type, elements in zip(warning_types, [actions, intents, entities]):
        assert set(domain_warnings[_type]["in_training_data"]) == set(elements)

    # all other domain elements should be in `in_domain` diff
    for _type, elements in zip(
        warning_types,
        [domain.user_actions + domain.form_names, domain.intents, domain.entities],
    ):
        assert set(domain_warnings[_type]["in_domain"]) == set(elements)

    # fully aligned domain and elements should yield empty diff
    domain_warnings = domain.domain_warnings(
        intents=domain.intents,
        entities=domain.entities,
        actions=domain.user_actions + domain.form_names,
        slots=[s.name for s in domain._user_slots],
    )

    for diff_dict in domain_warnings.values():
        assert all(not diff_set for diff_set in diff_dict.values())


def test_unfeaturized_slot_in_domain_warnings():
    # create empty domain
    featurized_slot_name = "text_slot"
    unfeaturized_slot_name = "unfeaturized_slot"
    domain = Domain.from_dict(
        {
            "slots": {
                featurized_slot_name: {
                    "initial_value": "value2",
                    "type": "text",
                    "mappings": [{"type": "from_text"}],
                },
                unfeaturized_slot_name: {
                    "type": "text",
                    "initial_value": "value1",
                    "influence_conversation": False,
                    "mappings": [{"type": "from_text"}],
                },
            }
        }
    )

    # ensure both are in domain
    for slot in (featurized_slot_name, unfeaturized_slot_name):
        assert slot in [slot.name for slot in domain.slots]

    # text slot should appear in domain warnings, unfeaturized slot should not
    in_domain_slot_warnings = domain.domain_warnings()["slot_warnings"]["in_domain"]
    assert featurized_slot_name in in_domain_slot_warnings
    assert unfeaturized_slot_name not in in_domain_slot_warnings


def test_check_domain_sanity_on_invalid_domain():
    with pytest.raises(InvalidDomain):
        Domain(
            intents={},
            entities=[],
            slots=[],
            responses={},
            action_names=["random_name", "random_name"],
            forms={},
            data={},
        )

    with pytest.raises(InvalidDomain):
        Domain(
            intents={},
            entities=[],
            slots=[
                TextSlot("random_name", mappings=[{}]),
                TextSlot("random_name", mappings=[{}]),
            ],
            responses={},
            action_names=[],
            forms={},
            data={},
        )

    with pytest.raises(InvalidDomain):
        Domain(
            intents={},
            entities=["random_name", "random_name", "other_name", "other_name"],
            slots=[],
            responses={},
            action_names=[],
            forms={},
            data={},
        )


def test_load_on_invalid_domain_duplicate_intents():
    with pytest.raises(InvalidDomain):
        Domain.load("data/test_domains/duplicate_intents.yml")


def test_load_on_invalid_domain_duplicate_actions():
    with pytest.raises(InvalidDomain):
        Domain.load("data/test_domains/duplicate_actions.yml")


def test_schema_error_with_forms_as_lists():
    with pytest.raises(YamlException):
        Domain.from_yaml(
            """
        version: '3.0'
        forms: []
        """
        )


def test_schema_error_with_forms_and_slots_but_without_required_slots_key():
    with pytest.raises(YamlException):
        Domain.from_yaml(
            """
        version: '3.0'
        forms:
          my_form:
            cool_slot:
            - type: from_text
        """
        )


def test_load_on_invalid_domain_duplicate_responses():
    with pytest.raises(YamlSyntaxException):
        Domain.load("data/test_domains/duplicate_responses.yml")


def test_load_on_invalid_domain_duplicate_entities():
    with pytest.raises(InvalidDomain):
        Domain.load("data/test_domains/duplicate_entities.yml")


def test_load_domain_with_entity_roles_groups():
    domain = Domain.load("data/test_domains/travel_form.yml")

    assert domain.entities is not None
    assert "GPE" in domain.entities
    assert "name" in domain.entities
    assert "name" not in domain.entity_properties.roles
    assert "GPE" in domain.entity_properties.roles
    assert "origin" in domain.entity_properties.roles["GPE"]
    assert "destination" in domain.entity_properties.roles["GPE"]


def test_is_empty():
    assert Domain.empty().is_empty()


def test_load_intents_from_as_dict_representation():
    domain_path = "data/test_domains/default_unfeaturized_entities.yml"
    domain = Domain.load(domain_path)
    transformed = domain.as_dict().get(KEY_INTENTS)

    expected = [
        {"ask": {USE_ENTITIES_KEY: True}},
        {"default": {IGNORE_ENTITIES_KEY: ["unrelated_recognized_entity"]}},
        {"goodbye": {USE_ENTITIES_KEY: []}},
        {"greet": {USE_ENTITIES_KEY: ["name", "used_entity"]}},
        "pure_intent",
        {"thank": {USE_ENTITIES_KEY: []}},
        {"why": {USE_ENTITIES_KEY: []}},
    ]

    assert transformed == expected


def test_load_intents_with_entities_from_as_dict():
    domain_path = "data/test_domains/test_domain_from_directory_for_entities"
    domain = Domain.load(domain_path)
    transformed = domain.as_dict().get(KEY_INTENTS)

    expected = [
        {"certify": {USE_ENTITIES_KEY: True}},
        {"play": {USE_ENTITIES_KEY: ["ball", "chess"]}},
        "question",
        {"stow_away": {USE_ENTITIES_KEY: True}},
        {
            "support_encouraging": {
                USE_ENTITIES_KEY: ["automatic_cupcakes", "anti_freeze_blankets"]
            }
        },
        {"vacationing": {"ignore_entities": ["tornadoes"]}},
    ]

    assert transformed == expected


def test_load_intents_for_file_from_as_dict():
    domain_path = "data/test_domains/default_with_mapping.yml"
    domain = Domain.load(domain_path)
    transformed = domain.as_dict().get(KEY_INTENTS)

    expected = [
        {"default": {"triggers": "utter_default"}},
        "goodbye",
        {"greet": {"triggers": "utter_greet"}},
    ]

    assert transformed == expected


def test_load_intents_with_entity_roles_groups_from_as_dict():
    domain_path = "data/test_domains/travel_form.yml"
    domain = Domain.load(domain_path)
    transformed = domain.as_dict().get(KEY_INTENTS)

    expected = [
        {"greet": {IGNORE_ENTITIES_KEY: ["GPE"]}},
        {"inform": {USE_ENTITIES_KEY: ["GPE"]}},
    ]

    assert transformed == expected


def test_load_entities_from_as_dict():
    domain_path = "data/test_domains/travel_form.yml"
    domain = Domain.load(domain_path)
    transformed = domain.as_dict().get(KEY_ENTITIES)

    expected = [{"GPE": {ENTITY_ROLES_KEY: ["destination", "origin"]}}, "name"]

    assert transformed == expected


def test_not_add_knowledge_base_slots():
    test_domain = Domain.empty()

    slot_names = [s.name for s in test_domain.slots]

    assert SLOT_LISTED_ITEMS not in slot_names
    assert SLOT_LAST_OBJECT not in slot_names
    assert SLOT_LAST_OBJECT_TYPE not in slot_names


def test_add_knowledge_base_slots():
    test_domain = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        actions:
        - {DEFAULT_KNOWLEDGE_BASE_ACTION}
        """
    )

    slot_names = [s.name for s in test_domain.slots]

    assert SLOT_LISTED_ITEMS in slot_names
    assert SLOT_LAST_OBJECT in slot_names
    assert SLOT_LAST_OBJECT_TYPE in slot_names


@pytest.mark.parametrize(
    "input_domain, expected_session_expiration_time, expected_carry_over_slots",
    [
        (
            f"""session_config:
    session_expiration_time: {DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES}
    carry_over_slots_to_new_session: true""",
            DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES,
            True,
        ),
        ("", DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES, True),
        (
            """session_config:
        carry_over_slots_to_new_session: false""",
            DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES,
            False,
        ),
        (
            """session_config:
        session_expiration_time: 20.2
        carry_over_slots_to_new_session: False""",
            20.2,
            False,
        ),
        ("""session_config: {}""", DEFAULT_SESSION_EXPIRATION_TIME_IN_MINUTES, True),
    ],
)
def test_session_config(
    input_domain,
    expected_session_expiration_time: float,
    expected_carry_over_slots: bool,
):
    domain = Domain.from_yaml(input_domain)
    assert (
        domain.session_config.session_expiration_time
        == expected_session_expiration_time
    )
    assert domain.session_config.carry_over_slots == expected_carry_over_slots


def test_domain_as_dict_with_session_config():
    session_config = SessionConfig(123, False)
    domain = Domain([], [], [], {}, [], {}, {}, None, True, session_config)

    serialized = domain.as_dict()
    deserialized = Domain.from_dict(serialized)

    assert deserialized.session_config == session_config


@pytest.mark.parametrize(
    "session_config, enabled",
    [
        (SessionConfig(0, True), False),
        (SessionConfig(1, True), True),
        (SessionConfig(-1, False), False),
    ],
)
def test_are_sessions_enabled(session_config: SessionConfig, enabled: bool):
    assert session_config.are_sessions_enabled() == enabled


def test_domain_from_dict_does_not_change_input():
    input_before = {
        "intents": [
            {"greet": {USE_ENTITIES_KEY: ["name"]}},
            {"default": {IGNORE_ENTITIES_KEY: ["unrelated_recognized_entity"]}},
            {"goodbye": {USE_ENTITIES_KEY: None}},
            {"thank": {USE_ENTITIES_KEY: False}},
            {"ask": {USE_ENTITIES_KEY: True}},
            {"why": {USE_ENTITIES_KEY: []}},
            "pure_intent",
        ],
        "entities": ["name", "unrelated_recognized_entity", "other"],
        "slots": {"name": {"type": "text", "mappings": [{"type": "from_text"}]}},
        "responses": {
            "utter_greet": [{"text": "hey there {name}!"}],
            "utter_goodbye": [{"text": "goodbye 😢"}, {"text": "bye bye 😢"}],
            "utter_default": [{"text": "default message"}],
        },
    }

    input_after = copy.deepcopy(input_before)
    Domain.from_dict(input_after)

    assert input_after == input_before


@pytest.mark.parametrize(
    "domain_dict", [{}, {"intents": DEFAULT_INTENTS}, {"intents": [DEFAULT_INTENTS[0]]}]
)
def test_add_default_intents(domain_dict: Dict):
    domain = Domain.from_dict(domain_dict)

    assert all(intent_name in domain.intents for intent_name in DEFAULT_INTENTS)


def test_domain_deepcopy(domain: Domain):
    new_domain = copy.deepcopy(domain)

    assert isinstance(new_domain, Domain)

    # equalities
    assert new_domain.intent_properties == domain.intent_properties
    assert new_domain.overridden_default_intents == domain.overridden_default_intents
    assert new_domain.entities == domain.entities
    assert new_domain.forms == domain.forms
    assert new_domain.form_names == domain.form_names
    assert new_domain.responses == domain.responses
    assert new_domain.action_texts == domain.action_texts
    assert new_domain.session_config == domain.session_config
    assert new_domain._custom_actions == domain._custom_actions
    assert new_domain.user_actions == domain.user_actions
    assert new_domain.action_names_or_texts == domain.action_names_or_texts
    assert new_domain.store_entities_as_slots == domain.store_entities_as_slots

    # not the same objects
    assert new_domain is not domain
    assert new_domain.intent_properties is not domain.intent_properties
    assert (
        new_domain.overridden_default_intents is not domain.overridden_default_intents
    )
    assert new_domain.entities is not domain.entities
    assert new_domain.forms is not domain.forms
    assert new_domain.form_names is not domain.form_names
    assert new_domain.slots is not domain.slots
    assert new_domain.responses is not domain.responses
    assert new_domain.action_texts is not domain.action_texts
    assert new_domain.session_config is not domain.session_config
    assert new_domain._custom_actions is not domain._custom_actions
    assert new_domain.user_actions is not domain.user_actions
    assert new_domain.action_names_or_texts is not domain.action_names_or_texts


@pytest.mark.parametrize(
    "response_key, validation",
    [("utter_chitchat/faq", True), ("utter_chitchat", False)],
)
def test_is_retrieval_intent_response(response_key, validation, domain: Domain):
    assert domain.is_retrieval_intent_response((response_key, [{}])) == validation


def test_retrieval_intent_response_seggregation():
    domain = Domain.load("data/test_domains/mixed_retrieval_intents.yml")
    assert domain.responses != domain.retrieval_intent_responses
    assert domain.responses and domain.retrieval_intent_responses
    assert list(domain.retrieval_intent_responses.keys()) == [
        "utter_chitchat/ask_weather",
        "utter_chitchat/ask_name",
    ]


def test_get_featurized_entities():
    domain = Domain.load("data/test_domains/travel_form.yml")

    user_uttered = UserUttered(
        text="Hello, I am going to London",
        intent={"name": "greet", "confidence": 1.0},
        entities=[{"entity": "GPE", "value": "London", "role": "destination"}],
    )

    featurized_entities = domain._get_featurized_entities(user_uttered)

    assert featurized_entities == set()

    user_uttered = UserUttered(
        text="I am going to London",
        intent={"inform": "greet", "confidence": 1.0},
        entities=[{"entity": "GPE", "value": "London", "role": "destination"}],
    )

    featurized_entities = domain._get_featurized_entities(user_uttered)

    assert featurized_entities == {"GPE", f"GPE{ENTITY_LABEL_SEPARATOR}destination"}


def test_featurized_entities_ordered_consistently():
    """Check that entities get ordered -- needed for consistent state representations.

    Previously, no ordering was applied to entities, but they were ordered implicitly
    due to how python sets work -- a set of all entity names was internally created,
    which was ordered by the hashes of the entity names. Now, entities are sorted alpha-
    betically. Since even sorting based on randomised hashing can produce alphabetical
    ordering once in a while, we here check with a large number of entities, pushing to
    ~0 the probability of correctly sorting the elements just by accident, without
    actually doing proper sorting.
    """
    # Create a sorted list of entity names from 'a' to 'z', and two randomly shuffled
    # copies.
    entity_names_sorted = [chr(i) for i in range(ord("a"), ord("z") + 1)]
    entity_names_shuffled1 = entity_names_sorted.copy()
    random.shuffle(entity_names_shuffled1)
    entity_names_shuffled2 = entity_names_sorted.copy()
    random.shuffle(entity_names_shuffled2)

    domain = Domain.from_dict(
        {KEY_INTENTS: ["inform"], KEY_ENTITIES: entity_names_shuffled1}
    )

    tracker = DialogueStateTracker.from_events(
        "story123",
        [
            UserUttered(
                text="hey there",
                intent={"name": "inform", "confidence": 1.0},
                entities=[
                    {"entity": e, "value": e.upper()} for e in entity_names_shuffled2
                ],
            )
        ],
    )
    state = domain.get_active_state(tracker)

    # Whatever order the entities were listed in, they should get sorted alphabetically
    # so the states' representations are consistent and entity-order-agnostic.
    assert state["user"]["entities"] == tuple(entity_names_sorted)


@pytest.mark.parametrize(
    "domain_as_dict",
    [
        # No slots
        {KEY_SLOTS: {}},
        # Valid slot mappings
        {
            KEY_SLOTS: {
                "slot_x": {
                    "type": "text",
                    "mappings": [{"type": "from_entity", "entity": "name"}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "slot_x": {
                    "type": "float",
                    "mappings": [{"type": "from_intent", "value": 5}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "slot_x": {
                    "type": "text",
                    "mappings": [{"type": "from_intent", "value": "some value"}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "slot_x": {
                    "type": "bool",
                    "mappings": [{"type": "from_intent", "value": False}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "slot_x": {
                    "type": "float",
                    "mappings": [{"type": "from_trigger_intent", "value": 5}],
                }
            },
            KEY_FORMS: {"some_form": {"required_slots": ["slot_x"]}},
        },
        {
            KEY_SLOTS: {
                "slot_x": {
                    "type": "text",
                    "mappings": [
                        {"type": "from_trigger_intent", "value": "some value"}
                    ],
                }
            },
            KEY_FORMS: {"some_form": {"required_slots": ["slot_x"]}},
        },
        {KEY_SLOTS: {"slot_x": {"type": "text", "mappings": [{"type": "from_text"}]}}},
    ],
)
def test_valid_slot_mappings(domain_as_dict: Dict[Text, Any]):
    Domain.from_dict(domain_as_dict)


@pytest.mark.parametrize(
    "domain_as_dict",
    [
        # Wrong type for forms
        {KEY_FORMS: []},
        # Wrong type for required_slots
        {KEY_FORMS: {"my_form": []}},
        {KEY_FORMS: {"my_form": 5}},
        # ignored_intent in forms, but no required_slots
        {KEY_FORMS: {"my_form": {"ignored_intents": ["greet"]}}},
    ],
)
def test_form_invalid_mappings(domain_as_dict: Dict[Text, Any]):
    with pytest.raises(InvalidDomain):
        Domain.from_dict(domain_as_dict)


def test_form_invalid_required_slots_raises():
    with pytest.raises(YamlValidationException):
        Domain.from_yaml(
            f"""
            version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
            entities:
            - some_entity
            forms:
              my_form:
                required_slots:
                  some_slot:
                  - type: from_entity
                    entity: some_entity
        """
        )


@pytest.mark.parametrize(
    "domain_as_dict",
    [
        # Unknown mapping
        {KEY_SLOTS: {"my_slot": {"type": "text", "mappings": [{"type": "test"}]}}},
        # Mappings with missing keys
        {
            KEY_SLOTS: {
                "my_slot": {
                    "type": "text",
                    "mappings": [{"type": "from_entity", "intent": "greet"}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "my_slot": {"type": "text", "mappings": [{"type": "from_intent"}]}
            }
        },
        {
            KEY_SLOTS: {
                "my_slot": {
                    "type": "text",
                    "mappings": [{"type": "from_intent", "value": None}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "my_slot": {
                    "type": "text",
                    "mappings": [{"type": "from_trigger_intent"}],
                }
            }
        },
        {
            KEY_SLOTS: {
                "my_slot": {
                    "type": "text",
                    "mappings": [{"type": "from_trigger_intent", "value": None}],
                }
            }
        },
    ],
)
def test_slot_invalid_mappings(domain_as_dict: Dict[Text, Any]):
    with pytest.raises(InvalidDomain):
        Domain.from_dict(domain_as_dict)


@pytest.mark.parametrize(
    "domain_yaml",
    [
        # Wrong type for slots
        (
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        slots:
          []
        """
        ),
        # Wrong type for slot names
        (
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        slots:
          some_slot: 5
        """
        ),
        (
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        slots:
          some_slot: []
        """
        ),
    ],
)
def test_invalid_slots_raises_yaml_exception(domain_yaml: Text):
    with pytest.raises(YamlValidationException):
        Domain.from_yaml(domain_yaml)


def test_slot_order_is_preserved():
    test_yaml = f"""version: '{LATEST_TRAINING_DATA_FORMAT_VERSION}'
session_config:
  session_expiration_time: 60
  carry_over_slots_to_new_session: true
slots:
  confirm:
    type: bool
    influence_conversation: false
    mappings:
    - type: custom
  previous_email:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  caller_id:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  email:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  incident_title:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  priority:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  problem_description:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  requested_slot:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  handoff_to:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
"""

    domain = Domain.from_yaml(test_yaml)
    assert domain.as_yaml() == test_yaml


def test_slot_order_is_preserved_when_merging():
    slot_1 = """
  b:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  a:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text"""

    test_yaml_1 = f"""
slots:{slot_1}
"""

    slot_2 = """
  d:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text
  c:
    type: text
    influence_conversation: false
    mappings:
    - type: from_text"""

    test_yaml_2 = f"""
slots:{slot_2}
"""

    test_yaml_merged = f"""version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
slots:{slot_2}{slot_1}
session_config:
  session_expiration_time: 60
  carry_over_slots_to_new_session: true
"""

    domain_1 = Domain.from_yaml(test_yaml_1)
    domain_2 = Domain.from_yaml(test_yaml_2)
    domain_merged = domain_1.merge(domain_2)

    assert domain_merged.as_yaml() == test_yaml_merged


def test_responses_text_multiline_is_preserved():
    test_yaml = f"""version: '{LATEST_TRAINING_DATA_FORMAT_VERSION}'
session_config:
  session_expiration_time: 60
  carry_over_slots_to_new_session: true
responses:
  utter_confirm:
  - text: |-
      First line
      Second line
      Third line
  - text: One more response
  utter_cancel:
  - text: First line
  - text: Second line
"""

    domain = Domain.from_yaml(test_yaml)
    assert domain.as_yaml() == test_yaml


def test_is_valid_domain_doesnt_raise_with_valid_domain(tmpdir: Path):
    domain_path = str(tmpdir / "domain.yml")
    rasa.shared.utils.io.write_text_file(
        """
       responses:
         utter_greet:
           - text: hey there! """,
        domain_path,
    )
    assert Domain.is_domain_file(domain_path)


def test_is_valid_domain_doesnt_raise_with_invalid_domain(tmpdir: Path):
    domain_path = str(tmpdir / "domain.yml")
    rasa.shared.utils.io.write_text_file(
        """
       invalid""",
        domain_path,
    )
    assert not Domain.is_domain_file(domain_path)


def test_is_valid_domain_doesnt_raise_with_invalid_yaml(tmpdir: Path):
    potential_domain_path = str(tmpdir / "domain.yml")
    rasa.shared.utils.io.write_text_file(
        """
       script:
        - echo "Latest SDK version is ${RASA_SDK_VERSION}""",
        potential_domain_path,
    )
    assert not Domain.is_domain_file(potential_domain_path)


def test_domain_with_empty_intent_mapping():
    # domain.yml with intent (intent_name) that has a `:` character
    # and nothing after it.
    test_yaml = """intents:
    - intent_name:"""

    with pytest.raises(InvalidDomain):
        Domain.from_yaml(test_yaml).as_dict()


def test_domain_with_empty_entity_mapping():
    # domain.yml with entity (entity_name) that has a `:` character
    # and nothing after it.
    test_yaml = """entities:
    - entity_name:"""

    with pytest.raises(InvalidDomain):
        Domain.from_yaml(test_yaml).as_dict()


def test_ignored_intents_slot_mappings_invalid_domain():
    domain_as_dict = {
        KEY_FORMS: {
            "my_form": {
                IGNORED_INTENTS: "some_not_intent",
                "slot_x": [
                    {
                        "type": "from_entity",
                        "entity": "name",
                        "not_intent": "other_not_intent",
                    }
                ],
            }
        }
    }
    with pytest.raises(InvalidDomain):
        Domain.from_dict(domain_as_dict)


def test_domain_count_conditional_response_variations():
    domain = Domain.from_file(
        path="data/test_domains/conditional_response_variations.yml"
    )
    count_conditional_responses = domain.count_conditional_response_variations()
    assert count_conditional_responses == 5


def test_domain_with_no_form_slots():
    domain = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        forms:
         contract_form:
          required_slots: []
        """
    )
    assert domain.required_slots_for_form("contract_form") == []


def test_domain_with_empty_required_slots():
    with pytest.raises(YamlException):
        Domain.from_yaml(
            f"""
            version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
            forms:
              contract_form:
            """
        )


def test_domain_invalid_yml_in_folder():
    """
    Check if invalid YAML files in a domain folder lead to the proper UserWarning
    """
    with pytest.warns(UserWarning, match="The file .* your file\\."):
        Domain.from_directory("data/test_domains/test_domain_from_directory/")


def test_invalid_domain_dir_with_duplicates(recwarn: WarningsRecorder):
    """
    Raises InvalidDomain if a domain is loaded from a directory with duplicated slots,
    responses and intents in domain files.
    """
    Domain.from_directory("data/test_domains/test_domain_with_duplicates/")

    error_message = (
        "The following duplicated intents have been found across multiple domain files: greet \n"
        "The following duplicated responses have been found across multiple domain files: "
        "utter_did_that_help, utter_greet \n"
        "The following duplicated slots have been found across multiple domain files: mood"
    )
    for warning in recwarn.list:
        # filter expected warnings
        if not any(
            type(warning.message) == warning_type
            and re.search(warning_message, str(warning.message))
            for warning_type, warning_message in EXPECTED_WARNINGS
        ):
            assert error_message == warning.message.args[0]


def test_domain_fingerprint_consistency_across_runs():
    domain_yaml = f"""
         version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
         intents:
         - greet
         - goodbye
         entities:
         - name
         slots:
           name:
             type: text
             mappings:
             - type: from_entity
               entity: name
         responses:
           utter_greet:
             - text: "Hi"
         forms:
          test_form:
            required_slots:
               - name
         actions:
         - action_test
    """
    domain1 = Domain.from_yaml(domain_yaml)
    domain2 = Domain.from_yaml(domain_yaml)

    f1 = domain1.fingerprint()
    f2 = domain2.fingerprint()
    assert f1 == f2


def test_domain_fingerprint_uniqueness():
    domain = Domain.from_yaml(
        f"""
         version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
         intents:
         - greet
         - goodbye
         actions:
         - action_test
         """
    )
    f1 = domain.fingerprint()

    domain_with_extra_intent = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        intents:
        - greet
        - goodbye
        - test
        actions:
        - action_test
        """
    )
    f2 = domain_with_extra_intent.fingerprint()
    assert f1 != f2

    domain_with_extra_action = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        intents:
        - greet
        - goodbye
        actions:
        - action_test
        - action_double_test
        """
    )
    f3 = domain_with_extra_action.fingerprint()
    assert f1 != f3

    domain_with_extra_responses = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        intents:
        - greet
        - goodbye
        responses:
          utter_greet:
           - text: "Hi!"
        actions:
        - action_test
        """
    )
    f4 = domain_with_extra_responses.fingerprint()
    assert f1 != f4


def test_domain_slots_for_entities_with_mapping_conditions_no_slot_set():
    domain = Domain.from_yaml(
        textwrap.dedent(
            f"""
            version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
            entities:
            - city
            slots:
              location:
                type: text
                influence_conversation: false
                mappings:
                - type: from_entity
                  entity: city
                  conditions:
                  - active_loop: booking_form
            forms:
              booking_form:
                required_slots:
                  - location
            """
        )
    )
    events = domain.slots_for_entities([{"entity": "city", "value": "Berlin"}])
    assert len(events) == 0


def test_domain_slots_for_entities_with_mapping_conditions_no_active_loop():
    domain = Domain.from_yaml(
        textwrap.dedent(
            f"""
            version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
            entities:
            - city
            slots:
              location:
                type: text
                influence_conversation: false
                mappings:
                - type: from_entity
                  entity: city
                  conditions:
                  - active_loop: null
            forms:
              booking_form:
                required_slots:
                  - location
            """
        )
    )
    events = domain.slots_for_entities([{"entity": "city", "value": "Berlin"}])
    assert events == [SlotSet("location", "Berlin")]


def test_domain_slots_for_entities_sets_valid_slot():
    domain = Domain.from_yaml(
        textwrap.dedent(
            f"""
            version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
            entities:
            - city
            slots:
              location:
                type: text
                influence_conversation: false
                mappings:
                - type: from_entity
                  entity: city
            """
        )
    )
    events = domain.slots_for_entities([{"entity": "city", "value": "Berlin"}])
    assert events == [SlotSet("location", "Berlin")]


def test_domain_slots_for_entities_sets_valid_list_slot():
    domain = Domain.from_yaml(
        textwrap.dedent(
            f"""
            version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
            entities:
            - topping
            slots:
              toppings:
                type: list
                influence_conversation: false
                mappings:
                - type: from_entity
                  entity: topping
            """
        )
    )
    events = domain.slots_for_entities(
        [
            {"entity": "topping", "value": "parmesan"},
            {"entity": "topping", "value": "prosciutto"},
        ]
    )
    assert events == [SlotSet("toppings", ["parmesan", "prosciutto"])]


def test_domain_slots_for_entities_with_entity_mapping_to_multiple_slots():
    domain = Domain.from_yaml(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        entities:
        - city
        slots:
          departure_city:
            type: text
            mappings:
            - type: from_entity
              entity: city
              role: from
          arrival_city:
            type: text
            mappings:
            - type: from_entity
              entity: city
              role: to
        """
    )
    events = domain.slots_for_entities(
        [
            {"entity": "city", "value": "London", "role": "from"},
            {"entity": "city", "value": "Berlin", "role": "to"},
        ]
    )
    assert events == [
        SlotSet("departure_city", "London"),
        SlotSet("arrival_city", "Berlin"),
    ]


def test_merge_domain_with_separate_session_config():
    domain_dir = "data/test_domains/test_domain_with_separate_session_config"
    domain = Domain.load(domain_dir)

    expected_session_expiration_time = 1

    assert (
        domain.session_config.session_expiration_time
        == expected_session_expiration_time
    )


@pytest.mark.parametrize(
    "actions, expected_result",
    [
        (
            [
                {"action_hello_world": {"send_domain": False}},
                {"action_say_something": {"send_domain": True}},
                {"action_calculate": {"send_domain": True}},
                "action_no_domain",
                "validate_my_form",
            ],
            ["action_say_something", "action_calculate", "validate_my_form"],
        ),
        (
            [
                "action_no_domain",
                "validate_my_form",
            ],
            ["validate_my_form"],
        ),
        (
            [
                {"action_hello_world": {"send_domain": False}},
                {"action_say_something": {"send_domain": False}},
                {"action_calculate": {"send_domain": False}},
            ],
            [],
        ),
        (
            [
                {"action_hello_world": {"send_domain": True}},
                {"action_say_something": {"send_domain": True}},
                {"action_calculate": {"send_domain": True}},
                "validate_my_form",
            ],
            [
                "action_hello_world",
                "action_say_something",
                "action_calculate",
                "validate_my_form",
            ],
        ),
        ([], []),
        (
            ["action_say_something", "action_calculate"],
            [],
        ),
    ],
)
def test_collect_actions_which_explicitly_need_domain(
    actions: List[Union[Dict[Text, Any], str]], expected_result: List[str]
):
    result = Domain._collect_actions_which_explicitly_need_domain(actions)

    # assert that two unordered lists have same elements
    assert sorted(result) == sorted(expected_result)


@pytest.mark.parametrize(
    "actions, expected_result",
    [
        (
            [
                {"action_hello_world": {"send_domain": False}},
                {"action_say_something": {"send_domain": True}},
                {"action_calculate": {"send_domain": True}},
                "action_no_domain",
                "validate_my_form",
            ],
            [
                "action_hello_world",
                "action_say_something",
                "action_calculate",
                "action_no_domain",
                "validate_my_form",
            ],
        )
    ],
)
def test_collect_actions(
    actions: List[Union[Dict[Text, Any], str]], expected_result: List[str]
):
    result = Domain._collect_action_names(actions)

    # assert that two unordered lists have same elements
    assert sorted(result) == sorted(expected_result)


@pytest.mark.parametrize(
    "content, expected_user_actions, expected_actions_which_explicitly_need_domain",
    [
        (
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        intents:
            - greet

        entities:
            - name

        responses:
            utter_greet:
                - text: hey there!

        actions:
          - action_hello: {{send_domain: True}}
          - action_bye: {{send_domain: True}}
          - action_no_domain
          - validate_my_form
          """,
            ["action_hello", "action_bye", "action_no_domain", "validate_my_form"],
            ["action_hello", "action_bye", "validate_my_form"],
        ),
        (
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        intents:
            - greet

        entities:
            - name

        responses:
            utter_greet:
                - text: hey there!

        actions:
          - action_hello
          - action_bye
          - action_no_domain
          - validate_my_form
          """,
            [
                "action_hello",
                "action_bye",
                "action_no_domain",
                "validate_my_form",
            ],
            [
                "validate_my_form",
            ],
        ),
        (
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        intents:
            - greet

        entities:
            - name

        responses:
            utter_greet:
                - text: hey there!

        actions:
          - action_hello
          - action_bye
          - action_no_domain
          """,
            ["action_hello", "action_bye", "action_no_domain"],
            [],
        ),
    ],
)
def test_domain_loads_actions_which_explicitly_need_domain(
    content: str,
    expected_user_actions: List[str],
    expected_actions_which_explicitly_need_domain: List[str],
):
    domain = Domain.from_yaml(content)
    assert domain._custom_actions == expected_user_actions
    assert (
        domain._actions_which_explicitly_need_domain
        == expected_actions_which_explicitly_need_domain
    )


def test_merge_yaml_domains_loads_actions_which_explicitly_need_domain():
    test_yaml_1 = textwrap.dedent(
        """
        actions:
          - action_hello
          - action_bye
          - action_send_domain: {send_domain: True}"""
    )

    test_yaml_2 = textwrap.dedent(
        """
        actions:
          - action_find_restaurants:
                send_domain: True"""
    )

    domain_1 = Domain.from_yaml(test_yaml_1)
    domain_2 = Domain.from_yaml(test_yaml_2)

    domain = domain_1.merge(domain_2)

    # single attribute should be taken from domain_1
    expected_actions = [
        "action_hello",
        "action_bye",
        "action_send_domain",
        "action_find_restaurants",
    ]
    expected_actions_that_need_domain = [
        "action_send_domain",
        "action_find_restaurants",
    ]
    assert sorted(domain._custom_actions) == sorted(expected_actions)
    assert sorted(domain._actions_which_explicitly_need_domain) == sorted(
        expected_actions_that_need_domain
    )


@pytest.mark.parametrize(
    "domain_yaml, expected",
    [
        (
            """
            responses:
                utter_greet:
                - text: hey there!
                  id: '1233'
                - text: hey ho!
                  id: '1234'
            """,
            {
                "utter_greet": [
                    {
                        "text": "hey there!",
                        "id": "1233",
                    },
                    {
                        "text": "hey ho!",
                        "id": "1234",
                    },
                ],
            },
        ),
        (
            """
            responses:
                utter_greet:
                - text: hey there!
                - text: hey ho!
                  id: '1234'
            """,
            {
                "utter_greet": [
                    {
                        "text": "hey there!",
                    },
                    {
                        "text": "hey ho!",
                        "id": "1234",
                    },
                ],
            },
        ),
        (
            """
            responses:
                utter_greet:
                - text: hey there!
                - text: hey ho!
            """,
            {
                "utter_greet": [
                    {
                        "text": "hey there!",
                    },
                    {
                        "text": "hey ho!",
                    },
                ],
            },
        ),
    ],
)
def test_domain_responses_with_ids_are_loaded(domain_yaml, expected) -> None:
    domain = Domain.from_yaml(domain_yaml)
    assert domain.responses == expected