content: # -*- coding: utf-8 -*-
  import json
  import os
  import tempfile
  import uuid
  
  import pytest
  from freezegun import freeze_time
  
  import rasa.core
  from rasa.core import events, constants
  from rasa.core.events import (
      UserUttered, BotUttered, SlotSet, Event)
  from rasa.model import unpack_model, add_evaluation_file_to_model
  from tests.core.conftest import DEFAULT_STORIES_FILE, END_TO_END_STORY_FILE
  
  # a couple of event instances that we can use for testing
  test_events = [
      Event.from_parameters({"event": UserUttered.type_name,
                             "text": "/goodbye",
                             "parse_data": {
                                 "intent": {
                                     "confidence": 1.0, "name": "greet"},
                                 "entities": []}
                             }),
      BotUttered("Welcome!", {"test": True}),
      SlotSet("cuisine", 34),
      SlotSet("cuisine", "34"),
      SlotSet("location", None),
      SlotSet("location", [34, "34", None]),
  ]
  
  
  @pytest.fixture
  def app(core_server):
      return core_server.test_client
  
  
  @pytest.fixture
  def secured_app(core_server_secured):
      return core_server_secured.test_client
  
  
  def test_root(app):
      _, response = app.get("/")
      content = response.text
      assert response.status == 200
      assert content.startswith("hello")
  
  
  def test_root_secured(secured_app):
      _, response = secured_app.get("/")
      content = response.text
      assert response.status == 200 and content.startswith("hello")
  
  
  def test_version(app):
      _, response = app.get("/version")
      content = response.json
      assert response.status == 200
      assert content.get("version") == rasa.__version__
      assert (content.get("minimum_compatible_version") ==
              constants.MINIMUM_COMPATIBLE_VERSION)
  
  
  def test_status(app):
      _, response = app.get("/status")
      content = response.json
      assert response.status == 200
      assert content.get("is_ready")
      assert content.get("model_fingerprint") is not None
  
  
  @freeze_time("2018-01-01")
  def test_requesting_non_existent_tracker(app):
      _, response = app.get("/conversations/madeupid/tracker")
      content = response.json
      assert response.status == 200
      assert content["paused"] is False
      assert content["slots"] == {"location": None, "cuisine": None}
      assert content["sender_id"] == "madeupid"
      assert content["events"] == [{"event": "action",
                                    "name": "action_listen",
                                    "policy": None,
                                    "confidence": None,
                                    "timestamp": 1514764800}]
      assert content["latest_message"] == {"text": None,
                                           "intent": {},
                                           "entities": []}
  
  
  def test_respond(app):
      data = json.dumps({"query": "/greet"})
      _, response = app.post("/conversations/myid/respond",
                             data=data,
                             headers={"Content-Type": "application/json"})
      content = response.json
      assert response.status == 200
      assert content == [{'text': 'hey there!', 'recipient_id': 'myid'}]
  
  
  def test_parse(app):
      data = json.dumps({"q": """/greet{"name": "Rasa"}"""})
      _, response = app.post("/parse",
                             data=data,
                             headers={"Content-Type": "application/json"})
      content = response.json
      assert response.status == 200
      assert content == {
          'entities': [
              {'end': 22, 'entity': 'name', 'start': 6, 'value': 'Rasa'}],
          'intent': {'confidence': 1.0, 'name': 'greet'},
          'intent_ranking': [{'confidence': 1.0, 'name': 'greet'}],
          'text': '/greet{"name": "Rasa"}'
      }
  
  
  @pytest.mark.parametrize("event", test_events)
  def test_pushing_event(app, event):
      cid = str(uuid.uuid1())
      conversation = "/conversations/{}".format(cid)
      data = json.dumps({"query": "/greet"})
      _, response = app.post("{}/respond".format(conversation),
                             data=data,
                             headers={"Content-Type": "application/json"})
      assert response.json is not None
      assert response.status == 200
  
      data = json.dumps(event.as_dict())
      _, response = app.post("{}/tracker/events".format(conversation),
                             data=data,
                             headers={"Content-Type": "application/json"})
      assert (response.json is not None)
      assert response.status == 200
  
      _, tracker_response = app.get("/conversations/{}/tracker"
                                    "".format(cid))
      tracker = tracker_response.json
      assert tracker is not None
      assert len(tracker.get("events")) == 6
  
      evt = tracker.get("events")[5]
      assert Event.from_parameters(evt) == event
  
  
  def test_put_tracker(app):
      data = json.dumps([event.as_dict() for event in test_events])
      _, response = app.put(
          "/conversations/pushtracker/tracker/events",
          data=data, headers={"Content-Type": "application/json"})
      content = response.json
      assert response.status == 200
      assert len(content["events"]) == len(test_events)
      assert content["sender_id"] == "pushtracker"
  
      _, tracker_response = app.get(
          "/conversations/pushtracker/tracker")
      tracker = tracker_response.json
      assert tracker is not None
      evts = tracker.get("events")
      assert events.deserialise_events(evts) == test_events
  
  
  def test_sorted_predict(app):
      data = json.dumps([event.as_dict() for event in test_events[:3]])
      _, response = app.put(
          "/conversations/sortedpredict/tracker/events",
          data=data, headers={"Content-Type": "application/json"})
  
      assert response.status == 200
  
      _, response = app.post("/conversations/sortedpredict/predict")
      scores = response.json["scores"]
      sorted_scores = sorted(scores, key=lambda k: (-k['score'], k['action']))
      assert scores == sorted_scores
  
  
  def test_list_conversations(app):
      data = json.dumps({"query": "/greet"})
      _, response = app.post("/conversations/myid/respond",
                             data=data,
                             headers={"Content-Type": "application/json"})
      assert response.json is not None
      assert response.status == 200
  
      _, response = app.get("/conversations")
      content = response.json
      assert response.status == 200
  
      assert len(content) > 0
      assert "myid" in content
  
  
  def test_evaluate(app):
      with open(DEFAULT_STORIES_FILE, 'r') as f:
          stories = f.read()
      _, response = app.post('/evaluate',
                             data=stories)
      assert response.status == 200
      js = response.json
      assert set(js.keys()) == {"report",
                                "precision",
                                "f1",
                                "accuracy",
                                "actions",
                                "in_training_data_fraction",
                                "is_end_to_end_evaluation"}
      assert not js["is_end_to_end_evaluation"]
      assert set(js["actions"][0].keys()) == {
          "action",
          "predicted",
          "confidence",
          "policy"}
  
  
  def test_stack_training(app,
                          default_domain_path,
                          default_stories_file,
                          default_stack_config,
                          default_nlu_data):
      domain_file = open(default_domain_path)
      config_file = open(default_stack_config)
      stories_file = open(default_stories_file)
      nlu_file = open(default_nlu_data)
  
      payload = dict(
          domain=domain_file.read(),
          config=config_file.read(),
          stories=stories_file.read(),
          nlu=nlu_file.read()
      )
  
      domain_file.close()
      config_file.close()
      stories_file.close()
      nlu_file.close()
  
      _, response = app.post('/jobs', json=payload)
      assert response.status == 200
  
      # save model to temporary file
      tempdir = tempfile.mkdtemp()
      model_path = os.path.join(tempdir, 'model.tar.gz')
      with open(model_path, 'wb') as f:
          f.write(response.body)
  
      # unpack model and ensure fingerprint is present
      model_path = unpack_model(model_path)
      assert os.path.exists(os.path.join(model_path, 'fingerprint.json'))
  
  
  def test_intent_evaluation(app, default_nlu_data, trained_stack_model):
      with open(default_nlu_data, 'r') as f:
          nlu_data = f.read()
  
      # add evaluation data to model archive
      zipped_path = add_evaluation_file_to_model(trained_stack_model,
                                                 nlu_data, data_format='md')
  
      # post zipped stack model with evaluation file
      with open(zipped_path, 'r+b') as f:
          _, response = app.post('/intentEvaluation', data=f.read())
  
      assert response.status == 200
      assert set(response.json.keys()) == {"intent_evaluation",
                                           "entity_evaluation"}
  
  
  def test_end_to_end_evaluation(app):
      with open(END_TO_END_STORY_FILE, 'r') as f:
          stories = f.read()
      _, response = app.post('/evaluate?e2e=true',
                             data=stories)
      assert response.status == 200
      js = response.json
      assert set(js.keys()) == {"report",
                                "precision",
                                "f1",
                                "accuracy",
                                "actions",
                                "in_training_data_fraction",
                                "is_end_to_end_evaluation"}
      assert js["is_end_to_end_evaluation"]
      assert set(js["actions"][0].keys()) == {"action",
                                              "predicted",
                                              "confidence",
                                              "policy"}
  
  
  def test_list_conversations_with_jwt(secured_app):
      # token generated with secret "core" and algorithm HS256
      # on https://jwt.io/
  
      # {"user": {"username": "testadmin", "role": "admin"}}
      jwt_header = {
          "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
                           "eyJ1c2VyIjp7InVzZXJuYW1lIjoidGVzdGFkbWluIiwic"
                           "m9sZSI6ImFkbWluIn19.NAQr0kbtSrY7d28XTqRzawq2u"
                           "QRre7IWTuIDrCn5AIw"
      }
      _, response = secured_app.get("/conversations",
                                    headers=jwt_header)
      assert response.status == 200
  
      # {"user": {"username": "testuser", "role": "user"}}
      jwt_header = {
          "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
                           "eyJ1c2VyIjp7InVzZXJuYW1lIjoidGVzdHVzZXIiLCJyb"
                           "2xlIjoidXNlciJ9fQ.JnMTLYd56qut2w9h7hRQlDm1n3l"
                           "HJHOxxC_w7TtwCrs"
      }
      _, response = secured_app.get("/conversations",
                                    headers=jwt_header)
      assert response.status == 403
  
  
  def test_get_tracker_with_jwt(secured_app):
      # token generated with secret "core" and algorithm HS256
      # on https://jwt.io/
  
      # {"user": {"username": "testadmin", "role": "admin"}}
      jwt_header = {
          "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
                           "eyJ1c2VyIjp7InVzZXJuYW1lIjoidGVzdGFkbWluIiwic"
                           "m9sZSI6ImFkbWluIn19.NAQr0kbtSrY7d28XTqRzawq2u"
                           "QRre7IWTuIDrCn5AIw"
      }
      _, response = secured_app.get("/conversations/testadmin/tracker",
                                    headers=jwt_header)
      assert response.status == 200
  
      _, response = secured_app.get("/conversations/testuser/tracker",
                                    headers=jwt_header)
      assert response.status == 200
  
      # {"user": {"username": "testuser", "role": "user"}}
      jwt_header = {
          "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
                           "eyJ1c2VyIjp7InVzZXJuYW1lIjoidGVzdHVzZXIiLCJyb"
                           "2xlIjoidXNlciJ9fQ.JnMTLYd56qut2w9h7hRQlDm1n3l"
                           "HJHOxxC_w7TtwCrs"
      }
      _, response = secured_app.get("/conversations/testadmin/tracker",
                                    headers=jwt_header)
      assert response.status == 403
  
      _, response = secured_app.get("/conversations/testuser/tracker",
                                    headers=jwt_header)
      assert response.status == 200
  
  
  def test_list_conversations_with_token(secured_app):
      _, response = secured_app.get("/conversations?token=rasa")
      assert response.status == 200
  
  
  def test_list_conversations_with_wrong_token(secured_app):
      _, response = secured_app.get("/conversations?token=Rasa")
      assert response.status == 401
  
  
  def test_list_conversations_without_auth(secured_app):
      _, response = secured_app.get("/conversations")
      assert response.status == 401
  
  
  def test_list_conversations_with_wrong_jwt(secured_app):
      jwt_header = {
          "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ"
                           "zdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIi"
                           "wiaWF0IjoxNTE2MjM5MDIyfQ.qdrr2_a7Sd80gmCWjnDomO"
                           "Gl8eZFVfKXA6jhncgRn-I"
      }
      _, response = secured_app.get("/conversations",
                                    headers=jwt_header)
      assert response.status == 401
  
  
  def test_story_export(app):
      data = json.dumps({"query": "/greet"})
      _, response = app.post("/conversations/mynewid/respond",
                             data=data,
                             headers={"Content-Type": "application/json"})
      assert response.status == 200
      _, response = app.get("/conversations/mynewid/story")
      assert response.status == 200
      story_lines = response.text.strip().split('\n')
      assert story_lines == ["## mynewid",
                             "* greet: /greet",
                             "    - utter_greet"]
  