test_contacts.py

# Copyright (c) 2019 Nico Alt
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md

import json

import pytest
import requests_mock

from briar_wrapper.models.contacts import Contacts
from briar_wrapper.models.socket_listener import SocketListener

BASE_HTTP_URL = "http://localhost:7000/v1/contacts/"

TEST_ALIAS = "Alice"
TEST_CONTACT_ID = 42

TEST_CONTACT_FIRST = {
    "lastChatActivity": 1
}
TEST_CONTACT_SECOND = {
    "lastChatActivity": 2
}
TEST_CONTACT_RESPONSE_SINGLE = [
    TEST_CONTACT_FIRST
]
TEST_CONTACT_RESPONSE_TWO_UNORDERED = [
    TEST_CONTACT_FIRST,
    TEST_CONTACT_SECOND
]
TEST_CONTACT_RESPONSE_TWO_ORDERED = [
    TEST_CONTACT_SECOND,
    TEST_CONTACT_FIRST
]

TEST_LINK = "briar://wvui4uvhbfv4tzo6xwngknebsxrafainnhldyfj63x6ipp4q2vigy"


@requests_mock.Mocker(kw="requests_mock")
def test_add_pending(api, request_headers, requests_mock):
    contacts = Contacts(api)
    requests_mock.register_uri("POST",
                               BASE_HTTP_URL + "add/pending/",
                               request_headers=request_headers,
                               additional_matcher=match_request_add_pending)
    contacts.add_pending(TEST_LINK, TEST_ALIAS)


def match_request_add_pending(request):
    return {"alias": TEST_ALIAS, "link": TEST_LINK} == request.json()


@requests_mock.Mocker(kw="requests_mock")
def test_delete(api, request_headers, requests_mock):
    contacts = Contacts(api)
    contact_id = 137
    requests_mock.register_uri("DELETE",
                               BASE_HTTP_URL + str(contact_id),
                               request_headers=request_headers)
    contacts.delete(contact_id)


@requests_mock.Mocker(kw='requests_mock')
def test_get(api, request_headers, requests_mock):
    contacts = Contacts(api)
    requests_mock.register_uri("GET", BASE_HTTP_URL,
                               request_headers=request_headers,
                               text=json.dumps(TEST_CONTACT_RESPONSE_SINGLE))
    assert contacts.get() == TEST_CONTACT_RESPONSE_SINGLE


@requests_mock.Mocker(kw='requests_mock')
def test_get_empty(api, request_headers, requests_mock):
    contacts = Contacts(api)
    requests_mock.register_uri("GET", BASE_HTTP_URL,
                               request_headers=request_headers,
                               text=json.dumps([]))
    assert contacts.get() == []


@requests_mock.Mocker(kw='requests_mock')
def test_get_unordered(api, request_headers, requests_mock):
    contacts = Contacts(api)
    requests_mock.register_uri("GET", BASE_HTTP_URL,
                               request_headers=request_headers,
                               text=json.dumps(
                                   TEST_CONTACT_RESPONSE_TWO_UNORDERED)
                               )
    assert contacts.get() == TEST_CONTACT_RESPONSE_TWO_ORDERED


@requests_mock.Mocker(kw='requests_mock')
def test_get_ordered(api, request_headers, requests_mock):
    contacts = Contacts(api)
    requests_mock.register_uri("GET", BASE_HTTP_URL,
                               request_headers=request_headers,
                               text=json.dumps(
                                   TEST_CONTACT_RESPONSE_TWO_ORDERED)
                               )
    assert contacts.get() == TEST_CONTACT_RESPONSE_TWO_ORDERED


@requests_mock.Mocker(kw='requests_mock')
def test_get_link(api, request_headers, requests_mock):
    contacts = Contacts(api)
    response = {"link": TEST_LINK}

    requests_mock.register_uri("GET", BASE_HTTP_URL + "add/link/",
                               request_headers=request_headers,
                               text=json.dumps(response))
    assert contacts.get_link() == TEST_LINK


def test_watch_signal_added(api):
    contacts = Contacts(api)
    contacts._api.socket_listener = SocketListener(api)
    contacts._api.socket_listener._highest_signal_id = 136

    assert contacts._api.socket_listener._signals == dict()

    contacts.watch_connections(None)

    expected_signals = {
        137: {
            "event": "ContactConnectedEvent",
            "callback": contacts.handle_connections_callback
        },
        138: {
            "event": "ContactDisconnectedEvent",
            "callback": contacts.handle_connections_callback
        }
    }

    assert contacts._api.socket_listener._signals == expected_signals


def test_handle_connections_callback_contact_connected(api, mocker):
    contacts = Contacts(api)
    contacts._connections_callback = mocker.MagicMock()

    message = {
        "data": {
            "contactId": TEST_CONTACT_ID
        },
        "name": "ContactConnectedEvent"
    }

    contacts.handle_connections_callback(message)

    contacts._connections_callback.assert_called_once_with(
        TEST_CONTACT_ID, True
    )


def test_handle_connections_callback_contact_disconnected(api, mocker):
    contacts = Contacts(api)
    contacts._connections_callback = mocker.MagicMock()

    message = {
        "data": {
            "contactId": TEST_CONTACT_ID
        },
        "name": "ContactDisconnectedEvent"
    }

    contacts.handle_connections_callback(message)

    contacts._connections_callback.assert_called_once_with(
        TEST_CONTACT_ID, False
    )


def test_handle_connections_callback_wrong_event(api, mocker):
    contacts = Contacts(api)
    contacts._connections_callback = mocker.MagicMock()

    message = {
        "data": {
            "contactId": TEST_CONTACT_ID
        },
        "name": "SomeEvent"
    }

    with pytest.raises(Exception,
                       match="Wrong event in callback: SomeEvent"):
        contacts.handle_connections_callback(message)