From 9d95cbd60637f5e17d5eac3bfd8fd19be3ad2b8e Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Mon, 3 Jun 2019 15:48:24 -0400 Subject: [PATCH] rename test_bumper to test_init - rename test_bumper to test_init - added more xmppserver tests --- tests/{test_bumper.py => test_init.py} | 0 tests/test_xmppserver.py | 310 ++++++++++++++++++++++++- 2 files changed, 303 insertions(+), 7 deletions(-) rename tests/{test_bumper.py => test_init.py} (100%) diff --git a/tests/test_bumper.py b/tests/test_init.py similarity index 100% rename from tests/test_bumper.py rename to tests/test_init.py diff --git a/tests/test_xmppserver.py b/tests/test_xmppserver.py index 8bdf6cd..95fa8e1 100644 --- a/tests/test_xmppserver.py +++ b/tests/test_xmppserver.py @@ -6,6 +6,7 @@ import os import json import tinydb import pytest_asyncio +import xml.etree.ElementTree as ET def return_send_data(data, *args, **kwargs): @@ -16,15 +17,15 @@ def mock_transport_extra_info(*args, **kwargs): return ("127.0.0.1", 1234) -async def test_client_init(*args, **kwargs): +async def test_client_connect_no_starttls(*args, **kwargs): test_transport = asyncio.Transport() test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) test_transport.write = mock.Mock(return_value=return_send_data) xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) - xmppclient.state = xmppclient.INIT # Set client state to INIT + xmppclient.state = xmppclient.CONNECT # Set client state to CONNECT mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) - # Send init connect stream from "client" + # Send connect stream from "client" test_data = "".encode( "utf-8" ) @@ -40,22 +41,109 @@ async def test_client_init(*args, **kwargs): # Server tells client available features assert ( mock_send.mock_calls[1].args[0] - == '' + == 'PLAIN' ) # Reset mock calls mock_send.reset_mock() + # Client sendss auth - Ignoring the starttls, we don't force this with bumper + test_data = 'AGZ1aWRfdG1wdXNlcgAwL0lPU0Y1M0QwN0JBL3VzXzg5ODgwMmZkYmM0NDQxYjBiYzgxNWIxZDFjNjgzMDJl'.encode( + "utf-8" + ) + xmppclient._parse_data(test_data) -async def test_bot_init(*args, **kwargs): + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) # Client successfully authenticated + assert xmppclient.state == xmppclient.INIT # Client moved to INIT state + + +async def test_client_connect_starttls_called(*args, **kwargs): test_transport = asyncio.Transport() test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) test_transport.write = mock.Mock(return_value=return_send_data) xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) - xmppclient.state = xmppclient.INIT # Set client state to INIT + xmppclient.state = xmppclient.CONNECT # Set client state to CONNECT mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) - # Send init connect stream from "client" + # Send connect stream from "client" + test_data = "".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + # Expect 2 calls to send + assert mock_send.call_count == 2 + # Server opens stream + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) + # Server tells client available features + assert ( + mock_send.mock_calls[1].args[0] + == 'PLAIN' + ) + + # Reset mock calls + mock_send.reset_mock() + + mock_tls = xmppclient._handle_starttls = mock.Mock() + + # Send start tls from "client" + test_data = "".encode("utf-8") + xmppclient._parse_data(test_data) + + # After upgrading connection, server tells client to proceed with auth again + assert mock_tls.called + xmppclient.TLSUpgraded = True + + # After TLS is upgraded, Client establishes session again and will auth this time + # Send connect stream from "client" + test_data = "".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + # Expect 2 calls to send + assert mock_send.call_count == 2 + # Server opens stream + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) + # Server tells client available features (without STARTTLS) + assert ( + mock_send.mock_calls[1].args[0] + == 'PLAIN' + ) + # Reset mock calls + mock_send.reset_mock() + + # Client sends auth + test_data = 'AGZ1aWRfdG1wdXNlcgAwL0lPU0Y1M0QwN0JBL3VzXzg5ODgwMmZkYmM0NDQxYjBiYzgxNWIxZDFjNjgzMDJl'.encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) # Client successfully authenticated + assert xmppclient.state == xmppclient.INIT # Client moved to INIT state + + +async def test_bot_connect(*args, **kwargs): + test_transport = asyncio.Transport() + test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) + test_transport.write = mock.Mock(return_value=return_send_data) + xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient.state = xmppclient.CONNECT # Set client state to CONNECT + mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) + + # Send connect stream from "bot" test_data = "".encode( "utf-8" ) @@ -69,6 +157,52 @@ async def test_bot_init(*args, **kwargs): == '' ) # Server tells client available features + assert ( + mock_send.mock_calls[1].args[0] + == 'PLAIN' + ) + + # Reset mock calls + mock_send.reset_mock() + + # Send auth from "bot" + test_data = "AEUwMDAwMDAwMDAwMDAwMDAxMjM0AGVuY3J5cHRlZF9wYXNz".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) # Bot successfully authenticated + assert xmppclient.state == xmppclient.INIT # Bot moved to INIT state + assert xmppclient.type == xmppclient.BOT # Client type is now bot + + +async def test_bot_init(*args, **kwargs): + test_transport = asyncio.Transport() + test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) + test_transport.write = mock.Mock(return_value=return_send_data) + xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient.state = xmppclient.INIT # Set client state to INIT + xmppclient.uid = "E0000000000000001234" + xmppclient.devclass = "159" + mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) + + # Send connect stream from "bot" + test_data = "".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + # Expect 2 calls to send + assert mock_send.call_count == 2 + # Server opens stream + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) + # Server tells client binds assert ( mock_send.mock_calls[1].args[0] == '' @@ -77,3 +211,165 @@ async def test_bot_init(*args, **kwargs): # Reset mock calls mock_send.reset_mock() + # Send bind from "bot" + test_data = "atom".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == 'E0000000000000001234@159.ecorobot.net/atom' + ) # Bot successfully binded + assert xmppclient.state == xmppclient.BIND # Bot moved to BIND state + + # Reset mock calls + mock_send.reset_mock() + + # Send set session from bot + test_data = "".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert xmppclient.state == xmppclient.READY # Bot moved to READY state + assert ( + mock_send.mock_calls[0].args[0] == '' + ) # Bot ready + + # Reset mock calls + mock_send.reset_mock() + + # Send presense from bot + test_data = "hello world".encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == ' dummy ' + ) # bot presence - dummy response + + +async def test_ping_server(*args, **kwargs): + test_transport = asyncio.Transport() + test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) + test_transport.write = mock.Mock(return_value=return_send_data) + xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient.state = xmppclient.READY # Set client state to READY + xmppclient.uid = "E0000000000000001234" + xmppclient.devclass = "159" + mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) + + # Ping from bot + test_data = ''.encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) # ping response + + +async def test_ping_client_to_client(*args, **kwargs): + + test_transport = asyncio.Transport() + test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) + test_transport.write = mock.Mock(return_value=return_send_data) + xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient.state = xmppclient.READY # Set client state to READY + xmppclient.uid = "E0000000000000001234" + xmppclient.devclass = "159" + xmppclient.bumper_jid = "E0000000000000001234@159.ecorobot.net/atom" + mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) + + xmppclient2 = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient2.state = xmppclient.READY # Set client state to READY + xmppclient2.uid = "fuid_tmpuser" + xmppclient2.resource = "IOSF53D07BA" + xmppclient2.bumper_jid = "fuid_tmpuser@ecouser.net/IOSF53D07BA" + mock_send2 = xmppclient2.send = mock.Mock(side_effect=return_send_data) + + bumper.xmppserver.XMPPServer.clients.append(xmppclient) + bumper.xmppserver.XMPPServer.clients.append(xmppclient2) + + # Ping from user to bot + test_data = ''.encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + # Ping response from bot to user + test_data = "".encode( + "utf-8" + ) + xmppclient2._parse_data(test_data) + + +async def test_client_send_iq(*args, **kwargs): + test_transport = asyncio.Transport() + test_transport.get_extra_info = mock.Mock(return_value=mock_transport_extra_info()) + test_transport.write = mock.Mock(return_value=return_send_data) + xmppclient = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient.state = xmppclient.READY # Set client state to READY + xmppclient.uid = "fuid_tmpuser" + xmppclient.resource = "IOSF53D07BA" + xmppclient.bumper_jid = "fuid_tmpuser@ecouser.net/IOSF53D07BA" + xmppclient.type - xmppclient.CONTROLLER + mock_send = xmppclient.send = mock.Mock(side_effect=return_send_data) + bumper.xmppserver.XMPPServer.clients.append(xmppclient) + + xmppclient2 = bumper.xmppserver.XMPPAsyncClient(test_transport) + xmppclient2.state = xmppclient.READY # Set client state to READY + xmppclient2.uid = "E0000000000000001234" + xmppclient2.devclass = "159" + xmppclient2.bumper_jid = "E0000000000000001234@159.ecorobot.net/atom" + xmppclient2.type = xmppclient2.BOT + mock_send2 = xmppclient2.send = mock.Mock(side_effect=return_send_data) + + bumper.xmppserver.XMPPServer.clients.append(xmppclient2) + + # Roster IQ - Only seen from Android app so far + test_data = ''.encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) # ping response + + # Reset mock calls + mock_send.reset_mock() + + # Bot Command + test_data = ''.encode( + "utf-8" + ) + xmppclient._parse_data(test_data) + + assert ( + mock_send2.mock_calls[0].args[0] + == '' + ) # command was sent to bot + + # Reset mock calls + mock_send.reset_mock() + + # Bot response to query + test_data = ''.encode( + "utf-8" + ) + xmppclient2._parse_data(test_data) + + assert ( + mock_send.mock_calls[0].args[0] + == '' + ) # result sent to client + + # Reset mock calls + mock_send.reset_mock()