contacts.py

# Copyright (c) 2019 Nico Alt
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md
"""
Wrapper around Briar API's _/contacts/_ resource
"""

from operator import itemgetter
from typing import Callable, List
from urllib.parse import urljoin

from requests import delete as _delete
from requests import get as _get
from requests import post as _post
from requests import put as _put

from briar_wrapper.constants import BASE_HTTP_URL
from briar_wrapper.exception import BriarWrapperException
from briar_wrapper.exceptions.pending_already_exists_contact import \
    PendingContactAlreadyExistsContact
from briar_wrapper.exceptions.pending_already_exists_pending_contact import \
    PendingContactAlreadyExistsPendingContact
from briar_wrapper.exceptions.pending_invalid_link import \
    PendingContactInvalidLinkException
from briar_wrapper.exceptions.pending_invalid_public_key import \
    PendingContactInvalidPublicKeyException
from briar_wrapper.model import Model


class Contacts(Model):
    _API_ENDPOINT = "contacts/"
    _CONNECTION_EVENTS = ("ContactConnectedEvent", "ContactDisconnectedEvent")

    _connections_callback = None

    def add_pending(self, link: str, alias: str) -> bool:
        # pylint: disable=line-too-long
        """
        Adds pending contact to Briar with `link` URL and `alias`

        Raises:
        * `briar_wrapper.exceptions.pending_already_exists_pending_contact.PendingContactAlreadyExistsContact`
        * `briar_wrapper.exceptions.pending_already_exists_pending_contact.PendingContactAlreadyExistsPendingContact`
        * `briar_wrapper.exceptions.pending_invalid_link.PendingContactInvalidLinkException`
        * `briar_wrapper.exceptions.pending_invalid_public_key.PendingContactInvalidPublicKeyException`
        * `briar_wrapper.exception.BriarWrapperException` for unknown API errors

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#adding-a-contact)

        .. versionadded:: 0.0.3
        .. versionchanged:: 0.0.7
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + "add/pending/")
        response = _post(url, headers=self._headers,
                         json={"link": link, "alias": alias})
        if response.status_code == 200:
            return True
        self._handle_add_pending_error(response)

    @staticmethod
    def _handle_add_pending_error(response):
        error = response.json()
        if response.status_code == 400:
            if error["error"] == "INVALID_PUBLIC_KEY":
                raise PendingContactInvalidPublicKeyException(response)
            if error["error"] == "INVALID_LINK":
                raise PendingContactInvalidLinkException(response)
        if response.status_code == 403:
            if error["error"] == "CONTACT_EXISTS":
                raise PendingContactAlreadyExistsContact(response)
            if error["error"] == "PENDING_EXISTS":
                raise PendingContactAlreadyExistsPendingContact(response)
        raise BriarWrapperException(response, "An unknown error occurred while"
                                    f"adding a pending contact: {response.text}")

    def set_alias(self, contact_id: int, alias: str) -> None:
        # pylint: disable=line-too-long
        """
        Sets the alias of a given user

        [Upstream documentation](https://code.briarproject.org/briar/briar/-/blob/master/briar-headless/README.md#changing-alias-of-a-contact)

        .. versionadded:: 0.0.5
        .. versionchanged:: 0.0.6
        """
        url = urljoin(BASE_HTTP_URL,
                      self._API_ENDPOINT + f"{str(contact_id)}/alias")
        _put(url, headers=self._headers, json={"alias": alias})

    def delete(self, contact_id: int) -> None:
        # pylint: disable=line-too-long
        """

        Deletes the contact with `contact_id`

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#removing-a-contact)

        .. versionadded:: 0.0.4
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + str(contact_id))
        _delete(url, headers=self._headers)

    def get(self) -> list:
        # pylint: disable=line-too-long
        """
        Returns sorted list containing all contacts

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#listing-all-contacts)

        .. versionadded:: 0.0.3
        .. versionchanged:: 0.0.4
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT)
        request = _get(url, headers=self._headers)
        contacts = request.json()
        contacts = Contacts._sort_contact_list(contacts)
        return contacts

    def get_link(self) -> str:
        # pylint: disable=line-too-long
        """
        Returns _briar://_ link

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#adding-a-contact)

        .. versionadded:: 0.0.3
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + "add/link/")
        request = _get(url, headers=self._headers).json()
        return request['link']

    def watch_connections(self, callback: Callable) -> List[int]:
        # pylint: disable=line-too-long
        """
        Calls `callback` whenever a contact's connection status changes

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#a-contact-connected-or-disconnected)

        .. versionadded:: 0.0.4
        """
        self._connections_callback = callback
        signal_ids = list()
        event_callback = self.handle_connections_callback
        for event in self._CONNECTION_EVENTS:
            signal_id = self._api.socket_listener.connect(event,
                                                          event_callback)
            signal_ids.append(signal_id)
        return signal_ids

    def handle_connections_callback(self, message: str) -> None:
        contact_id = message["data"]["contactId"]
        if message["name"] == "ContactConnectedEvent":
            self._connections_callback(contact_id, True)
        elif message["name"] == "ContactDisconnectedEvent":
            self._connections_callback(contact_id, False)
        else:
            raise Exception(f"Wrong event in callback: {message['name']}")

    @staticmethod
    def _sort_contact_list(contacts: list) -> list:
        contacts.sort(key=itemgetter("lastChatActivity"),
                      reverse=True)
        return contacts