Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
test.py
#!/usr/bin/env python
import requests
import json
import uuid
import secrets
import yaml
base_url = 'http://172.28.1.2:5005'
def api_post(path, data = '{}', debug_level = 0):
return call_api(path, data, "post", debug_level)
def api_get(path, data = '{}', debug_level = 0):
return call_api(path, data, "get", debug_level)
def call_api(path, data = '{}', method = "post", debug_level = 0):
if debug_level > 0:
print ("\n [x] Called %s" % path)
if debug_level > 1:
print (" [x] Payload %r" % data)
if method == "post":
response = requests.post(base_url+path, data=data)
else:
response = requests.get(base_url+path, data=data)
if(response.ok):
if debug_level > 0:
print (" [x] Status: %d" % response.status_code)
reply = response.json()
if debug_level > 1:
print(" [x] Received: %r" % json.dumps(reply))
else:
# If response code is not ok (200), print the resulting http error code with description
print (" [x] Status: %d ERROR!" % response.status_code)
print (" [x] Payload %r" % data)
print(" [x] Received: %s" % response.text)
reply = {}
return reply
def predict_next_action(conversation_id, debug_level = 0):
res = api_post("/conversations/"+conversation_id +"/predict", '{}', debug_level)
predicted = res['scores'][0]['action']
return predicted
def get_actions(conversation_id, message_text, debug_level = 0):
actions = []
# Add a message to the tracker
res = api_post("/conversations/"+conversation_id +"/messages", "{\"text\": \"" + message_text + "\", \"sender\": \"user\"}", debug_level)
# Predict next action
predicted = predict_next_action(conversation_id, debug_level)
actions.append(predicted)
count = 1
while (predicted != "action_listen"):
# Run action
res = api_post("/conversations/"+conversation_id +"/execute", "{\"name\": \"" + predicted + "\" }", debug_level)
#if res['messages'] and res['messages'][0]['text']:
# print(" > "+res['messages'][0]['text'])
# Predict next action
predicted = predict_next_action(conversation_id, debug_level)
actions.append(predicted)
count = count + 1
if count > 20:
print ("STOPPED: TOO MANY ACTIONS!!")
break
# Add actions to tracker for next run
for action in actions:
# Add action to tracker
res = api_post("/conversations/"+conversation_id +"/tracker/events", "{ \"event\": \"action\", \"name\": \"" + action + "\" }", debug_level)
return actions[:-1] # Remove last action, 'action_listen'
def load_config():
with open("/rasa-pebble/tools/test.yml", 'r') as stream:
try:
return yaml.safe_load(stream)
except yaml.YAMLError as exc:
print(exc)
def print_green(text):
print("\033[92m"+text+"\033[0m")
def print_red(text):
print("\033[91m"+text+"\033[0m")
# -------------------------------------------- Start main program --------------------------------------------
print ("\nStarting...\n")
stories = load_config()
conversation_id = uuid.uuid4().hex # Random ID
# Loop through each story
for story, tests in stories.items():
print("\n------------------------------------------------------------------------\n")
print("\nStory to test: "+story)
# Loop through all tests in story
for n in range(len(tests)):
test_item = tests[n]
if type(test_item) is not str and test_item.get('message'):
message = test_item.get('message')[0]
print("\n Message: "+message)
# Predict next actions
actual_actions = get_actions(conversation_id, message)
if type(test_item) is not str and test_item.get('actions'):
expected_actions = test_item.get('actions')
if (expected_actions == actual_actions):
print_green(" TEST OK")
print(" Actions: %r " % expected_actions)
else:
print_red(" TEST FAILED")
print(" Expected actions: %r " % expected_actions)
print(" Actual actions: %r " % actual_actions)
print ("\nFinished\n")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.