Permalink
Browse files

Create unit tests for sms_bear_gateway

  • Loading branch information...
osteele committed Feb 10, 2018
1 parent 23c883c commit ea381ade7035032f016a9211de30aaa9004950b5
Showing with 87 additions and 7 deletions.
  1. +24 −7 examples/sms_bear_gateway.py
  2. +63 −0 tests/sms_bear_gateway_test.py
@@ -5,8 +5,8 @@
import sys

import click
import twilio.rest
from profanityfilter import ProfanityFilter
from twilio.rest import Client

sys.path.append(os.path.join(os.path.dirname(__file__), './..')) # noqa: I003
import mqtt_json # noqa: E402,I001
@@ -24,7 +24,8 @@
assert AUTH_TOKEN, 'Error: the TWILIO_AUTH_TOKEN is not set'
assert PHONE_NUMBER, 'Error: the TWILIO_PHONE_NUMBER is not set'

REPLY_TEXT = os.getenv('BEAR_REPLY_TEXT') or "The bear has received your message."
# FIXME REPLY_TEXT isn't used. Find a place for it, or remove it.
REPLY_TEXT = os.getenv('BEAR_REPLY_TEXT', "The bear has received your message.")
UNCLEAN_MESSAGE_REPLY_TEXT = "Hey! That's not very nice. Keep it clean, kids!"

CANNED_SPEECHES = [
@@ -51,7 +52,7 @@
mqtt_client = mqtt_json.Client()


def process_message(message, reply_text=None):
def process_text_message(message, reply_text=None):
from_number = message['From']
message_body = message['Body']

@@ -63,14 +64,28 @@ def process_message(message, reply_text=None):
else:
response_text = UNCLEAN_MESSAGE_REPLY_TEXT
if response_text:
client = Client(ACCOUNT_SID, AUTH_TOKEN)
client.api.account.messages.create(
to=from_number, # sic
twilio_client = twilio.rest.Client(ACCOUNT_SID, AUTH_TOKEN)
# `to` and `from` are reversed: this message goes *to* the number that
# the incoming message came *from*.
twilio_client.api.account.messages.create(
to=from_number, # sic — see previous comment
from_=PHONE_NUMBER,
body=response_text)


def parse_command(command):
"""Parse a possible text command, and returns the text to speak.
* 'say' -> speak a canned speech
* 'say something' -> speak the words after 'say'
* 'speak' or 'speak something' -> 'speak' is the same as 'say'
* anything else -> speak the anything else
"""
# remove 8-bit characters
# FIXME if the intent is to remove non-ASCII, it should remove < 32 too
# TODO probably the worker should do this instead, since it's connected
# to the synthesizer and knows whether the synthesizer can handle accents
# and other non-ASCII
command = ''.join(c for c in command
if ord(c) < 128 and c not in string.punctuation)

@@ -81,6 +96,8 @@ def parse_command(command):
else:
return random.choice(CANNED_SPEECHES)
else:
# TODO probably the worker should do this instead. Some synthesizers
# might be able to do meaningful things with case.
return command.lower()


@@ -91,7 +108,7 @@ def main(reply_text=None):
topic = 'incoming-sms-' + PHONE_NUMBER.strip('+')
logger.info('Waiting for messages on {}'.format(topic))
for payload in mqtt_client.create_subscription_queue(topic):
process_message(payload, reply_text=reply_text)
process_text_message(payload, reply_text=reply_text)


if __name__ == '__main__':
@@ -0,0 +1,63 @@
import os
import sys
from unittest.mock import patch
# noqa: I003

os.environ['TWILIO_ACCOUNT_SID'] = '--test-account-id--'
os.environ['TWILIO_AUTH_TOKEN'] = '--test-auth-token--'
os.environ['TWILIO_PHONE_NUMBER'] = '--test-phone-number--'
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'examples'))
import sms_bear_gateway # noqa: E402,I001,I003


@patch('random.choice', return_value='random response')
def test_parse_command(random_choice):
# canned responses
assert sms_bear_gateway.parse_command('say') == 'random response'
assert sms_bear_gateway.parse_command('speak') == 'random response'
# verbatim
assert sms_bear_gateway.parse_command('say hello') == 'hello'
assert sms_bear_gateway.parse_command('speak hello') == 'hello'
# downcases test
assert sms_bear_gateway.parse_command('speak Hello') == 'hello'
# removes punctuation
assert sms_bear_gateway.parse_command('speak Hello, world!') == 'hello world'
# removes non-ASCII characters
assert sms_bear_gateway.parse_command('say Héllo') == 'hllo'


@patch('twilio.rest.Client')
@patch('sms_bear_gateway.pf.is_clean', wraps=lambda s: 'profanity' not in s)
@patch('sms_bear_gateway.mqtt_client')
def test_process_text_message(mqtt_client, profanity_filter, twilio_client):
def mock_message(text):
"""Create a mock Twilio message from a text string."""
return dict(From='+16175552323', Body=text)

def process_text_message_with(text, reply_text=None):
"""Call process_text_message.
The `text` value is wrapped in a Twilio message.
This function resets the mocks before making the call.
"""
mqtt_client.reset_mock()
twilio_client.reset_mock()
sms_bear_gateway.process_text_message(mock_message(text),
reply_text=reply_text)

# publishes the message
process_text_message_with('say hello')
mqtt_client.publish.assert_called_with('speak', message='hello')

# sends a response message when reply_text is specified
process_text_message_with('say hello', reply_text='Got your message')
twilio_client.api.account.messages.create.assert_not_called()

# doesn't respond when reply_text isn't specified
process_text_message_with('say hello')
twilio_client.api.account.messages.create.assert_not_called()

# handles profanity
process_text_message_with('say profanity')
mqtt_client.publish.assert_not_called()
twilio_client.api.account.messages.create.assert_not_called()

0 comments on commit ea381ad

Please sign in to comment.