RasaHQ/rasa_core

View on GitHub
tests/core/test_server.py

Summary

Maintainability
B
5 hrs
Test Coverage
# -*- 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"]