Tests

.coveragerc

@@ -0,0 +1,2 @@
+[run]
+source_pkgs = transWhat

setup.py

@@ -35,7 +35,6 @@ def read_file(filename, encoding="utf8"):
     packages=["transWhat"],
     scripts=["transWhat/transwhat.py"],
     install_requires=[
-        "protobuf",
         "yowsup",
         "pyspectrum2",
         "python-dateutil",

tests/conftest.py

@@ -0,0 +1,99 @@
+import pytest
+from asyncio import Transport
+from spectrum2 import Backend
+from transWhat.session import Session
+from unittest.mock import create_autospec
+from yowsup.stacks import YowStack
+from collections import deque
+from transWhat.yowsupwrapper import YowsupAppLayer
+
+
+class MockYowstack:
+    def __init__(self):
+        self.events = deque()
+        self.props = {}
+
+    def setProp(self, key, value):
+        self.props[key] = value
+
+    def broadcastEvent(self, event):
+        self.events.append(event)
+
+    def assert_event_called(self, event_name, **kwargs):
+        for event in self.events:
+            if event.name != event_name:
+                continue
+
+            if event.args != kwargs:
+                continue
+
+            return
+
+        raise AssertionError(
+            f"event name={event_name}, args={kwargs} not found in {self.events}"
+        )
+
+    def assert_entity_called(self, entity):
+        _qualname = type(entity).__qualname__
+
+        for event in self.events:
+            if event.name != YowsupAppLayer.TO_LOWER_EVENT:
+                continue
+
+            got_entity = event.args["entity"]
+
+            is_equal = all(
+                (
+                    _qualname == type(got_entity).__qualname__,
+                    got_entity.__dict__ == entity.__dict__,
+                )
+            )
+
+            if is_equal:
+                return
+
+        raise AssertionError(f"entity {entity} not found in {self.events}")
+
+    def assert_iq_called(self, iq_type, success=None, failure=None):
+        _qualname = iq_type.__qualname__
+
+        for event in self.events:
+            if event.name != YowsupAppLayer.SEND_IQ:
+                continue
+
+            iq = event.args["iq"]
+
+            if type(iq).__qualname__ != _qualname:
+                continue
+
+            if event.args["success"] != success:
+                continue
+
+            if event.args["failure"] != failure:
+                continue
+
+            return
+
+        raise AssertionError(
+            f"iq type={iq_type} success={success} error={error} not found in {self.events}"
+        )
+
+
+@pytest.fixture
+def mock_transport() -> Transport:
+    yield create_autospec(Transport)
+
+
+@pytest.fixture
+def mock_session() -> Session:
+    yield create_autospec(Session)
+
+
+@pytest.fixture
+def mock_backend() -> Backend:
+    yield create_autospec(Backend)
+
+
+@pytest.fixture
+def mock_yowstack():
+    return MockYowstack()

tests/spectrum2_utils.py

@@ -0,0 +1,21 @@
+from faker import Faker
+from spectrum2.protocol_pb2 import (
+    Buddy, Buddies
+)
+
+
+def generate_buddy(faker: Faker) -> Buddy:
+    """
+    Generate a fake buddy.
+    """
+    return Buddy(
+        userName=faker.user_name(),
+        buddyName=faker.user_name(),
+
+    )
+
+def generate_buddies(faker: Faker, count: int = 10) -> Buddies:
+    """
+    Generate fake buddies
+    """
+    return Buddies(buddy=[generate_buddy(faker) for _ in range(count)])

tests/test_deferred.py

@@ -0,0 +1,51 @@
+from pytest import raises
+from unittest.mock import Mock
+from transWhat.deferred import Deferred, DeferredHasValue, call
+
+
+def test_deferred_argument_forwarding_works():
+    test_cases = [[], [1], [1, 2], [1, 2, 3]]
+
+    for test_case in test_cases:
+        cb = Mock()
+
+        fut = Deferred()
+        fut.then(cb)
+
+        fut.run(*test_case)
+
+        cb.assert_called_with(*test_case)
+
+
+def test_deferred_pipelining_works():
+    cb = Mock()
+
+    fut = Deferred()
+    fut.append(2)
+    fut.then(cb)
+    fut.run([1])
+
+    cb.assert_called_with([1, 2])
+
+
+def test_deferred_raises_when_running_twice():
+    cb = Mock()
+
+    fut = Deferred()
+    fut.then(cb)
+    fut.run()
+
+    with raises(DeferredHasValue):
+        fut.run()
+
+
+def test_call_appends_arguments():
+    cb = Mock()
+
+    fut = Deferred()
+    fut.extend((1, 2))
+    fut.run([])
+
+    call(cb, fut)
+
+    cb.assert_called_with([1, 2])

tests/test_session.py

@@ -0,0 +1,149 @@
+import pytest
+
+from unittest.mock import patch, Mock, create_autospec
+from transWhat.session import Session
+from typing import Optional
+from yowsup.structs import ProtocolEntity
+from yowsup.stacks import YowStack
+from yowsup.layers import YowLayerEvent
+from yowsup.layers.network import YowNetworkLayer
+from yowsup.layers.protocol_privacy.protocolentities.privacylist_iq import (
+    PrivacyListIqProtocolEntity,
+)
+from yowsup.layers.protocol_iq.protocolentities.iq_push import PushIqProtocolEntity
+from yowsup.layers.protocol_iq.protocolentities.iq_props import PropsIqProtocolEntity
+from transWhat.yowsupwrapper import YowsupAppLayer
+from contextlib import contextmanager
+from yowsup.layers.protocol_presence.protocolentities import (
+    AvailablePresenceProtocolEntity,
+    UnavailablePresenceProtocolEntity,
+)
+
+from yowsup.layers.protocol_media.protocolentities import (
+    MediaMessageProtocolEntity,
+    DownloadableMediaMessageProtocolEntity,
+    ImageDownloadableMediaMessageProtocolEntity,
+)
+
+
+@pytest.fixture
+def session_instance(mock_backend, mock_yowstack):
+    mock_user = "[email protected]"
+    mock_legacy_name = "legacy name"
+
+    sess = Session(
+        backend=mock_backend, user=mock_user, legacyName=mock_legacy_name, extra=[]
+    )
+    sess.stack = mock_yowstack
+
+    yield sess
+
+
+def test_login_works(session_instance, mock_yowstack):
+    password = "test password"
+
+    session_instance.login(password)
+    assert session_instance.password == password
+    mock_yowstack.assert_event_called(YowNetworkLayer.EVENT_STATE_CONNECT)
+
+
+def test_logout_works(session_instance, mock_yowstack):
+    session_instance.loggedIn = True
+
+    session_instance.logout()
+    assert not session_instance.loggedIn
+
+    mock_yowstack.assert_event_called(YowNetworkLayer.EVENT_STATE_DISCONNECT)
+
+
+def test_send_message_to_xmpp_queued(session_instance, mock_backend):
+    session_instance.initialized = False
+
+    session_instance.sendMessageToXMPP(
+        buddy="[email protected]",
+        message="this is a random message",
+        timestamp=1,
+    )
+
+    assert not mock_backend.handle_message.called
+
+
+def test_send_message_to_xmpp_proper(session_instance, mock_backend):
+    session_instance.initialized = True
+
+    session_instance.sendMessageToXMPP(
+        buddy="[email protected]", message="this is a random message", timestamp=1
+    )
+
+    mock_backend.handle_message.assert_called_with(
+        "[email protected]",
+        "[email protected]",
+        "this is a random message",
+        "",  # why
+        "",  # why
+        "19700101T000001",
+    )
+
+
+@pytest.mark.skip("updateRoomList and _updateGroups are broken and do not work")
+def test_update_room_list(session_instance):
+    raise NotImplementedError()
+
+
+@pytest.mark.skip("joinRoom depends on updateRoomList which does not work yet")
+def test_join_room(session_instance):
+    raise NotImplementedError()
+
+
+@pytest.mark.skip("leaveRoom depends on updateRoomList which does not work yet")
+def test_leave_room(session_instance):
+    raise NotImplementedError()
+
+
+def test_callback_on_auth_success(session_instance, mock_backend, mock_yowstack):
+    session_instance.onAuthSuccess("success", 123, 1, 1, 1, 1, 1)
+
+    assert session_instance.initialized
+    assert session_instance.loggedIn
+
+    # Interaction with backend works
+    mock_backend.handle_connected.assert_called_with(session_instance.user)
+    mock_backend.handle_buddy_changed.assert_called()
+
+    # Interaction with Yowsup, these are called in subfunctions
+    # TODO: check for toLower call
+    mock_yowstack.assert_iq_called(PushIqProtocolEntity)
+    mock_yowstack.assert_iq_called(PrivacyListIqProtocolEntity)
+    mock_yowstack.assert_iq_called(PropsIqProtocolEntity)
+
+    # .. and presence is true now
+    mock_yowstack.assert_entity_called(AvailablePresenceProtocolEntity())
+
+
+def test_callback_on_auth_failed(session_instance, mock_backend):
+    session_instance.onAuthFailed("reason")
+
+    mock_backend.handle_disconnected.assert_called()
+
+    assert not session_instance.loggedIn
+    assert session_instance.password is None
+
+
+def test_callback_on_disconnect(session_instance, mock_backend):
+    session_instance.onDisconnect()
+
+    mock_backend.handle_disconnected.assert_called()
+
+
+def test_callback_on_media_works_with_images(session_instance, mock_backend):
+    media_inst = create_autospec(DownloadableMediaMessageProtocolEntity)
+
+    media_inst.media_type = MediaMessageProtocolEntity.TYPE_MEDIA_IMAGE
+    media_inst.url = "http://test.com/a-picture"
+    media_inst.caption = "my first image"
+    media_inst.participant = None
+    media_inst.getFrom.return_value = "buddy"
+    media_inst.getTimestamp.return_value = 1
+    # media_inst.isEncrypted.return_value = False
+
+    session_instance.onMedia(media_inst, "image")

tests/test_threadutils.py

@@ -0,0 +1,16 @@
+from unittest.mock import Mock
+from transWhat.threadutils import eventQueue, runInThread
+
+
+def test_runInThread():
+    magic = object()
+
+    def thread_func():
+        return magic
+
+    def callback(result):
+        return result == magic
+
+    runInThread(thread_func, callback)
+    result = eventQueue.get(timeout=1.0)
+    assert result