rasahq--rasa/tests/core/test_test.py

import shutil
import textwrap
from pathlib import Path
from typing import Text, Optional, Dict, Any, List, Callable, Coroutine
import pytest
import rasa.core.test
import rasa.shared.utils.io
from rasa.core.policies.ensemble import DefaultPolicyPredictionEnsemble
from rasa.core.policies.policy import PolicyPrediction
from rasa.shared.constants import LATEST_TRAINING_DATA_FORMAT_VERSION
from rasa.shared.core.events import UserUttered
from _pytest.monkeypatch import MonkeyPatch
from _pytest.capture import CaptureFixture
from rasa.core.agent import Agent, load_agent
from rasa.utils.tensorflow.constants import (
    QUERY_INTENT_KEY,
    NAME,
    THRESHOLD_KEY,
    SEVERITY_KEY,
    SCORE_KEY,
)
from rasa.core.constants import STORIES_WITH_WARNINGS_FILE
from rasa.shared.core.constants import ACTION_UNLIKELY_INTENT_NAME
from rasa.shared.core.trackers import DialogueStateTracker
from rasa.shared.core.domain import Domain

from rasa.core.policies.rule_policy import RulePolicy
from rasa.shared.core.domain import State
from rasa.core.policies.policy import SupportedData
from rasa.shared.utils.io import read_file, read_yaml


def _probabilities_with_action_unlikely_intent_for(
    intent_names: List[Text],
    metadata_for_intent: Optional[Dict[Text, Dict[Text, Any]]] = None,
) -> Callable[
    [DefaultPolicyPredictionEnsemble, DialogueStateTracker, Domain, Any],
    PolicyPrediction,
]:
    _original = DefaultPolicyPredictionEnsemble.combine_predictions_from_kwargs

    def combine_predictions_from_kwargs(
        self, tracker: DialogueStateTracker, domain: Domain, **kwargs: Any
    ) -> PolicyPrediction:
        latest_event = tracker.events[-1]
        if (
            isinstance(latest_event, UserUttered)
            and latest_event.parse_data["intent"]["name"] in intent_names
        ):
            intent_name = latest_event.parse_data["intent"]["name"]
            # Here we return `action_unlikely_intent` if the name of the
            # latest intent is present in `intent_names`. Accompanying
            # metadata is fetched from `metadata_for_intent` if it is present.
            # We need to do it because every time the tests are run,
            # training will result in different model weights which might
            # result in different predictions of `action_unlikely_intent`.
            # Because we're not testing `UnexpecTEDIntentPolicy`,
            # here we simply trigger it by
            # predicting `action_unlikely_intent` in a specified moment
            # to make the tests deterministic.
            return PolicyPrediction.for_action_name(
                domain,
                ACTION_UNLIKELY_INTENT_NAME,
                action_metadata=metadata_for_intent.get(intent_name)
                if metadata_for_intent
                else None,
            )

        return _original(self, tracker, domain, **kwargs)

    return combine_predictions_from_kwargs


def _custom_prediction_states_for_rules(
    ignore_action_unlikely_intent: bool = False,
) -> Callable[[RulePolicy, DialogueStateTracker, Domain, bool], List[State],]:
    """Creates prediction states for `RulePolicy`.

    `RulePolicy` does not ignore `action_unlikely_intent` in reality.
    We use this helper method to monkey patch it for tests so that we can
    test `rasa test`'s behaviour when `action_unlikely_intent` is predicted.

    Args:
        ignore_action_unlikely_intent: Whether to ignore `action_unlikely_intent`.

    Returns:
        Monkey-patched method to create prediction states.
    """

    def _prediction_states(
        self: RulePolicy,
        tracker: DialogueStateTracker,
        domain: Domain,
        use_text_for_last_user_input: bool = False,
        rule_only_data: Optional[Dict[Text, Any]] = None,
    ) -> List[State]:
        return self.featurizer.prediction_states(
            [tracker],
            domain,
            use_text_for_last_user_input=use_text_for_last_user_input,
            ignore_rule_only_turns=self.supported_data() == SupportedData.ML_DATA,
            rule_only_data=rule_only_data,
            ignore_action_unlikely_intent=ignore_action_unlikely_intent,
        )[0]

    return _prediction_states


# FIXME: these tests take too long to run in the CI, disabling them for now
@pytest.mark.skip_on_ci
async def test_testing_warns_if_action_unknown(
    capsys: CaptureFixture,
    e2e_bot_agent: Agent,
    e2e_bot_test_stories_with_unknown_bot_utterances: Path,
    tmp_path: Path,
):
    await rasa.core.test.test(
        e2e_bot_test_stories_with_unknown_bot_utterances,
        e2e_bot_agent,
        out_directory=str(tmp_path),
    )
    output = capsys.readouterr().out
    assert "Test story" in output
    assert "contains the bot utterance" in output
    assert "which is not part of the training data / domain" in output


async def test_testing_with_utilizing_retrieval_intents(
    response_selector_agent: Agent, response_selector_test_stories: Path, tmp_path: Path
):
    result = await rasa.core.test.test(
        stories=response_selector_test_stories,
        agent=response_selector_agent,
        e2e=True,
        out_directory=str(tmp_path),
        disable_plotting=True,
        warnings=False,
    )
    failed_stories_path = tmp_path / "failed_test_stories.yml"
    failed_stories = read_yaml(read_file(failed_stories_path, "utf-8"))
    # check that the intent is shown correctly in the failed test stories file
    target_intents = {
        "test 0": "chitchat/ask_name",
        "test 1": "chitchat/ask_name",
        "test 2": "chitchat",
        "test 3": "chitchat",
    }
    for story in failed_stories["stories"]:
        test_name = story["story"].split("-")[0].strip()
        assert story["steps"][0]["intent"] == target_intents[test_name]
    # check that retrieval intent for actions is retrieved correctly
    # and only when it's needed.
    target_actions = {
        "utter_chitchat": "utter_chitchat",
        "utter_chitchat/ask_name": "utter_chitchat/ask_name",
        "utter_chitchat/ask_weather": "utter_chitchat/ask_name",
        "utter_goodbye": "utter_chitchat/ask_name",
    }
    predicted_actions = result["actions"][::2]
    for predicted_action in predicted_actions:
        assert (
            target_actions[predicted_action["action"]] == predicted_action["predicted"]
        )


async def test_testing_does_not_warn_if_intent_in_domain(
    default_agent: Agent, stories_path: Text, tmp_path: Path
):
    with pytest.warns(UserWarning) as record:
        await rasa.core.test.test(
            Path(stories_path), default_agent, out_directory=str(tmp_path)
        )

    assert not any("Found intent" in r.message.args[0] for r in record)
    assert all(
        "in stories which is not part of the domain" not in r.message.args[0]
        for r in record
    )


async def test_testing_valid_with_non_e2e_core_model(core_agent: Agent, tmp_path: Path):
    result = await rasa.core.test.test(
        "data/test_yaml_stories/test_stories_entity_annotations.yml",
        core_agent,
        out_directory=str(tmp_path),
    )
    assert "report" in result.keys()


@pytest.fixture()
async def _train_rule_based_agent(
    moodbot_domain: Domain,
    tmp_path: Path,
    trained_async: Callable,
    monkeypatch: MonkeyPatch,
    moodbot_domain_path: Path,
) -> Callable[[Path, bool], Coroutine]:

    # We need `RulePolicy` to predict the correct actions
    # in a particular conversation context as seen during training.
    # Since it can get affected by `action_unlikely_intent` being triggered in
    # some cases. We monkey-patch the method which creates
    # prediction states to ignore `action_unlikely_intent`s if needed.

    async def inner(file_name: Path, ignore_action_unlikely_intent: bool) -> Agent:
        config = textwrap.dedent(
            f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        assistant_id: placeholder_default
        pipeline: []
        policies:
        - name: RulePolicy
          restrict_rules: false
        """
        )
        config_path = tmp_path / "config.yml"
        rasa.shared.utils.io.write_text_file(config, config_path)

        rule_file = tmp_path / "rules.yml"
        shutil.copy2(file_name, rule_file)
        training_data = rule_file.read_text()
        training_data_for_rules = training_data.replace("stories:", "rules:")
        training_data_for_rules = training_data_for_rules.replace("story:", "rule:")
        rule_file.write_text(training_data_for_rules)

        model_path = await trained_async(
            moodbot_domain_path, str(config_path), str(rule_file)
        )

        monkeypatch.setattr(
            RulePolicy,
            "_prediction_states",
            _custom_prediction_states_for_rules(ignore_action_unlikely_intent),
        )

        return await load_agent(model_path)

    return inner


async def test_action_unlikely_intent_warning(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
):
    monkeypatch.setattr(
        DefaultPolicyPredictionEnsemble,
        "combine_predictions_from_kwargs",
        _probabilities_with_action_unlikely_intent_for(["mood_unhappy"]),
    )

    file_name = tmp_path / "test_action_unlikely_intent_1.yml"
    file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: unlikely path
            steps:
              - user: |
                  very terrible
                intent: mood_unhappy
              - action: utter_cheer_up
              - action: utter_did_that_help
              - intent: affirm
              - action: utter_happy
        """
    )

    # We train on the above story so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(file_name, True)

    result = await rasa.core.test.test(
        str(file_name),
        agent,
        out_directory=str(tmp_path),
        fail_on_prediction_errors=True,
    )
    assert "report" in result.keys()
    assert result["report"]["conversation_accuracy"]["correct"] == 1
    assert result["report"]["conversation_accuracy"]["with_warnings"] == 1

    # Ensure that the story with warning is correctly formatted
    with open(str(tmp_path / "stories_with_warnings.yml"), "r") as f:
        content = f.read()
        assert f"# predicted: {ACTION_UNLIKELY_INTENT_NAME}" in content


async def test_action_unlikely_intent_correctly_predicted(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
):
    monkeypatch.setattr(
        DefaultPolicyPredictionEnsemble,
        "combine_predictions_from_kwargs",
        _probabilities_with_action_unlikely_intent_for(["mood_unhappy"]),
    )

    file_name = tmp_path / "test_action_unlikely_intent_2.yml"
    file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: unlikely path (with action_unlikely_intent)
            steps:
              - user: |
                  very terrible
                intent: mood_unhappy
              - action: action_unlikely_intent
              - action: utter_cheer_up
              - action: utter_did_that_help
              - intent: affirm
              - action: utter_happy
        """
    )

    # We train on the above story so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(file_name, False)

    result = await rasa.core.test.test(
        str(file_name),
        agent,
        out_directory=str(tmp_path),
        fail_on_prediction_errors=True,
    )
    assert "report" in result.keys()
    assert result["report"]["conversation_accuracy"]["correct"] == 1
    assert result["report"]["conversation_accuracy"]["with_warnings"] == 0


async def test_wrong_action_after_action_unlikely_intent(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
):
    monkeypatch.setattr(
        DefaultPolicyPredictionEnsemble,
        "combine_predictions_from_kwargs",
        _probabilities_with_action_unlikely_intent_for(["greet", "mood_great"]),
    )

    test_file_name = tmp_path / "test.yml"
    test_file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: happy path
            steps:
              - user: |
                  hello there!
                intent: greet
              - action: utter_greet
              - user: |
                  amazing
                intent: mood_great
              - action: utter_happy
        """
    )

    train_file_name = tmp_path / "train.yml"
    train_file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: happy path
            steps:
              - user: |
                  hello there!
                intent: greet
              - action: utter_happy
              - user: |
                  amazing
                intent: mood_great
              - action: utter_goodbye
        """
    )

    # We train on the above story so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(train_file_name, True)

    result = await rasa.core.test.test(
        str(test_file_name),
        agent,
        out_directory=str(tmp_path),
        fail_on_prediction_errors=False,
    )
    assert "report" in result.keys()
    assert result["report"]["conversation_accuracy"]["correct"] == 0
    assert result["report"]["conversation_accuracy"]["with_warnings"] == 0
    assert result["report"]["conversation_accuracy"]["total"] == 1

    # Ensure that the failed story is correctly formatted
    with open(str(tmp_path / "failed_test_stories.yml"), "r") as f:
        content = f.read()
        assert (
            f"# predicted: utter_happy after {ACTION_UNLIKELY_INTENT_NAME}" in content
        )
        assert (
            f"# predicted: action_default_fallback after {ACTION_UNLIKELY_INTENT_NAME}"
            in content
        )


async def test_action_unlikely_intent_not_found(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
):
    test_file_name = tmp_path / "test_action_unlikely_intent_complete.yml"
    test_file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: happy path
            steps:
              - user: |
                  hello there!
                intent: greet
              - action: action_unlikely_intent
              - action: utter_greet
              - user: |
                  amazing
                intent: mood_great
              - action: utter_happy
        """
    )

    train_file_name = tmp_path / "train_without_action_unlikely_intent.yml"
    train_file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: happy path
            steps:
              - user: |
                  hello there!
                intent: greet
              - action: utter_greet
              - user: |
                  amazing
                intent: mood_great
              - action: utter_happy
        """
    )

    # We train on the above story so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(train_file_name, False)

    result = await rasa.core.test.test(
        str(test_file_name), agent, out_directory=str(tmp_path)
    )
    assert "report" in result.keys()
    assert result["report"]["conversation_accuracy"]["correct"] == 0
    assert result["report"]["conversation_accuracy"]["with_warnings"] == 0
    assert result["report"]["conversation_accuracy"]["total"] == 1

    # Ensure that the failed story is correctly formatted
    with open(str(tmp_path / "failed_test_stories.yml"), "r") as f:
        content = f.read()
        assert "# predicted: utter_greet" in content


async def test_action_unlikely_intent_warning_and_story_error(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
):
    monkeypatch.setattr(
        DefaultPolicyPredictionEnsemble,
        "combine_predictions_from_kwargs",
        _probabilities_with_action_unlikely_intent_for(["greet"]),
    )

    test_file_name = tmp_path / "test.yml"
    test_file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: happy path
            steps:
              - user: |
                  hello there!
                intent: greet
              - action: utter_greet
              - user: |
                  amazing
                intent: mood_great
              - action: utter_happy
        """
    )

    train_file_name = tmp_path / "train.yml"
    train_file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: happy path
            steps:
              - user: |
                  hello there!
                intent: greet
              - action: utter_greet
              - user: |
                  amazing
                intent: mood_great
              - action: utter_goodbye
        """
    )

    # We train on the above story so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(train_file_name, True)

    result = await rasa.core.test.test(
        str(test_file_name), agent, out_directory=str(tmp_path)
    )
    assert "report" in result.keys()
    assert result["report"]["conversation_accuracy"]["correct"] == 0
    assert result["report"]["conversation_accuracy"]["with_warnings"] == 0
    assert result["report"]["conversation_accuracy"]["total"] == 1

    # Ensure that the failed story is correctly formatted
    with open(str(tmp_path / "failed_test_stories.yml"), "r") as f:
        content = f.read()
        assert f"# predicted: {ACTION_UNLIKELY_INTENT_NAME}" in content
        assert "# predicted: utter_goodbye" in content


async def test_fail_on_prediction_errors(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
):
    monkeypatch.setattr(
        DefaultPolicyPredictionEnsemble,
        "combine_predictions_from_kwargs",
        _probabilities_with_action_unlikely_intent_for(["mood_unhappy"]),
    )

    file_name = tmp_path / "test_action_unlikely_intent_2.yml"
    file_name.write_text(
        f"""
        version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
        stories:
          - story: unlikely path (with action_unlikely_intent)
            steps:
              - user: |
                  very terrible
                intent: mood_unhappy
              - action: utter_cheer_up
              - action: action_unlikely_intent
              - action: utter_did_that_help
              - intent: affirm
              - action: utter_happy
        """
    )

    # We train on the above story so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(file_name, False)

    with pytest.raises(rasa.core.test.WrongPredictionException):
        await rasa.core.test.test(
            str(file_name),
            agent,
            out_directory=str(tmp_path),
            fail_on_prediction_errors=True,
        )


@pytest.mark.parametrize(
    "metadata_for_intents, story_order",
    [
        (
            {
                "mood_unhappy": {
                    QUERY_INTENT_KEY: {
                        NAME: "mood_unhappy",
                        SEVERITY_KEY: 2.0,
                        THRESHOLD_KEY: 0.0,
                        SCORE_KEY: -2.0,
                    }
                },
                "mood_great": {
                    QUERY_INTENT_KEY: {
                        NAME: "mood_great",
                        SEVERITY_KEY: 3.0,
                        THRESHOLD_KEY: 0.2,
                        SCORE_KEY: -1.0,
                    }
                },
                "affirm": {
                    QUERY_INTENT_KEY: {
                        NAME: "affirm",
                        SEVERITY_KEY: 4.2,
                        THRESHOLD_KEY: 0.2,
                        SCORE_KEY: -4.0,
                    }
                },
            },
            ["path 2", "path 1"],
        ),
        (
            {
                "mood_unhappy": {
                    QUERY_INTENT_KEY: {
                        NAME: "mood_unhappy",
                        SEVERITY_KEY: 2.0,
                        THRESHOLD_KEY: 0.0,
                        SCORE_KEY: -2.0,
                    }
                },
                "mood_great": {
                    QUERY_INTENT_KEY: {
                        NAME: "mood_great",
                        SEVERITY_KEY: 5.0,
                        THRESHOLD_KEY: 0.2,
                        SCORE_KEY: -1.0,
                    }
                },
                "affirm": {
                    QUERY_INTENT_KEY: {
                        NAME: "affirm",
                        SEVERITY_KEY: 4.2,
                        THRESHOLD_KEY: 0.2,
                        SCORE_KEY: -4.0,
                    }
                },
            },
            ["path 1", "path 2"],
        ),
    ],
)
async def test_multiple_warnings_sorted_on_severity(
    monkeypatch: MonkeyPatch,
    tmp_path: Path,
    _train_rule_based_agent: Callable[[Path, bool], Coroutine],
    metadata_for_intents: Dict,
    story_order: List[Text],
):
    monkeypatch.setattr(
        DefaultPolicyPredictionEnsemble,
        "combine_predictions_from_kwargs",
        _probabilities_with_action_unlikely_intent_for(
            list(metadata_for_intents.keys()), metadata_for_intents
        ),
    )

    test_story_path = (
        "data/test_yaml_stories/test_multiple_action_unlikely_intent_warnings.yml"
    )

    # We train on the stories as it is so that RulePolicy can memorize
    # it and we don't have to worry about other actions being
    # predicted correctly.
    agent = await _train_rule_based_agent(Path(test_story_path), True)

    await rasa.core.test.test(
        test_story_path,
        agent,
        out_directory=str(tmp_path),
        fail_on_prediction_errors=True,
    )

    warnings_file = tmp_path / STORIES_WITH_WARNINGS_FILE
    warnings_data = rasa.shared.utils.io.read_yaml_file(warnings_file)

    for index, story_name in enumerate(story_order):
        assert warnings_data["stories"][index]["story"].startswith(story_name)