Add API tests

briar_wrapper/api.py

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 # License-Filename: LICENSE.md
 
-from os.path import isfile
+import os
 from subprocess import Popen, PIPE, STDOUT
 from threading import Thread
 from time import sleep
@@ -25,7 +25,7 @@ class Api:
 
     @staticmethod
     def has_account():
-        return isfile(BRIAR_DB)
+        return os.path.isfile(BRIAR_DB)
 
     def is_running(self):
         return (self._process is not None) and (self._process.poll() is None)
@@ -79,7 +79,7 @@ class Api:
     def _login(self, password):
         if not self.is_running():
             raise Exception("Can't login; API not running")
-        self._process.communicate(("%s\n" % password).encode("utf-8"))
+        self._process.communicate((f"{password}\n").encode("utf-8"))
 
     def _register(self, credentials):
         if not self.is_running():

tests/briar_wrapper/test_api.py

+# Copyright (c) 2019 Nico Alt
+# SPDX-License-Identifier: AGPL-3.0-only
+# License-Filename: LICENSE.md
+
+import pytest
+import subprocess
+
+from briar_wrapper.api import Api
+from briar_wrapper.constants import BRIAR_DB
+
+HEADLESS_JAR = 'briar-headless.jar'
+PASSWORD = 'LjnM6/WPQ]V?@<=$'
+CREDENTIALS = ('Alice', PASSWORD)
+
+
+def test_has_account(mocker):
+    isfile_mock = mocker.patch('os.path.isfile')
+    isfile_mock.return_value = False
+    api = Api(HEADLESS_JAR)
+
+    assert api.has_account() is False
+    isfile_mock.assert_called_once_with(BRIAR_DB)
+
+
+def test_is_running(mocker, process):
+    process.poll.return_value = None
+    api = Api(HEADLESS_JAR)
+    api._process = process
+
+    assert api.is_running() is True
+
+
+def test_is_running_none():
+    api = Api(HEADLESS_JAR)
+    api._process = None
+
+    assert api.is_running() is False
+
+
+def test_is_running_poll_none(mocker, process):
+    process.poll.return_value = 0
+    api = Api(HEADLESS_JAR)
+    api._process = process
+
+    assert api.is_running() is False
+
+
+def test_login(callback, start_and_watch, thread):
+    api = Api(HEADLESS_JAR)
+
+    api.login(PASSWORD, callback)
+
+    start_and_watch.assert_called_once_with(callback)
+    thread.assert_called_once_with(target=api._login, args=(PASSWORD,),
+                                   daemon=True)
+
+
+def test_login_already_running(callback, is_running, thread):
+    api = Api(HEADLESS_JAR)
+
+    with pytest.raises(Exception, match='API already running'):
+        api.login(PASSWORD, callback)
+
+
+def test_login_not_running():
+    # TODO: Write test for failed login due to API not running
+    # Not easy to test because exception is thrown in Thread
+    pass
+
+
+def test_login_communicate(callback, is_running, mocker,
+                           process, start_and_watch):
+    api = Api(HEADLESS_JAR)
+    api._process = process
+
+    api.login(PASSWORD, callback)
+
+    process.communicate.assert_called_once_with(
+        (PASSWORD + "\n").encode("utf-8")
+    )
+
+
+def test_register(callback, start_and_watch, thread):
+    api = Api(HEADLESS_JAR)
+
+    api.register(CREDENTIALS, callback)
+
+    start_and_watch.assert_called_once_with(callback)
+    thread.assert_called_once_with(target=api._register, args=(CREDENTIALS,),
+                                   daemon=True)
+
+
+def test_register_already_running(callback, is_running, thread):
+    api = Api(HEADLESS_JAR)
+
+    with pytest.raises(Exception, match='API already running'):
+        api.register(CREDENTIALS, callback)
+
+
+def test_register_invalid_credentials(callback):
+    api = Api(HEADLESS_JAR)
+
+    with pytest.raises(Exception, match="Can't process credentials"):
+        api.register(PASSWORD, callback)
+
+
+def test_register_communicate(callback, is_running, mocker,
+                              process, start_and_watch):
+    api = Api(HEADLESS_JAR)
+    api._process = process
+
+    api.register(CREDENTIALS, callback)
+
+    process.communicate.assert_called_once_with(
+        (CREDENTIALS[0] + '\n' +
+         CREDENTIALS[1] + '\n' +
+         CREDENTIALS[1] + '\n').encode("utf-8")
+    )
+
+
+def test_stop(mocker, is_running, process):
+    api = Api(HEADLESS_JAR)
+    api._process = process
+
+    api.stop()
+
+    api._process.terminate.assert_called_once()
+
+
+def test_stop_not_running():
+    api = Api(HEADLESS_JAR)
+
+    with pytest.raises(Exception, match='Nothing to stop'):
+        api.stop()
+
+
+def test_start_and_watch():
+    # TODO: Various tests needed here, for both register and login
+    pass
+
+
+@pytest.fixture
+def callback(mocker):
+    return mocker.MagicMock()
+
+
+@pytest.fixture
+def is_running(mocker):
+    is_running_mock = mocker.patch(
+        'briar_wrapper.api.Api.is_running'
+        )
+    is_running_mock.return_value = True
+    return is_running_mock
+
+
+@pytest.fixture
+def process(mocker):
+    return mocker.MagicMock()
+
+
+@pytest.fixture
+def start_and_watch(mocker):
+    return mocker.patch('briar_wrapper.api.Api._start_and_watch')
+
+
+@pytest.fixture
+def thread(mocker):
+    return mocker.patch('briar_wrapper.api.Thread')

tests/conftest.py

@@ -2,16 +2,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 # License-Filename: LICENSE.md
 
-from unittest import mock
-
 import pytest
 
 # pylint: disable=redefined-outer-name
 
 
 @pytest.fixture
-def api(auth_token):
-    api = mock.Mock()
+def api(auth_token, mocker):
+    api = mocker.Mock()
     api.auth_token = auth_token
     return api