Repository: RasaHQ/rasa
Pull request: Integrate nlu http interpreter.
**Proposed changes**:
Closes #9765


Problem statement:
Revamp: Adapt RasaNLUHttpInterpreter for the graph.
### Description
In the old architecture the nlu interpreter and the core policy predictions were separate. 
So the interpreter could just be replaced by the `RasaNLUHttpInterpreter` as needed.
Now that the message parsing and policy prediction happen within the graph we need to implement the `RasaNLUHttpInterpreter` as a graph component. We also need to be able to optionally swap to using it with an already trained model (?).

### Definition of Done
- [ ] We can use an http nlu interpreter when predicting with the graph.

Hints from issue discussion:

Commit message: Integrate nlu http interpreter.

Patch to review:
diff --git a/rasa/cli/test.py b/rasa/cli/test.py
index 47c65b07061d..824826d69b0a 100644
--- a/rasa/cli/test.py
+++ b/rasa/cli/test.py
@@ -92,7 +92,7 @@ def _print_core_test_execution_info(args: argparse.Namespace) -> None:
         )
 
 
-def run_core_test(args: argparse.Namespace) -> None:
+async def run_core_test_async(args: argparse.Namespace) -> None:
     """Run core tests."""
     from rasa.model_testing import (
         test_core_models_in_directory,
@@ -130,7 +130,7 @@ def run_core_test(args: argparse.Namespace) -> None:
                 args.model, stories, output, use_conversation_test_files=args.e2e
             )
         else:
-            test_core(
+            await test_core(
                 model=model_path,
                 stories=stories,
                 output=output,
@@ -139,7 +139,7 @@ def run_core_test(args: argparse.Namespace) -> None:
             )
 
     else:
-        test_core_models(
+        await test_core_models(
             args.model, stories, output, use_conversation_test_files=args.e2e
         )
 
@@ -212,7 +212,7 @@ async def run_nlu_test_async(
                     f"Ignoring file '{file}' as it is not a valid config file."
                 )
                 continue
-        compare_nlu_models(
+        await compare_nlu_models(
             configs=config_files,
             test_data=nlu_data,
             output=output,
@@ -227,13 +227,13 @@ async def run_nlu_test_async(
         config_importer = TrainingDataImporter.load_from_dict(config_path=config)
 
         config_dict = config_importer.get_config()
-        perform_nlu_cross_validation(config_dict, nlu_data, output, all_args)
+        await perform_nlu_cross_validation(config_dict, nlu_data, output, all_args)
     else:
         model_path = rasa.cli.utils.get_validated_path(
             models_path, "model", DEFAULT_MODELS_PATH
         )
 
-        test_nlu(model_path, data_path, output, all_args)
+        await test_nlu(model_path, data_path, output, all_args)
 
 
 def run_nlu_test(args: argparse.Namespace) -> None:
@@ -257,6 +257,15 @@ def run_nlu_test(args: argparse.Namespace) -> None:
     )
 
 
+def run_core_test(args: argparse.Namespace) -> None:
+    """Runs NLU tests.
+
+    Args:
+        args: the parsed CLI arguments for 'rasa test nlu'.
+    """
+    rasa.utils.common.run_in_loop(run_core_test_async(args))
+
+
 def test(args: argparse.Namespace) -> None:
     """Run end-to-end tests."""
     setattr(args, "e2e", True)
diff --git a/rasa/core/agent.py b/rasa/core/agent.py
index fd89bd78ab79..fadf9e8e6721 100644
--- a/rasa/core/agent.py
+++ b/rasa/core/agent.py
@@ -20,6 +20,7 @@
 from rasa.core import jobs
 from rasa.core.channels.channel import OutputChannel, UserMessage
 from rasa.core.constants import DEFAULT_REQUEST_TIMEOUT
+from rasa.core.http_interpreter import RasaNLUHttpInterpreter
 from rasa.shared.core.domain import Domain
 from rasa.core.exceptions import AgentNotReady
 from rasa.shared.constants import DEFAULT_SENDER_ID
@@ -233,6 +234,7 @@ async def load_agent(
     lock_store = None
     generator = None
     action_endpoint = None
+    http_interpreter = None
 
     if endpoints:
         broker = await EventBroker.create(endpoints.event_broker, loop=loop)
@@ -243,6 +245,8 @@ async def load_agent(
         generator = endpoints.nlg
         action_endpoint = endpoints.action
         model_server = endpoints.model if endpoints.model else model_server
+        if endpoints.nlu:
+            http_interpreter = RasaNLUHttpInterpreter(endpoints.nlu)
 
     agent = Agent(
         generator=generator,
@@ -251,6 +255,7 @@ async def load_agent(
         action_endpoint=action_endpoint,
         model_server=model_server,
         remote_storage=remote_storage,
+        http_interpreter=http_interpreter,
     )
 
     try:
@@ -280,11 +285,11 @@ async def load_agent(
 
 
 class Agent:
-    """The Agent class provides a convenient interface for the most important
-    Rasa functionality.
+    """The Agent class provides an interface for the most important Rasa functionality.
 
     This includes training, handling messages, loading a dialogue model,
-    getting the next action, and handling a channel."""
+    getting the next action, and handling a channel.
+    """
 
     def __init__(
         self,
@@ -296,6 +301,7 @@ def __init__(
         fingerprint: Optional[Text] = None,
         model_server: Optional[EndpointConfig] = None,
         remote_storage: Optional[Text] = None,
+        http_interpreter: Optional[RasaNLUHttpInterpreter] = None,
     ):
         """Initializes an `Agent`."""
         self.domain = domain
@@ -305,6 +311,7 @@ def __init__(
         self.tracker_store = self._create_tracker_store(tracker_store, self.domain)
         self.lock_store = self._create_lock_store(lock_store)
         self.action_endpoint = action_endpoint
+        self.http_interpreter = http_interpreter
 
         self._set_fingerprint(fingerprint)
         self.model_server = model_server
@@ -322,6 +329,7 @@ def load(
         fingerprint: Optional[Text] = None,
         model_server: Optional[EndpointConfig] = None,
         remote_storage: Optional[Text] = None,
+        http_interpreter: Optional[RasaNLUHttpInterpreter] = None,
     ) -> Agent:
         """Constructs a new agent and loads the processer and model."""
         agent = Agent(
@@ -333,6 +341,7 @@ def load(
             fingerprint=fingerprint,
             model_server=model_server,
             remote_storage=remote_storage,
+            http_interpreter=http_interpreter,
         )
         agent.load_model(model_path=model_path, fingerprint=fingerprint)
         return agent
@@ -347,6 +356,7 @@ def load_model(
             lock_store=self.lock_store,
             action_endpoint=self.action_endpoint,
             generator=self.nlg,
+            http_interpreter=self.http_interpreter,
         )
         self.domain = self.processor.domain
 
@@ -371,7 +381,7 @@ def is_ready(self) -> bool:
         """Check if all necessary components are instantiated to use agent."""
         return self.tracker_store is not None and self.processor is not None
 
-    def parse_message(self, message_data: Text) -> Dict[Text, Any]:
+    async def parse_message(self, message_data: Text) -> Dict[Text, Any]:
         """Handles message text and intent payload input messages.
 
         The return value of this function is parsed_data.
@@ -400,7 +410,7 @@ def parse_message(self, message_data: Text) -> Dict[Text, Any]:
                 "processor and a tracker store."
             )
         message = UserMessage(message_data)
-        return self.processor.parse_message(message)
+        return await self.processor.parse_message(message)
 
     async def handle_message(
         self, message: UserMessage,
diff --git a/rasa/core/http_interpreter.py b/rasa/core/http_interpreter.py
index 0861da2731ac..57c42a498cdd 100644
--- a/rasa/core/http_interpreter.py
+++ b/rasa/core/http_interpreter.py
@@ -5,39 +5,34 @@
 from typing import Text, Dict, Any, Optional
 
 from rasa.core import constants
-from rasa.shared.core.trackers import DialogueStateTracker
+from rasa.core.channels import UserMessage
 from rasa.shared.nlu.constants import INTENT_NAME_KEY
 from rasa.utils.endpoints import EndpointConfig
 
 logger = logging.getLogger(__name__)
 
 
-# TODO: This needs to be converted into a graph component
 class RasaNLUHttpInterpreter:
+    """Allows for an HTTP endpoint to be used to parse messages."""
+
     def __init__(self, endpoint_config: Optional[EndpointConfig] = None) -> None:
+        """Initializes a `RasaNLUHttpInterpreter`"""
         if endpoint_config:
             self.endpoint_config = endpoint_config
         else:
             self.endpoint_config = EndpointConfig(constants.DEFAULT_SERVER_URL)
 
-    async def parse(
-        self,
-        text: Text,
-        message_id: Optional[Text] = None,
-        tracker: Optional[DialogueStateTracker] = None,
-        metadata: Optional[Dict] = None,
-    ) -> Dict[Text, Any]:
+    async def parse(self, message: UserMessage,) -> Dict[Text, Any]:
         """Parse a text message.
 
         Return a default value if the parsing of the text failed."""
-
         default_return = {
             "intent": {INTENT_NAME_KEY: "", "confidence": 0.0},
             "entities": [],
             "text": "",
         }
-        result = await self._rasa_http_parse(text, message_id)
 
+        result = await self._rasa_http_parse(message.text, message.sender_id)
         return result if result is not None else default_return
 
     async def _rasa_http_parse(
diff --git a/rasa/core/processor.py b/rasa/core/processor.py
index dc18ff4de6e5..a06315d34b25 100644
--- a/rasa/core/processor.py
+++ b/rasa/core/processor.py
@@ -7,6 +7,7 @@
 from types import LambdaType
 from typing import Any, Dict, List, Optional, Text, Tuple, Union
 
+from rasa.core.http_interpreter import RasaNLUHttpInterpreter
 from rasa.engine import loader
 from rasa.engine.constants import PLACEHOLDER_MESSAGE, PLACEHOLDER_TRACKER
 from rasa.engine.runner.dask import DaskGraphRunner
@@ -81,6 +82,7 @@ def __init__(
         action_endpoint: Optional[EndpointConfig] = None,
         max_number_of_predictions: int = MAX_NUMBER_OF_PREDICTIONS,
         on_circuit_break: Optional[LambdaType] = None,
+        http_interpreter: Optional[RasaNLUHttpInterpreter] = None,
     ) -> None:
         """Initializes a `MessageProcessor`."""
         self.nlg = generator
@@ -92,6 +94,7 @@ def __init__(
         self.model_metadata, self.graph_runner = self._load_model(model_path)
         self.model_path = Path(model_path)
         self.domain = self.model_metadata.domain
+        self.http_interpreter = http_interpreter
 
     @staticmethod
     def _load_model(model_path: Union[Text, Path]) -> Tuple[ModelMetadata, GraphRunner]:
@@ -340,7 +343,7 @@ async def log_message(
             message.sender_id, message.output_channel, message.metadata
         )
 
-        self._handle_message_with_tracker(message, tracker)
+        await self._handle_message_with_tracker(message, tracker)
 
         if should_save_tracker:
             self._save_tracker(tracker)
@@ -573,7 +576,34 @@ def _get_action(
             action_name, self.domain, self.action_endpoint
         )
 
-    def parse_message(
+    async def parse_message(
+        self, message: UserMessage, only_output_properties: bool = True
+    ) -> Dict[Text, Any]:
+        """Interprets the passed message.
+
+        Arguments:
+            message: Message to handle
+
+        Returns:
+            Parsed data extracted from the message.
+        """
+        if self.http_interpreter:
+            parse_data = await self.http_interpreter.parse(message)
+        else:
+            parse_data = self._parse_message_with_graph(message, only_output_properties)
+
+        logger.debug(
+            "Received user message '{}' with intent '{}' "
+            "and entities '{}'".format(
+                parse_data["text"], parse_data["intent"], parse_data["entities"]
+            )
+        )
+
+        self._check_for_unseen_features(parse_data)
+
+        return parse_data
+
+    def _parse_message_with_graph(
         self, message: UserMessage, only_output_properties: bool = True
     ) -> Dict[Text, Any]:
         """Interprets the passed message.
@@ -598,26 +628,16 @@ def parse_message(
         parse_data.update(
             parsed_message.as_dict(only_output_properties=only_output_properties)
         )
+        return parsed_message
 
-        logger.debug(
-            "Received user message '{}' with intent '{}' "
-            "and entities '{}'".format(
-                parse_data["text"], parse_data["intent"], parse_data["entities"]
-            )
-        )
-
-        self._check_for_unseen_features(parse_data)
-
-        return parse_data
-
-    def _handle_message_with_tracker(
+    async def _handle_message_with_tracker(
         self, message: UserMessage, tracker: DialogueStateTracker
     ) -> None:
 
         if message.parse_data:
             parse_data = message.parse_data
         else:
-            parse_data = self.parse_message(message)
+            parse_data = await self.parse_message(message)
 
         # don't ever directly mutate the tracker
         # - instead pass its events to log
diff --git a/rasa/core/test.py b/rasa/core/test.py
index 35331403e982..192ffd50cf42 100644
--- a/rasa/core/test.py
+++ b/rasa/core/test.py
@@ -581,7 +581,7 @@ def emulate_loop_rejection(partial_tracker: DialogueStateTracker) -> None:
     partial_tracker.update(ActionExecutionRejected(rejected_action_name))
 
 
-def _get_e2e_entity_evaluation_result(
+async def _get_e2e_entity_evaluation_result(
     processor: "MessageProcessor",
     tracker: DialogueStateTracker,
     prediction: PolicyPrediction,
@@ -597,7 +597,7 @@ def _get_e2e_entity_evaluation_result(
         entity_targets = previous_event.entities
         if entity_targets or entities_predicted_by_policies:
             text = previous_event.text
-            parsed_message = processor.parse_message(UserMessage(text=text))
+            parsed_message = await processor.parse_message(UserMessage(text=text))
             if parsed_message:
                 tokens = [
                     Token(text[start:end], start, end)
@@ -638,7 +638,7 @@ def _get_predicted_action_name(
     return predicted_action_name
 
 
-def _run_action_prediction(
+async def _run_action_prediction(
     processor: "MessageProcessor",
     partial_tracker: DialogueStateTracker,
     expected_action: Text,
@@ -648,7 +648,7 @@ def _run_action_prediction(
         action, partial_tracker, expected_action
     )
 
-    policy_entity_result = _get_e2e_entity_evaluation_result(
+    policy_entity_result = await _get_e2e_entity_evaluation_result(
         processor, partial_tracker, prediction
     )
     if (
@@ -675,7 +675,7 @@ def _run_action_prediction(
     return predicted_action, prediction, policy_entity_result
 
 
-def _collect_action_executed_predictions(
+async def _collect_action_executed_predictions(
     processor: "MessageProcessor",
     partial_tracker: DialogueStateTracker,
     event: ActionExecuted,
@@ -692,9 +692,11 @@ def _collect_action_executed_predictions(
     prev_action_unlikely_intent = False
 
     try:
-        predicted_action, prediction, policy_entity_result = _run_action_prediction(
-            processor, partial_tracker, expected_action
-        )
+        (
+            predicted_action,
+            prediction,
+            policy_entity_result,
+        ) = await _run_action_prediction(processor, partial_tracker, expected_action)
     except ActionLimitReached:
         prediction = PolicyPrediction([], policy_name=None)
         predicted_action = "circuit breaker tripped"
@@ -715,7 +717,11 @@ def _collect_action_executed_predictions(
         prev_action_unlikely_intent = True
 
         try:
-            predicted_action, prediction, policy_entity_result = _run_action_prediction(
+            (
+                predicted_action,
+                prediction,
+                policy_entity_result,
+            ) = await _run_action_prediction(
                 processor, partial_tracker, expected_action
             )
         except ActionLimitReached:
@@ -783,7 +789,7 @@ def _form_might_have_been_rejected(
     )
 
 
-def _predict_tracker_actions(
+async def _predict_tracker_actions(
     tracker: DialogueStateTracker,
     agent: "Agent",
     fail_on_prediction_errors: bool = False,
@@ -815,7 +821,7 @@ def _predict_tracker_actions(
                 action_executed_result,
                 prediction,
                 entity_result,
-            ) = _collect_action_executed_predictions(
+            ) = await _collect_action_executed_predictions(
                 processor, partial_tracker, event, fail_on_prediction_errors,
             )
             if entity_result:
@@ -841,7 +847,7 @@ def _predict_tracker_actions(
             # in YAML format containing a user message, or in Markdown format.
             # Leaving that as it is because Markdown is in legacy mode.
             else:
-                predicted = processor.parse_message(UserMessage(event.text))
+                predicted = await processor.parse_message(UserMessage(event.text))
 
             user_uttered_result = _collect_user_uttered_predictions(
                 event, predicted, partial_tracker, fail_on_prediction_errors
@@ -905,7 +911,7 @@ def _sort_trackers_with_severity_of_warning(
     return [tracker for (_, tracker) in sorted_trackers_with_severity]
 
 
-def _collect_story_predictions(
+async def _collect_story_predictions(
     completed_trackers: List["DialogueStateTracker"],
     agent: "Agent",
     fail_on_prediction_errors: bool = False,
@@ -933,7 +939,9 @@ def _collect_story_predictions(
             predicted_tracker,
             tracker_actions,
             tracker_entity_results,
-        ) = _predict_tracker_actions(tracker, agent, fail_on_prediction_errors, use_e2e)
+        ) = await _predict_tracker_actions(
+            tracker, agent, fail_on_prediction_errors, use_e2e
+        )
 
         entity_results.extend(tracker_entity_results)
 
@@ -1018,7 +1026,7 @@ def _log_stories(
             f.write(YAMLStoryWriter().dumps(steps))
 
 
-def test(
+async def test(
     stories: Text,
     agent: "Agent",
     max_stories: Optional[int] = None,
@@ -1054,7 +1062,7 @@ def test(
     generator = _create_data_generator(stories, agent, max_stories, e2e)
     completed_trackers = generator.generate_story_trackers()
 
-    story_evaluation, _, entity_results = _collect_story_predictions(
+    story_evaluation, _, entity_results = await _collect_story_predictions(
         completed_trackers, agent, fail_on_prediction_errors, use_e2e=e2e
     )
 
@@ -1202,7 +1210,7 @@ def _plot_story_evaluation(
     )
 
 
-def compare_models_in_dir(
+async def compare_models_in_dir(
     model_dir: Text,
     stories_file: Text,
     output: Text,
@@ -1229,7 +1237,7 @@ def compare_models_in_dir(
             # The model files are named like <config-name>PERCENTAGE_KEY<number>.tar.gz
             # Remove the percentage key and number from the name to get the config name
             config_name = os.path.basename(model).split(PERCENTAGE_KEY)[0]
-            number_of_correct_stories = _evaluate_core_model(
+            number_of_correct_stories = await _evaluate_core_model(
                 model,
                 stories_file,
                 use_conversation_test_files=use_conversation_test_files,
@@ -1244,7 +1252,7 @@ def compare_models_in_dir(
     )
 
 
-def compare_models(
+async def compare_models(
     models: List[Text],
     stories_file: Text,
     output: Text,
@@ -1262,7 +1270,7 @@ def compare_models(
     number_correct = defaultdict(list)
 
     for model in models:
-        number_of_correct_stories = _evaluate_core_model(
+        number_of_correct_stories = await _evaluate_core_model(
             model, stories_file, use_conversation_test_files=use_conversation_test_files
         )
         number_correct[os.path.basename(model)].append(number_of_correct_stories)
@@ -1272,7 +1280,7 @@ def compare_models(
     )
 
 
-def _evaluate_core_model(
+async def _evaluate_core_model(
     model: Text, stories_file: Text, use_conversation_test_files: bool = False
 ) -> int:
     from rasa.core.agent import Agent
@@ -1286,7 +1294,7 @@ def _evaluate_core_model(
     completed_trackers = generator.generate_story_trackers()
 
     # Entities are ignored here as we only compare number of correct stories.
-    story_eval_store, number_of_stories, _ = _collect_story_predictions(
+    story_eval_store, number_of_stories, _ = await _collect_story_predictions(
         completed_trackers, agent
     )
     failed_stories = story_eval_store.failed_stories
diff --git a/rasa/model_testing.py b/rasa/model_testing.py
index 7bdb71aeb56d..431e9521eb67 100644
--- a/rasa/model_testing.py
+++ b/rasa/model_testing.py
@@ -99,7 +99,7 @@ def _get_sanitized_model_directory(model_directory: Text) -> Text:
     return model_directory
 
 
-def test_core_models(
+async def test_core_models(
     models: List[Text],
     stories: Text,
     output: Text,
@@ -116,7 +116,7 @@ def test_core_models(
     """
     from rasa.core.test import compare_models
 
-    compare_models(
+    await compare_models(
         models,
         stories,
         output,
@@ -124,7 +124,7 @@ def test_core_models(
     )
 
 
-def test_core(
+async def test_core(
     model: Optional[Text] = None,
     stories: Optional[Text] = None,
     output: Text = DEFAULT_RESULTS_PATH,
@@ -176,7 +176,7 @@ def test_core(
         additional_arguments, core_test, ["stories", "agent", "e2e"]
     )
 
-    core_test(
+    await core_test(
         stories,
         _agent,
         e2e=use_conversation_test_files,
@@ -185,7 +185,7 @@ def test_core(
     )
 
 
-def test_nlu(
+async def test_nlu(
     model: Optional[Text],
     nlu_data: Optional[Text],
     output_directory: Text = DEFAULT_RESULTS_PATH,
@@ -212,7 +212,7 @@ def test_nlu(
             additional_arguments, run_evaluation, ["data_path", "model"]
         )
         _agent = Agent.load(model_path=model)
-        run_evaluation(
+        await run_evaluation(
             nlu_data, _agent.processor, output_directory=output_directory, **kwargs
         )
     else:
@@ -222,7 +222,7 @@ def test_nlu(
         )
 
 
-def compare_nlu_models(
+async def compare_nlu_models(
     configs: List[Text],
     test_data: TrainingData,
     output: Text,
@@ -247,7 +247,7 @@ def compare_nlu_models(
         model_name: [[] for _ in range(runs)] for model_name in model_names
     }
 
-    training_examples_per_run = compare_nlu(
+    training_examples_per_run = await compare_nlu(
         configs,
         test_data,
         exclusion_percentages,
@@ -283,7 +283,7 @@ def plot_nlu_results(output_directory: Text, number_of_examples: List[int]) -> N
     )
 
 
-def perform_nlu_cross_validation(
+async def perform_nlu_cross_validation(
     config: Dict[Text, Any],
     data: TrainingData,
     output: Text,
@@ -314,7 +314,7 @@ def perform_nlu_cross_validation(
         additional_arguments, cross_validate
     )
 
-    results, entity_results, response_selection_results = cross_validate(
+    results, entity_results, response_selection_results = await cross_validate(
         data, folds, config, output, **kwargs
     )
     logger.info(f"CV evaluation (n={folds})")
diff --git a/rasa/nlu/run.py b/rasa/nlu/run.py
index cb86a4b8db78..94cd3c58a17c 100644
--- a/rasa/nlu/run.py
+++ b/rasa/nlu/run.py
@@ -4,7 +4,7 @@
 from rasa.core.agent import Agent
 from rasa.shared.utils.cli import print_info, print_success
 from rasa.shared.utils.io import json_to_string
-
+import rasa.utils.common
 
 logger = logging.getLogger(__name__)
 
@@ -22,6 +22,6 @@ def run_cmdline(model_path: Text) -> None:
             print_info("Wrapping up command line chat...")
             break
 
-        result = agent.parse_message(message)
+        result = rasa.utils.common.run_in_loop(agent.parse_message(message))
 
         print(json_to_string(result))
diff --git a/rasa/nlu/test.py b/rasa/nlu/test.py
index bd7c9df5e290..b2c65a405ace 100644
--- a/rasa/nlu/test.py
+++ b/rasa/nlu/test.py
@@ -1226,7 +1226,7 @@ def align_all_entity_predictions(
     return aligned_predictions
 
 
-def get_eval_data(
+async def get_eval_data(
     processor: MessageProcessor, test_data: TrainingData
 ) -> Tuple[
     List[IntentEvaluationResult],
@@ -1261,7 +1261,7 @@ def get_eval_data(
     should_eval_entities = len(test_data.entity_examples) > 0
 
     for example in tqdm(test_data.nlu_examples):
-        result = processor.parse_message(
+        result = await processor.parse_message(
             UserMessage(text=example.get(TEXT)), only_output_properties=False
         )
         _remove_entities_of_extractors(result, PRETRAINED_EXTRACTORS)
@@ -1345,7 +1345,7 @@ def _remove_entities_of_extractors(
     nlu_parse_result[ENTITIES] = filtered_entities
 
 
-def run_evaluation(
+async def run_evaluation(
     data_path: Text,
     processor: MessageProcessor,
     output_directory: Optional[Text] = None,
@@ -1387,7 +1387,7 @@ def run_evaluation(
     if output_directory:
         rasa.shared.utils.io.create_directory(output_directory)
 
-    (intent_results, response_selection_results, entity_results) = get_eval_data(
+    (intent_results, response_selection_results, entity_results) = await get_eval_data(
         processor, test_data
     )
 
@@ -1464,7 +1464,7 @@ def generate_folds(
         )
 
 
-def combine_result(
+async def combine_result(
     intent_metrics: IntentMetrics,
     entity_metrics: EntityMetrics,
     response_selection_metrics: ResponseSelectionMetrics,
@@ -1501,7 +1501,7 @@ def combine_result(
         current_intent_results,
         current_entity_results,
         current_response_selection_results,
-    ) = compute_metrics(processor, data)
+    ) = await compute_metrics(processor, data)
 
     if intent_results is not None:
         intent_results += current_intent_results
@@ -1534,7 +1534,7 @@ def _contains_entity_labels(entity_results: List[EntityEvaluationResult]) -> boo
     return False
 
 
-def cross_validate(
+async def cross_validate(
     data: TrainingData,
     n_folds: int,
     nlu_config: Union[RasaNLUModelConfig, Text, Dict],
@@ -1598,7 +1598,7 @@ def cross_validate(
             processor = Agent.load(model_file).processor
 
             # calculate train accuracy
-            combine_result(
+            await combine_result(
                 intent_train_metrics,
                 entity_train_metrics,
                 response_selection_train_metrics,
@@ -1606,7 +1606,7 @@ def cross_validate(
                 train,
             )
             # calculate test accuracy
-            combine_result(
+            await combine_result(
                 intent_test_metrics,
                 entity_test_metrics,
                 response_selection_test_metrics,
@@ -1679,7 +1679,7 @@ def _targets_predictions_from(
     return zip(*[(getattr(r, target_key), getattr(r, prediction_key)) for r in results])
 
 
-def compute_metrics(
+async def compute_metrics(
     processor: MessageProcessor, training_data: TrainingData
 ) -> Tuple[
     IntentMetrics,
@@ -1698,7 +1698,7 @@ def compute_metrics(
 
     Returns: intent, response selection and entity metrics, and prediction results.
     """
-    intent_results, response_selection_results, entity_results = get_eval_data(
+    intent_results, response_selection_results, entity_results = await get_eval_data(
         processor, training_data
     )
 
@@ -1736,7 +1736,7 @@ def compute_metrics(
     )
 
 
-def compare_nlu(
+async def compare_nlu(
     configs: List[Text],
     data: TrainingData,
     exclusion_percentages: List[int],
@@ -1821,7 +1821,7 @@ def compare_nlu(
 
                 output_path = os.path.join(model_output_path, f"{model_name}_report")
                 processor = Agent.load(model_path=model_path).processor
-                result = run_evaluation(
+                result = await run_evaluation(
                     test_path, processor, output_directory=output_path, errors=True
                 )
 
diff --git a/rasa/server.py b/rasa/server.py
index 7d5bdc2e4844..0e8ede17be5f 100644
--- a/rasa/server.py
+++ b/rasa/server.py
@@ -1131,7 +1131,7 @@ async def evaluate_intents(
             test_data = payload.get("training_files")
 
             if cross_validation_folds:
-                test_coroutine = _cross_validate(
+                test_coroutine = await _cross_validate(
                     test_data, config_file, int(cross_validation_folds)
                 )
         else:
@@ -1199,7 +1199,7 @@ async def _cross_validate(data_file: Text, config_file: Text, folds: int) -> Dic
         config = importer.get_config()
         nlu_data = importer.get_nlu_data()
 
-        evaluations = rasa.nlu.test.cross_validate(
+        evaluations = await rasa.nlu.test.cross_validate(
             data=nlu_data,
             n_folds=folds,
             nlu_config=config,
@@ -1282,7 +1282,7 @@ async def parse(request: Request) -> HTTPResponse:
         try:
             data = emulator.normalise_request_json(request.json)
             try:
-                parsed_data = app.agent.parse_message(data.get("text"))
+                parsed_data = await app.agent.parse_message(data.get("text"))
             except Exception as e:
                 logger.debug(traceback.format_exc())
                 raise ErrorResponse(
diff --git a/tests/conftest.py b/tests/conftest.py
index a845ef88575b..470f99e9f3f1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -209,7 +209,7 @@ def event_loop(request: Request) -> Iterator[asyncio.AbstractEventLoop]:
 
 
 @pytest.fixture(scope="session")
-async def _trained_default_agent_model(
+async def trained_default_agent_model(
     tmp_path_factory: TempPathFactory,
     stories_path: Text,
     domain_path: Text,
@@ -254,8 +254,8 @@ def reset_conversation_state(agent: Agent) -> Agent:
 
 
 @pytest.fixture
-def default_agent(_trained_default_agent_model: Text) -> Agent:
-    return Agent.load(_trained_default_agent_model)
+def default_agent(trained_default_agent_model: Text) -> Agent:
+    return Agent.load(trained_default_agent_model)
 
 
 @pytest.fixture(scope="session")
@@ -563,11 +563,6 @@ async def response_selector_test_stories() -> Path:
     return Path("data/test_response_selector_bot/tests/test_stories.yml")
 
 
-@pytest.fixture(scope="session")
-async def response_selector_results() -> Path:
-    return Path("data/test_response_selector_bot/results/")
-
-
 @pytest.fixture(scope="session")
 async def trained_response_selector_bot(trained_async: Callable) -> Path:
     zipped_model = await trained_async(
diff --git a/tests/core/test_agent.py b/tests/core/test_agent.py
index 9533c86c01d3..77bb09cb2bac 100644
--- a/tests/core/test_agent.py
+++ b/tests/core/test_agent.py
@@ -1,9 +1,11 @@
 import asyncio
+import json
 from pathlib import Path
 from typing import Any, Dict, Text, Callable, Optional
 from unittest.mock import patch
 import uuid
 
+from aioresponses import aioresponses
 import pytest
 from _pytest.monkeypatch import MonkeyPatch
 from pytest_sanic.utils import TestClient
@@ -12,6 +14,7 @@
 from sanic.response import StreamingHTTPResponse
 
 import rasa.core
+from rasa.core.utils import AvailableEndpoints
 from rasa.exceptions import ModelNotFound
 from rasa.nlu.persistor import Persistor
 from rasa.shared.core.events import (
@@ -21,6 +24,7 @@
     SessionStarted,
     UserUttered,
 )
+from rasa.shared.nlu.constants import INTENT_NAME_KEY
 import rasa.shared.utils.common
 import rasa.utils.io
 from rasa.core import jobs
@@ -95,7 +99,7 @@ async def test_agent_train(default_agent: Agent):
 async def test_agent_parse_message(
     default_agent: Agent, text_message_data: Text, expected: Dict[Text, Any]
 ):
-    result = default_agent.parse_message(text_message_data)
+    result = await default_agent.parse_message(text_message_data)
     assert result == expected
 
 
@@ -296,3 +300,27 @@ async def test_agent_update_model(trained_core_model: Text, trained_nlu_model: T
         agent1.processor.model_metadata.predict_schema
         == agent2.processor.model_metadata.predict_schema
     )
+
+
+async def test_parse_with_http_interpreter(trained_default_agent_model: Text):
+    endpoints = AvailableEndpoints(nlu=EndpointConfig("https://interpreter.com"))
+    agent = await load_agent(
+        model_path=trained_default_agent_model, endpoints=endpoints
+    )
+    response_body = {
+        "intent": {INTENT_NAME_KEY: "some_intent", "confidence": 1.0},
+        "entities": [],
+        "text": "lunch?",
+    }
+
+    with aioresponses() as mocked:
+        mocked.post(
+            "https://interpreter.com/model/parse",
+            repeat=True,
+            status=200,
+            body=json.dumps(response_body),
+        )
+
+        # mock the parse function with the one defined for this test
+        result = await agent.parse_message("lunch?")
+        assert result == response_body
diff --git a/tests/core/test_processor.py b/tests/core/test_processor.py
index 1bc647e37e7a..afe5f1cba0bf 100644
--- a/tests/core/test_processor.py
+++ b/tests/core/test_processor.py
@@ -12,9 +12,10 @@
 from _pytest.logging import LogCaptureFixture
 from aioresponses import aioresponses
 from typing import Optional, Text, List, Callable, Type, Any
-from unittest.mock import patch
 
+from rasa.core.lock_store import InMemoryLockStore
 from rasa.core.policies.ensemble import DefaultPolicyPredictionEnsemble
+from rasa.core.tracker_store import InMemoryTrackerStore
 import rasa.shared.utils.io
 from rasa.core.actions.action import (
     ActionBotResponse,
@@ -55,7 +56,6 @@
 )
 from rasa.core.http_interpreter import RasaNLUHttpInterpreter
 from rasa.core.processor import MessageProcessor
-from rasa.shared.core.slots import Slot
 from rasa.shared.core.trackers import DialogueStateTracker
 from rasa.shared.nlu.constants import INTENT_NAME_KEY
 from rasa.shared.nlu.training_data.message import Message
@@ -91,7 +91,7 @@ async def test_message_processor(
 async def test_message_id_logging(default_processor: MessageProcessor):
     message = UserMessage("If Meg was an egg would she still have a leg?")
     tracker = DialogueStateTracker("1", [])
-    default_processor._handle_message_with_tracker(message, tracker)
+    await default_processor._handle_message_with_tracker(message, tracker)
     logged_event = tracker.events[-1]
 
     assert logged_event.message_id == message.message_id
@@ -100,12 +100,12 @@ async def test_message_id_logging(default_processor: MessageProcessor):
 
 async def test_parsing(default_processor: MessageProcessor):
     message = UserMessage('/greet{"name": "boy"}')
-    parsed = default_processor.parse_message(message)
+    parsed = await default_processor.parse_message(message)
     assert parsed["intent"][INTENT_NAME_KEY] == "greet"
     assert parsed["entities"][0]["entity"] == "name"
 
 
-def test_check_for_unseen_feature(default_processor: MessageProcessor):
+async def test_check_for_unseen_feature(default_processor: MessageProcessor):
     message = UserMessage('/greet{"name": "Joe"}')
     old_domain = default_processor.domain
     new_domain = Domain.from_dict(old_domain.as_dict())
@@ -117,7 +117,7 @@ def test_check_for_unseen_feature(default_processor: MessageProcessor):
     new_domain.entities = [e for e in new_domain.entities if e != "name"]
     default_processor.domain = new_domain
 
-    parsed = default_processor.parse_message(message)
+    parsed = await default_processor.parse_message(message)
     with pytest.warns(UserWarning) as record:
         default_processor._check_for_unseen_features(parsed)
     assert len(record) == 2
@@ -133,63 +133,81 @@ async def test_default_intent_recognized(
     default_processor: MessageProcessor, default_intent: Text
 ):
     message = UserMessage(f"/{default_intent}")
-    parsed = default_processor.parse_message(message)
+    parsed = await default_processor.parse_message(message)
     with pytest.warns(None) as record:
         default_processor._check_for_unseen_features(parsed)
     assert len(record) == 0
 
 
-# TODO: Fix and re-enable(x) once RasaNLUHttpInterpreter graph component is implemented
-async def xtest_http_parsing():
+async def test_http_parsing(trained_default_agent_model: Text, domain: Domain):
     message = UserMessage("lunch?")
 
     endpoint = EndpointConfig("https://interpreter.com")
+
+    response_body = {
+        "intent": {INTENT_NAME_KEY: "some_intent", "confidence": 1.0},
+        "entities": [],
+        "text": "lunch?",
+    }
+
     with aioresponses() as mocked:
-        mocked.post("https://interpreter.com/model/parse", repeat=True, status=200)
+        mocked.post(
+            "https://interpreter.com/model/parse",
+            repeat=True,
+            status=200,
+            body=json.dumps(response_body),
+        )
 
         inter = RasaNLUHttpInterpreter(endpoint_config=endpoint)
-        try:
-            MessageProcessor(inter, None, None, None, None, None).parse_message(message)
-        except KeyError:
-            pass  # logger looks for intent and entities, so we except
+        processor = MessageProcessor(
+            trained_default_agent_model,
+            InMemoryTrackerStore(domain),
+            InMemoryLockStore(),
+            NaturalLanguageGenerator(),
+            http_interpreter=inter,
+        )
+        data = await processor.parse_message(message)
 
         r = tests.utilities.latest_request(
             mocked, "POST", "https://interpreter.com/model/parse"
         )
 
         assert r
+        assert data == response_body
 
 
-async def mocked_parse(self, text, message_id=None, tracker=None, metadata=None):
-    """Mock parsing a text message and augment it with the slot
-    value from the tracker's state."""
-
-    return {
-        "intent": {INTENT_NAME_KEY: "", "confidence": 0.0},
-        "entities": [],
-        "text": text,
-        "requested_language": tracker.get_slot("requested_language"),
-    }
-
-
-# TODO: Fix and re-enable(x) once RasaNLUHttpInterpreter graph component is implemented
-async def xtest_parsing_with_tracker():
-    tracker = DialogueStateTracker.from_dict("1", [], [Slot("requested_language")])
-
-    # we'll expect this value 'en' to be part of the result from the interpreter
-    tracker._set_slot("requested_language", "en")
+async def test_http_parsing_default_response(
+    trained_default_agent_model: Text, domain: Domain
+):
+    message = UserMessage("lunch?")
 
     endpoint = EndpointConfig("https://interpreter.com")
+
     with aioresponses() as mocked:
-        mocked.post("https://interpreter.com/parse", repeat=True, status=200)
+        mocked.post(
+            "https://interpreter.com/model/parse", repeat=True, status=200, body=None,
+        )
+
+        inter = RasaNLUHttpInterpreter(endpoint_config=endpoint)
+        processor = MessageProcessor(
+            trained_default_agent_model,
+            InMemoryTrackerStore(domain),
+            InMemoryLockStore(),
+            NaturalLanguageGenerator(),
+            http_interpreter=inter,
+        )
+        data = await processor.parse_message(message)
 
-        # mock the parse function with the one defined for this test
-        with patch.object(RasaNLUHttpInterpreter, "parse", mocked_parse):
-            interpreter = RasaNLUHttpInterpreter(endpoint_config=endpoint)
-            agent = Agent(None, None, interpreter)
-            result = agent.parse_message("lunch?", tracker)
+        r = tests.utilities.latest_request(
+            mocked, "POST", "https://interpreter.com/model/parse"
+        )
 
-            assert result["requested_language"] == "en"
+        assert r
+        assert data == {
+            "intent": {INTENT_NAME_KEY: "", "confidence": 0.0},
+            "entities": [],
+            "text": "",
+        }
 
 
 async def test_reminder_scheduled(
@@ -1301,10 +1319,10 @@ async def test_processor_logs_text_tokens_in_tracker(mood_agent: Agent):
     assert event_tokens == indices
 
 
-def test_parse_message_nlu_only(trained_moodbot_nlu_path: Text):
+async def test_parse_message_nlu_only(trained_moodbot_nlu_path: Text):
     processor = Agent.load(model_path=trained_moodbot_nlu_path).processor
     message = UserMessage("/greet")
-    result = processor.parse_message(message)
+    result = await processor.parse_message(message)
     assert result == {
         "text": "/greet",
         "intent": {"name": "greet", "confidence": 1.0},
@@ -1313,14 +1331,14 @@ def test_parse_message_nlu_only(trained_moodbot_nlu_path: Text):
     }
 
     message = UserMessage("Hello")
-    result = processor.parse_message(message)
+    result = await processor.parse_message(message)
     assert result["intent"]["name"]
 
 
-def test_parse_message_core_only(trained_core_model: Text):
+async def test_parse_message_core_only(trained_core_model: Text):
     processor = Agent.load(model_path=trained_core_model).processor
     message = UserMessage("/greet")
-    result = processor.parse_message(message)
+    result = await processor.parse_message(message)
     assert result == {
         "text": "/greet",
         "intent": {"name": "greet", "confidence": 1.0},
@@ -1329,14 +1347,14 @@ def test_parse_message_core_only(trained_core_model: Text):
     }
 
     message = UserMessage("Hello")
-    result = processor.parse_message(message)
+    result = await processor.parse_message(message)
     assert not result["intent"]["name"]
 
 
-def test_parse_message_full_model(trained_moodbot_path: Text):
+async def test_parse_message_full_model(trained_moodbot_path: Text):
     processor = Agent.load(model_path=trained_moodbot_path).processor
     message = UserMessage("/greet")
-    result = processor.parse_message(message)
+    result = await processor.parse_message(message)
     assert result == {
         "text": "/greet",
         "intent": {"name": "greet", "confidence": 1.0},
@@ -1345,7 +1363,7 @@ def test_parse_message_full_model(trained_moodbot_path: Text):
     }
 
     message = UserMessage("Hello")
-    result = processor.parse_message(message)
+    result = await processor.parse_message(message)
     assert result["intent"]["name"]
 
 
diff --git a/tests/core/test_test.py b/tests/core/test_test.py
index 07a524c08fdd..d3bc4aa33fed 100644
--- a/tests/core/test_test.py
+++ b/tests/core/test_test.py
@@ -112,8 +112,13 @@ 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,
 ):
-    rasa.core.test.test(e2e_bot_test_stories_with_unknown_bot_utterances, e2e_bot_agent)
+    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
@@ -123,20 +128,17 @@ async def test_testing_warns_if_action_unknown(
 async def test_testing_with_utilizing_retrieval_intents(
     response_selector_agent: Agent,
     response_selector_test_stories: Path,
-    response_selector_results: Path,
+    tmp_path: Path,
 ):
-    if not response_selector_results.exists():
-        response_selector_results.mkdir()
-
-    result = rasa.core.test.test(
+    result = await rasa.core.test.test(
         stories=response_selector_test_stories,
         agent=response_selector_agent,
         e2e=True,
-        out_directory=response_selector_results,
+        out_directory=str(tmp_path),
         disable_plotting=True,
         warnings=False,
     )
-    failed_stories_path = response_selector_results / "failed_test_stories.yml"
+    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 = {
@@ -164,10 +166,12 @@ async def test_testing_with_utilizing_retrieval_intents(
 
 
 async def test_testing_does_not_warn_if_intent_in_domain(
-    default_agent: Agent, stories_path: Text,
+    default_agent: Agent, stories_path: Text, tmp_path: Path
 ):
     with pytest.warns(UserWarning) as record:
-        rasa.core.test.test(Path(stories_path), default_agent)
+        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(
@@ -176,9 +180,11 @@ async def test_testing_does_not_warn_if_intent_in_domain(
     )
 
 
-async def test_testing_valid_with_non_e2e_core_model(core_agent: Agent):
-    result = rasa.core.test.test(
-        "data/test_yaml_stories/test_stories_entity_annotations.yml", core_agent
+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()
 
@@ -265,7 +271,7 @@ async def test_action_unlikely_intent_warning(
     # predicted correctly.
     agent = await _train_rule_based_agent(file_name, True)
 
-    result = rasa.core.test.test(
+    result = await rasa.core.test.test(
         str(file_name),
         agent,
         out_directory=str(tmp_path),
@@ -315,7 +321,7 @@ async def test_action_unlikely_intent_correctly_predicted(
     # predicted correctly.
     agent = await _train_rule_based_agent(file_name, False)
 
-    result = rasa.core.test.test(
+    result = await rasa.core.test.test(
         str(file_name),
         agent,
         out_directory=str(tmp_path),
@@ -378,7 +384,7 @@ async def test_wrong_action_after_action_unlikely_intent(
     # predicted correctly.
     agent = await _train_rule_based_agent(train_file_name, True)
 
-    result = rasa.core.test.test(
+    result = await rasa.core.test.test(
         str(test_file_name),
         agent,
         out_directory=str(tmp_path),
@@ -448,7 +454,7 @@ async def test_action_unlikely_intent_not_found(
     # predicted correctly.
     agent = await _train_rule_based_agent(train_file_name, False)
 
-    result = rasa.core.test.test(
+    result = await rasa.core.test.test(
         str(test_file_name), agent, out_directory=str(tmp_path)
     )
     assert "report" in result.keys()
@@ -514,7 +520,7 @@ async def test_action_unlikely_intent_warning_and_story_error(
     # predicted correctly.
     agent = await _train_rule_based_agent(train_file_name, True)
 
-    result = rasa.core.test.test(
+    result = await rasa.core.test.test(
         str(test_file_name), agent, out_directory=str(tmp_path),
     )
     assert "report" in result.keys()
@@ -564,7 +570,7 @@ async def test_fail_on_prediction_errors(
     agent = await _train_rule_based_agent(file_name, False)
 
     with pytest.raises(rasa.core.test.WrongPredictionException):
-        rasa.core.test.test(
+        await rasa.core.test.test(
             str(file_name),
             agent,
             out_directory=str(tmp_path),
@@ -659,7 +665,7 @@ async def test_multiple_warnings_sorted_on_severity(
     # predicted correctly.
     agent = await _train_rule_based_agent(Path(test_story_path), True)
 
-    rasa.core.test.test(
+    await rasa.core.test.test(
         test_story_path,
         agent,
         out_directory=str(tmp_path),
diff --git a/tests/core/test_tracker_stores.py b/tests/core/test_tracker_stores.py
index e8c7820fcf5a..6e52a08dcea8 100644
--- a/tests/core/test_tracker_stores.py
+++ b/tests/core/test_tracker_stores.py
@@ -881,7 +881,7 @@ def test_tracker_store_connection_error(config: Dict, domain: Domain):
         TrackerStore.create(store, domain)
 
 
-def prepare_token_serialisation(
+async def prepare_token_serialisation(
     tracker_store: TrackerStore, response_selector_agent: Agent, sender_id: Text,
 ):
     text = "Good morning"
@@ -890,7 +890,7 @@ def prepare_token_serialisation(
     indices = [[t.start, t.end] for t in tokens]
 
     tracker = tracker_store.get_or_create_tracker(sender_id=sender_id)
-    parse_data = response_selector_agent.parse_message(text)
+    parse_data = await response_selector_agent.parse_message(text)
     event = UserUttered(
         "Good morning",
         parse_data.get("intent"),
diff --git a/tests/nlu/test_evaluation.py b/tests/nlu/test_evaluation.py
index 0dcb0d8136a1..1cedb0d09540 100644
--- a/tests/nlu/test_evaluation.py
+++ b/tests/nlu/test_evaluation.py
@@ -10,7 +10,7 @@
 
 import pytest
 from _pytest.monkeypatch import MonkeyPatch
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 
 import rasa.nlu.test
 import rasa.shared.nlu.training_data.loading
@@ -347,7 +347,7 @@ def test_drop_intents_below_freq():
     300, func_only=True
 )  # these can take a longer time than the default timeout
 async def test_run_evaluation(mood_agent: Agent, nlu_as_json_path: Text):
-    result = run_evaluation(
+    result = await run_evaluation(
         nlu_as_json_path,
         mood_agent.processor,
         errors=False,
@@ -358,7 +358,7 @@ async def test_run_evaluation(mood_agent: Agent, nlu_as_json_path: Text):
     assert result.get("intent_evaluation")
 
 
-def test_eval_data(
+async def test_eval_data(
     tmp_path: Path, project: Text, trained_rasa_model: Text,
 ):
     config_path = os.path.join(project, "config.yml")
@@ -373,7 +373,7 @@ def test_eval_data(
     processor = Agent.load(trained_rasa_model).processor
 
     data = data_importer.get_nlu_data()
-    (intent_results, response_selection_results, entity_results) = get_eval_data(
+    (intent_results, response_selection_results, entity_results) = await get_eval_data(
         processor, data
     )
 
@@ -385,7 +385,7 @@ def test_eval_data(
 @pytest.mark.timeout(
     180, func_only=True
 )  # these can take a longer time than the default timeout
-def test_run_cv_evaluation():
+async def test_run_cv_evaluation():
     td = rasa.shared.nlu.training_data.loading.load_data(
         "data/test/demo-rasa-more-ents-and-multiplied.yml"
     )
@@ -400,7 +400,7 @@ def test_run_cv_evaluation():
     }
 
     n_folds = 2
-    intent_results, entity_results, response_selection_results = cross_validate(
+    intent_results, entity_results, response_selection_results = await cross_validate(
         td,
         n_folds,
         nlu_config,
@@ -429,7 +429,7 @@ def test_run_cv_evaluation():
 @pytest.mark.timeout(
     180, func_only=True
 )  # these can take a longer time than the default timeout
-def test_run_cv_evaluation_no_entities():
+async def test_run_cv_evaluation_no_entities():
     td = rasa.shared.nlu.training_data.loading.load_data(
         "data/test/demo-rasa-no-ents.yml"
     )
@@ -444,7 +444,7 @@ def test_run_cv_evaluation_no_entities():
     }
 
     n_folds = 2
-    intent_results, entity_results, response_selection_results = cross_validate(
+    intent_results, entity_results, response_selection_results = await cross_validate(
         td,
         n_folds,
         nlu_config,
@@ -475,7 +475,7 @@ def test_run_cv_evaluation_no_entities():
 @pytest.mark.timeout(
     180, func_only=True
 )  # these can take a longer time than the default timeout
-def test_run_cv_evaluation_with_response_selector():
+async def test_run_cv_evaluation_with_response_selector():
     training_data_obj = rasa.shared.nlu.training_data.loading.load_data(
         "data/test/demo-rasa-more-ents-and-multiplied.yml"
     )
@@ -495,7 +495,7 @@ def test_run_cv_evaluation_with_response_selector():
     }
 
     n_folds = 2
-    intent_results, entity_results, response_selection_results = cross_validate(
+    intent_results, entity_results, response_selection_results = await cross_validate(
         training_data_obj,
         n_folds,
         nlu_config,
@@ -901,7 +901,7 @@ def test_label_replacement():
     assert substitute_labels(original_labels, "O", "no_entity") == target_labels
 
 
-def test_nlu_comparison(
+async def test_nlu_comparison(
     tmp_path: Path, monkeypatch: MonkeyPatch, nlu_as_json_path: Text
 ):
     config = {
@@ -919,7 +919,7 @@ def test_nlu_comparison(
     monkeypatch.setattr(
         sys.modules["rasa.nlu.test"],
         "get_eval_data",
-        Mock(return_value=(1, None, (None,),)),
+        AsyncMock(return_value=(1, None, (None,),)),
     )
     monkeypatch.setattr(
         sys.modules["rasa.nlu.test"],
@@ -932,7 +932,7 @@ def test_nlu_comparison(
         training_data_paths=[nlu_as_json_path]
     )
     test_data = test_data_importer.get_nlu_data()
-    compare_nlu_models(
+    await compare_nlu_models(
         configs, test_data, output, runs=2, exclusion_percentages=[50, 80]
     )
 
@@ -1122,13 +1122,13 @@ class ConstantProcessor:
     def __init__(self, prediction_to_return: Dict[Text, Any]) -> None:
         self.prediction = prediction_to_return
 
-    def parse_message(
+    async def parse_message(
         self, message: UserMessage, only_output_properties: bool = True,
     ) -> Dict[Text, Any]:
         return self.prediction
 
 
-def test_replacing_fallback_intent():
+async def test_replacing_fallback_intent():
     expected_intent = "greet"
     expected_confidence = 0.345
     fallback_prediction = {
@@ -1154,7 +1154,7 @@ def test_replacing_fallback_intent():
         [Message.build("hi", "greet"), Message.build("bye", "bye")]
     )
 
-    intent_evaluations, _, _ = get_eval_data(processor, training_data)
+    intent_evaluations, _, _ = await get_eval_data(processor, training_data)
 
     assert all(
         prediction.intent_prediction == expected_intent
@@ -1163,7 +1163,7 @@ def test_replacing_fallback_intent():
     )
 
 
-def test_remove_entities_of_extractors():
+async def test_remove_entities_of_extractors():
     extractor = "TestExtractor"
     extractor_2 = "DIET"
     extractor_3 = "YetAnotherExtractor"
diff --git a/tests/nlu/test_train.py b/tests/nlu/test_train.py
index ea3d2011714b..8c50097f150e 100644
--- a/tests/nlu/test_train.py
+++ b/tests/nlu/test_train.py
@@ -143,7 +143,7 @@ def test_all_components_are_in_at_least_one_test_pipeline():
 
 @pytest.mark.timeout(600, func_only=True)
 @pytest.mark.parametrize("language, pipeline", pipelines_for_tests())
-def test_train_persist_load_parse(
+async def test_train_persist_load_parse(
     language: Optional[Text],
     pipeline: List[Dict],
     tmp_path: Path,
@@ -163,7 +163,7 @@ def test_train_persist_load_parse(
     agent = Agent.load(persisted_path)
     assert agent.processor
     assert agent.is_ready()
-    assert agent.parse_message("Rasa is great!") is not None
+    assert await agent.parse_message("Rasa is great!") is not None
 
 
 @pytest.mark.timeout(600, func_only=True)
diff --git a/tests/test_model_testing.py b/tests/test_model_testing.py
index ee5d8bab3847..0e0b2445dfcc 100644
--- a/tests/test_model_testing.py
+++ b/tests/test_model_testing.py
@@ -146,7 +146,7 @@ def test_get_label_set(targets, exclude_label, expected):
     assert set(expected) == set(actual)
 
 
-def test_e2e_warning_if_no_nlu_model(
+async def test_e2e_warning_if_no_nlu_model(
     monkeypatch: MonkeyPatch, trained_core_model: Text, capsys: CaptureFixture
 ):
     from rasa.model_testing import test_core
@@ -157,7 +157,7 @@ def test_e2e_warning_if_no_nlu_model(
         sys.modules["rasa.core.test"], "test", asyncio.coroutine(lambda *_, **__: True)
     )
 
-    test_core(trained_core_model, use_conversation_test_files=True)
+    await test_core(trained_core_model, use_conversation_test_files=True)
 
     assert "No NLU model found. Using default" in capsys.readouterr().out
 


Code review comments:
diff_hunk: @@ -257,6 +257,15 @@ def run_nlu_test(args: argparse.Namespace) -> None:
     )
 
 
+def run_core_test(args: argparse.Namespace) -> None:
+    """Runs NLU tests.
+
+    Args:
+        args: the parsed CLI arguments for 'rasa test nlu'.
path: rasa/cli/test.py
text: @user1:
```suggestion
        args: the parsed CLI arguments for 'rasa test [core]'.
```

@author:
oops - I actually changed this and forgot to push 🤦

diff_hunk: @@ -573,7 +576,34 @@ def _get_action(
             action_name, self.domain, self.action_endpoint
         )
 
-    def parse_message(
+    async def parse_message(
+        self, message: UserMessage, only_output_properties: bool = True
+    ) -> Dict[Text, Any]:
+        """Interprets the passed message.
+
+        Arguments:
+            message: Message to handle
path: rasa/core/processor.py
text: @user1:
`only_output_properties` is missing from `Arguments`

diff_hunk: @@ -296,3 +300,27 @@ async def test_agent_update_model(trained_core_model: Text, trained_nlu_model: T
         agent1.processor.model_metadata.predict_schema
         == agent2.processor.model_metadata.predict_schema
     )
+
+
+async def test_parse_with_http_interpreter(trained_default_agent_model: Text):
+    endpoints = AvailableEndpoints(nlu=EndpointConfig("https://interpreter.com"))
+    agent = await load_agent(
+        model_path=trained_default_agent_model, endpoints=endpoints
+    )
+    response_body = {
+        "intent": {INTENT_NAME_KEY: "some_intent", "confidence": 1.0},
+        "entities": [],
+        "text": "lunch?",
+    }
+
+    with aioresponses() as mocked:
+        mocked.post(
+            "https://interpreter.com/model/parse",
+            repeat=True,
+            status=200,
path: tests/core/test_agent.py
text: @user1:
nit
```suggestion
            status=HTTPStatus.OK,
```