Initial MessagesAPI tests

Author: cmlccie

tests/api/test_messages.py

@@ -2,50 +2,211 @@
 
 """pytest Messages functions, fixtures and tests."""
 
+import itertools
 
 import pytest
 
 import ciscosparkapi
-from tests.utils import create_string
+from tests.conftest import TEST_FILE_URL
+from tests.utils import create_string, download_file
 
 
 # Helper Functions
 
-def send_direct_message_to_email(api, email, message=None):
-    msg_text = message if message else create_string("Message")
-    return api.messages.create(toPersonEmail=email, text=msg_text)
+def create_message(api, **message_attributes):
+    return api.messages.create(**message_attributes)
+
+
+def get_message_by_id(api, id):
+    return api.messages.get(id)
 
 
 def delete_message(api, message):
     api.messages.delete(message.id)
 
 
-def delete_messages(api, messages):
-    for message in messages:
+def delete_messages(api, iterable):
+    for message in iterable:
         delete_message(api, message)
 
 
+def list_messages(api, roomId, **query_parameters):
+    return list(api.messages.list(roomId, **query_parameters))
+
+
+def is_valid_message(obj):
+    return isinstance(obj, ciscosparkapi.Message) and obj.id is not None
+
+
+def are_valid_messages(iterable):
+    return all([is_valid_message(obj) for obj in iterable])
+
+
+def message_exists(api, message):
+    try:
+        api.messages.get(message.id)
+    except ciscosparkapi.SparkApiError:
+        return False
+    else:
+        return True
+
+
 # pytest Fixtures
 
 @pytest.fixture(scope="session")
-def direct_messages(api, additional_group_room_memberships, test_people):
-    # Using additional_group_room_memberships to ensure some test_people
-    # have been created
-    direct_messages = []
-    for person in test_people:
-        message = send_direct_message_to_email(api, person.emails[0])
-        direct_messages.append(message)
+def direct_message_by_email(api, test_people):
+    person = test_people["member_added_by_email"]
+    message = create_message(api, toPersonEmail=person.emails[0],
+                             text=create_string("Message"))
 
-    yield direct_messages
+    yield message
 
-    for message in direct_messages:
-        delete_message(api, message)
+    delete_message(api, message)
 
 
 @pytest.fixture(scope="session")
-def send_direct_message(api):
+def direct_message_by_id(api, test_people):
+    person = test_people["member_added_by_id"]
+    message = create_message(api, toPersonId=person.id,
+                             text=create_string("Message"))
+
+    yield message
 
-    def inner_function(email_address, message=None):
-        send_direct_message_to_email(api, email_address, message=message)
+    delete_message(api, message)
 
-    return inner_function
+
+@pytest.fixture(scope="session")
+def send_group_room_message(api):
+    messages = []
+
+    def inner_function(roomId, **message_attributes):
+        message = create_message(api, roomId=roomId, **message_attributes)
+        messages.append(message)
+        return message
+
+    yield inner_function
+
+    for message in messages:
+        try:
+            delete_message(api, message)
+        except ciscosparkapi.SparkApiError as e:
+            pass
+
+
+@pytest.fixture(scope="session")
+def group_room_text_message(group_room, send_group_room_message):
+    text = create_string("Message")
+    return send_group_room_message(group_room.id, text=text)
+
+
+@pytest.fixture(scope="session")
+def group_room_markdown_message(group_room, send_group_room_message, me,
+                                group_room_member_added_by_email,
+                                group_room_text_message):
+    # Uses / depends on group_room_text_message to ensure this message is
+    # created after group_room_text_message, so that we can be sure that a
+    # message exists 'before' this one - used to test 'before' list filters.
+    mention_email = me.emails[0]
+    markdown = create_string("<personEmail:{}>, This is **markdown** with a "
+                             "mention.".format(mention_email))
+    return send_group_room_message(group_room.id, markdown=markdown)
+
+
+@pytest.fixture(scope="session")
+def group_room_file_by_URL_message(group_room, send_group_room_message):
+    text = "File posted via URL"
+    return send_group_room_message(group_room.id, text=text,
+                                      files=[TEST_FILE_URL])
+
+
+@pytest.fixture(scope="session")
+def group_room_file_by_local_upload_message(api, group_room, local_file,
+                                            send_group_room_message):
+    text = "File posted via URL"
+    return send_group_room_message(group_room.id, text=text,
+                                   files=[local_file])
+
+
+@pytest.fixture(scope="session")
+def group_room_messages(api, group_room,
+                        group_room_text_message,
+                        group_room_markdown_message,
+                        group_room_file_by_URL_message,
+                        group_room_file_by_local_upload_message):
+    return list(list_messages(api, group_room.id))
+
+
+@pytest.fixture(scope="session")
+def direct_messages(api, direct_message_by_email, direct_message_by_id):
+    return [direct_message_by_email, direct_message_by_id]
+
+
+# Tests
+
+class TestMessagesAPI(object):
+    """Test MessagesAPI methods."""
+
+    def test_create_direct_messages_by_email(self, direct_message_by_email):
+        assert is_valid_message(direct_message_by_email)
+
+    def test_create_direct_messages_by_id(self, direct_message_by_id):
+        assert is_valid_message(direct_message_by_id)
+
+    def test_create_text_message(self, group_room_text_message):
+        assert is_valid_message(group_room_text_message)
+
+    def test_create_markdown_message(self, group_room_markdown_message):
+        assert is_valid_message(group_room_markdown_message)
+
+    def test_post_file_by_url(self, group_room_file_by_URL_message):
+        assert is_valid_message(group_room_file_by_URL_message)
+
+    def test_post_file_by_local_upload(self, group_room_file_by_local_upload_message):
+        assert is_valid_message(group_room_file_by_local_upload_message)
+
+    def test_get_message_by_id(self, api, group_room_text_message):
+        message = get_message_by_id(api, group_room_text_message.id)
+        assert is_valid_message(message)
+
+    def test_delete_message(self, api, group_room, send_group_room_message):
+        text = create_string("Message")
+        message = create_message(api, roomId=group_room.id, text=text)
+        assert is_valid_message(message)
+        delete_message(api, message)
+        assert not message_exists(api, message)
+
+    def test_list_all_messages_in_room(self, group_room_messages):
+        assert len(group_room_messages) >= 1
+        assert are_valid_messages(group_room_messages)
+
+    def test_list_messages_before(self, api, group_room,
+                                  group_room_markdown_message):
+        message_list = list_messages(api, group_room.id,
+                                     before=group_room_markdown_message.created)
+        assert len(message_list) >= 1
+        assert are_valid_messages(message_list)
+
+    def test_list_messages_before_message(self, api, group_room,
+                                          group_room_markdown_message):
+        message_list = list_messages(api, group_room.id,
+                                     beforeMessage=group_room_markdown_message.id)
+        assert len(message_list) >= 1
+        assert are_valid_messages(message_list)
+
+    def test_list_messages_with_paging(self, api, group_room,
+                                       group_room_messages):
+        page_size = 1
+        pages = 3
+        num_messages = pages * page_size
+        assert len(group_room_messages) >= num_messages
+        messages = list_messages(api, group_room.id, max=page_size)
+        messages_list = list(itertools.islice(messages, num_messages))
+        assert len(messages_list) == num_messages
+        assert are_valid_messages(messages_list)
+
+    # TODO: Investigate API list messages with 'me' mentions not working
+    # def test_list_messages_with_me_mention(self, api, group_room, me):
+    #     messages = list_messages(api, group_room.id, mentionedPeople=["me"])
+    #     messages_list = list(messages)
+    #     assert len(messages_list) >= 1
+    #     assert are_valid_messages(messages_list)

tests/conftest.py

@@ -45,6 +45,24 @@ def new_email_generator():
 
 # pytest Fixtures
 
+@pytest.fixture("session")
+def temp_directory():
+    directory_abs_path = tempfile.mkdtemp()
+
+    yield directory_abs_path
+
+    os.rmdir(directory_abs_path)
+
+
+@pytest.fixture("session")
+def local_file(temp_directory):
+        file = download_file(TEST_FILE_URL, temp_directory)
+
+        yield file
+
+        os.remove(file)
+
+
 @pytest.fixture(scope="session")
 def get_new_email_address():
     generator = new_email_generator()

tests/utils.py

@@ -4,8 +4,11 @@
 
 
 import datetime
+import os
 import string
 
+import requests
+
 
 STRING_PREFIX = "ciscosparkapi py.test"
 STRING_TEMPLATE = string.Template("$prefix $item [$datetime]")
@@ -14,6 +17,23 @@
 # Helper Functions
 
 def create_string(item):
+    """Create strings for tests; prefixed-timestamped strings."""
     return STRING_TEMPLATE.substitute(prefix=STRING_PREFIX,
                                       item=item,
                                       datetime=str(datetime.datetime.now()))
+
+
+def download_file(url, local_directory, local_filename=None):
+    """Download a file from a remote URL to a local directory."""
+    # http://stackoverflow.com/questions/16694907/
+    # how-to-download-large-file-in-python-with-requests-py
+    local_filename = local_filename if local_filename \
+        else url.split('/')[-1]
+    local_path = os.path.normpath(os.path.join(local_directory,
+                                               local_filename))
+    response = requests.get(url, stream=True)
+    with open(local_path, 'wb') as f:
+        for chunk in response.iter_content(chunk_size=1024):
+            if chunk:
+                f.write(chunk)
+    return local_path