test_contacts.py

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

import json

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_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_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, mocker):
    contacts = Contacts(api)
    contacts._api.socket_listener = SocketListener(None)
    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