From 32a5e05fad03ce7fbfc84a6a619de33a6e6c624e Mon Sep 17 00:00:00 2001
From: Nico Alt <nicoalt@posteo.org>
Date: Fri, 12 Jun 2020 12:00:00 +0000
Subject: [PATCH] Full test coverage for Api and ApiThread
---
briar_wrapper/api.py | 61 ++-----
briar_wrapper/api_thread.py | 65 +++++++
tests/briar_wrapper/test_api.py | 151 +++++++---------
tests/briar_wrapper/test_api_thread.py | 234 +++++++++++++++++++++++++
tools/tests/test-pytest.sh | 2 +-
5 files changed, 378 insertions(+), 135 deletions(-)
create mode 100644 briar_wrapper/api_thread.py
create mode 100644 tests/briar_wrapper/test_api_thread.py
diff --git a/briar_wrapper/api.py b/briar_wrapper/api.py
index 8804ae6..9ab7457 100644
--- a/briar_wrapper/api.py
+++ b/briar_wrapper/api.py
@@ -3,13 +3,10 @@
# License-Filename: LICENSE.md
import os
-from subprocess import Popen, PIPE, STDOUT
from threading import Thread
-from time import sleep
-from urllib.error import HTTPError, URLError
-from urllib.request import urlopen
-from briar_wrapper.constants import BASE_HTTP_URL, BRIAR_AUTH_TOKEN, BRIAR_DB
+from briar_wrapper.api_thread import ApiThread
+from briar_wrapper.constants import BRIAR_AUTH_TOKEN, BRIAR_DB
from briar_wrapper.models.socket_listener import SocketListener
@@ -18,76 +15,44 @@ class Api:
auth_token = None
socket_listener = None
- _process = None
+ _api_thread = None
def __init__(self, headless_jar):
- self._command = ["java", "-jar", headless_jar]
+ self._api_thread = ApiThread(self, headless_jar)
@staticmethod
def has_account():
return os.path.isfile(BRIAR_DB)
def is_running(self):
- return (self._process is not None) and (self._process.poll() is None)
+ return self._api_thread.is_running()
def login(self, password, callback):
self._start_and_watch(callback)
- startup_thread = Thread(target=self._login, args=(password,),
- daemon=True)
+ startup_thread = Thread(target=self._api_thread.login,
+ args=(password,), daemon=True)
startup_thread.start()
def register(self, credentials, callback):
if len(credentials) != 2:
raise Exception("Can't process credentials")
self._start_and_watch(callback)
- startup_thread = Thread(target=self._register, args=(credentials,),
- daemon=True)
+ startup_thread = Thread(target=self._api_thread.register,
+ args=(credentials,), daemon=True)
startup_thread.start()
def stop(self):
- if not self.is_running():
- raise Exception("Nothing to stop")
- self._process.terminate()
+ self._api_thread.stop()
def _start_and_watch(self, callback):
- if self.is_running():
- raise Exception("API already running")
- self._process = Popen(self._command, stdin=PIPE,
- stdout=PIPE, stderr=STDOUT)
- watch_thread = Thread(target=self._watch_thread, args=(callback,),
- daemon=True)
- watch_thread.start()
+ self._api_thread.start()
+ self._api_thread.watch(callback)
- def _watch_thread(self, callback):
- while self.is_running():
- try:
- urlopen(BASE_HTTP_URL)
- sleep(0.1)
- except HTTPError as http_error:
- if http_error.code == 404:
- return self._on_successful_startup(callback)
- except URLError as url_error:
- if not isinstance(url_error.reason, ConnectionRefusedError):
- raise url_error
- callback(False)
-
- def _on_successful_startup(self, callback):
+ def on_successful_startup(self, callback):
self._load_auth_token()
self.socket_listener = SocketListener(self)
callback(True)
- def _login(self, password):
- if not self.is_running():
- raise Exception("Can't login; API not running")
- self._process.communicate((f"{password}\n").encode("utf-8"))
-
- def _register(self, credentials):
- if not self.is_running():
- raise Exception("Can't register; API not running")
- self._process.communicate((credentials[0] + '\n' +
- credentials[1] + '\n' +
- credentials[1] + '\n').encode("utf-8"))
-
def _load_auth_token(self):
if not Api.has_account():
raise Exception("Can't load authentication token")
diff --git a/briar_wrapper/api_thread.py b/briar_wrapper/api_thread.py
new file mode 100644
index 0000000..18b5928
--- /dev/null
+++ b/briar_wrapper/api_thread.py
@@ -0,0 +1,65 @@
+# Copyright (c) 2020 Nico Alt
+# SPDX-License-Identifier: AGPL-3.0-only
+# License-Filename: LICENSE.md
+
+from subprocess import Popen, PIPE, STDOUT
+from threading import Thread
+from time import sleep
+from urllib.error import HTTPError, URLError
+from urllib.request import urlopen
+
+from briar_wrapper.constants import BASE_HTTP_URL
+
+
+class ApiThread:
+
+ _api = None
+ _process = None
+
+ def __init__(self, api, headless_jar):
+ self._api = api
+ self._command = ["java", "-jar", headless_jar]
+
+ def is_running(self):
+ return (self._process is not None) and (self._process.poll() is None)
+
+ def start(self):
+ if self.is_running():
+ raise Exception("API already running")
+ self._process = Popen(self._command, stdin=PIPE,
+ stdout=PIPE, stderr=STDOUT)
+
+ def watch(self, callback):
+ watch_thread = Thread(target=self._watch_thread, args=(callback,),
+ daemon=True)
+ watch_thread.start()
+
+ def login(self, password):
+ if not self.is_running():
+ raise Exception("Can't login; API not running")
+ self._process.communicate((f"{password}\n").encode("utf-8"))
+
+ def register(self, credentials):
+ if not self.is_running():
+ raise Exception("Can't register; API not running")
+ self._process.communicate((credentials[0] + '\n' +
+ credentials[1] + '\n' +
+ credentials[1] + '\n').encode("utf-8"))
+
+ def stop(self):
+ if not self.is_running():
+ raise Exception("Nothing to stop")
+ self._process.terminate()
+
+ def _watch_thread(self, callback):
+ while self.is_running():
+ try:
+ urlopen(BASE_HTTP_URL)
+ sleep(0.1)
+ except HTTPError as http_error:
+ if http_error.code == 404:
+ return self._api.on_successful_startup(callback)
+ except URLError as url_error:
+ if not isinstance(url_error.reason, ConnectionRefusedError):
+ raise url_error
+ callback(False)
diff --git a/tests/briar_wrapper/test_api.py b/tests/briar_wrapper/test_api.py
index 3566e77..9cc7cae 100644
--- a/tests/briar_wrapper/test_api.py
+++ b/tests/briar_wrapper/test_api.py
@@ -3,10 +3,11 @@
# License-Filename: LICENSE.md
import pytest
-import subprocess
from briar_wrapper.api import Api
-from briar_wrapper.constants import BRIAR_DB
+from briar_wrapper.constants import BRIAR_AUTH_TOKEN, BRIAR_DB
+
+MODULE = "briar_wrapper.api.%s"
HEADLESS_JAR = 'briar-headless.jar'
PASSWORD = 'LjnM6/WPQ]V?@<=$'
@@ -15,154 +16,132 @@ CREDENTIALS = ('Alice', PASSWORD)
def test_has_account(mocker):
isfile_mock = mocker.patch('os.path.isfile')
- isfile_mock.return_value = False
+ isfile_mock.return_value = True
api = Api(HEADLESS_JAR)
- assert api.has_account() is False
+ assert api.has_account() is True
isfile_mock.assert_called_once_with(BRIAR_DB)
-def test_is_running(mocker, process):
- process.poll.return_value = None
+def test_is_running(api_thread):
+ api_thread.is_running.return_value = True
api = Api(HEADLESS_JAR)
- api._process = process
+ api._api_thread = api_thread
assert api.is_running() is True
+ api_thread.is_running.assert_called_once()
-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
+def test_is_not_running(api_thread):
+ api_thread.is_running.return_value = False
api = Api(HEADLESS_JAR)
- api._process = process
+ api._api_thread = api_thread
assert api.is_running() is False
+ api_thread.is_running.assert_called_once()
-def test_login(callback, start_and_watch, thread):
+def test_login_thread_start(api_thread):
api = Api(HEADLESS_JAR)
+ api._api_thread = api_thread
- api.login(PASSWORD, callback)
+ api.login(PASSWORD, None)
- start_and_watch.assert_called_once_with(callback)
- thread.assert_called_once_with(target=api._login, args=(PASSWORD,),
- daemon=True)
+ api_thread.start.assert_called_once()
-def test_login_already_running(callback, is_running, thread):
+def test_login_thread_watch(api_thread, callback):
api = Api(HEADLESS_JAR)
+ api._api_thread = api_thread
- with pytest.raises(Exception, match='API already running'):
- api.login(PASSWORD, callback)
-
+ 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
+ api_thread.watch.assert_called_once_with(callback)
-def test_login_communicate(callback, is_running, mocker,
- process, start_and_watch):
+def test_register_thread_start(api_thread):
api = Api(HEADLESS_JAR)
- api._process = process
+ api._api_thread = api_thread
- api.login(PASSWORD, callback)
+ api.register(CREDENTIALS, None)
- process.communicate.assert_called_once_with(
- (PASSWORD + "\n").encode("utf-8")
- )
+ api_thread.start.assert_called_once()
-def test_register(callback, start_and_watch, thread):
+def test_register_thread_watch(api_thread, callback):
api = Api(HEADLESS_JAR)
+ api._api_thread = api_thread
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)
+ api_thread.watch.assert_called_once_with(callback)
-def test_register_invalid_credentials(callback):
+def test_register_invalid_credentials():
api = Api(HEADLESS_JAR)
with pytest.raises(Exception, match="Can't process credentials"):
- api.register(PASSWORD, callback)
+ api.register(PASSWORD, None)
-def test_register_communicate(callback, is_running, mocker,
- process, start_and_watch):
+def test_stop(api_thread):
api = Api(HEADLESS_JAR)
- api._process = process
+ api._api_thread = api_thread
- api.register(CREDENTIALS, callback)
+ api.stop()
- process.communicate.assert_called_once_with(
- (CREDENTIALS[0] + '\n' +
- CREDENTIALS[1] + '\n' +
- CREDENTIALS[1] + '\n').encode("utf-8")
- )
+ api_thread.stop.assert_called_once()
-def test_stop(mocker, is_running, process):
+def test_successful_startup_socket_listener(callback, mocker):
+ socket_listener_mock = mocker.patch(MODULE % "SocketListener")
+ isfile_mock = mocker.patch(MODULE % "os.path.isfile")
+ isfile_mock.return_value = True
+ open_mock = mocker.patch("builtins.open")
api = Api(HEADLESS_JAR)
- api._process = process
- api.stop()
+ api.on_successful_startup(callback)
- api._process.terminate.assert_called_once()
+ socket_listener_mock.assert_called_once_with(api)
-def test_stop_not_running():
+def test_successful_startup_callback(callback, mocker):
+ isfile_mock = mocker.patch(MODULE % "os.path.isfile")
+ isfile_mock.return_value = True
+ open_mock = mocker.patch("builtins.open")
api = Api(HEADLESS_JAR)
- with pytest.raises(Exception, match='Nothing to stop'):
- api.stop()
+ api.on_successful_startup(callback)
+ callback.assert_called_once_with(True)
-def test_start_and_watch():
- # TODO: Various tests needed here, for both register and login
- pass
+def test_successful_startup_no_db(mocker):
+ isfile_mock = mocker.patch(MODULE % "os.path.isfile")
+ isfile_mock.return_value = False
+ api = Api(HEADLESS_JAR)
-@pytest.fixture
-def callback(mocker):
- return mocker.MagicMock()
+ with pytest.raises(Exception, match="Can't load authentication token"):
+ api.on_successful_startup(None)
-@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
+def test_successful_startup_open_auth_token(callback, mocker):
+ socket_listener_mock = mocker.patch(MODULE % "SocketListener")
+ isfile_mock = mocker.patch(MODULE % "os.path.isfile")
+ isfile_mock.return_value = True
+ open_mock = mocker.patch("builtins.open")
+ api = Api(HEADLESS_JAR)
+ api.on_successful_startup(callback)
-@pytest.fixture
-def process(mocker):
- return mocker.MagicMock()
+ open_mock.assert_called_once_with(BRIAR_AUTH_TOKEN, 'r')
@pytest.fixture
-def start_and_watch(mocker):
- return mocker.patch('briar_wrapper.api.Api._start_and_watch')
+def api_thread(mocker):
+ return mocker.patch('briar_wrapper.api.ApiThread')
@pytest.fixture
-def thread(mocker):
- return mocker.patch('briar_wrapper.api.Thread')
+def callback(mocker):
+ return mocker.MagicMock()
diff --git a/tests/briar_wrapper/test_api_thread.py b/tests/briar_wrapper/test_api_thread.py
new file mode 100644
index 0000000..59e11bb
--- /dev/null
+++ b/tests/briar_wrapper/test_api_thread.py
@@ -0,0 +1,234 @@
+# Copyright (c) 2019 Nico Alt
+# SPDX-License-Identifier: AGPL-3.0-only
+# License-Filename: LICENSE.md
+
+import pytest
+from urllib.error import HTTPError, URLError
+
+from briar_wrapper.api_thread import ApiThread
+
+MODULE = 'briar_wrapper.api_thread.%s'
+
+HEADLESS_JAR = 'briar-headless.jar'
+PASSWORD = 'LjnM6/WPQ]V?@<=$'
+CREDENTIALS = ('Alice', PASSWORD)
+
+
+def test_is_running(api, process):
+ process.poll.return_value = None
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._process = process
+
+ assert api_thread.is_running() is True
+
+
+def test_is_running_none(api):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._process = None
+
+ assert api_thread.is_running() is False
+
+
+def test_is_running_poll_none(api, process):
+ process.poll.return_value = 0
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._process = process
+
+ assert api_thread.is_running() is False
+
+
+def test_start_already_running(api, is_running):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ with pytest.raises(Exception, match='API already running'):
+ api_thread.start()
+
+
+def test_start_init_popen(api, mocker):
+ popen_mock = mocker.patch(MODULE % "Popen")
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ api_thread.start()
+
+ popen_mock.assert_called_once()
+
+
+def test_watch_already_running(api, is_running, mocker):
+ thread_mock = mocker.patch(MODULE % "Thread")
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ api_thread.watch(None)
+
+ thread_mock.assert_called_once()
+
+
+def test_watch_init_thread(api, mocker):
+ thread_mock = mocker.patch(MODULE % "Thread")
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ api_thread.watch(None)
+
+ thread_mock.assert_called_once()
+
+
+def test_login_not_running(api):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ with pytest.raises(Exception, match="Can't login; API not running"):
+ api_thread.login(PASSWORD)
+
+
+def test_login_communicate(api, is_running, process):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._process = process
+
+ api_thread.login(PASSWORD)
+
+ process.communicate.assert_called_once_with(
+ (PASSWORD + "\n").encode("utf-8")
+ )
+
+
+def test_register_not_running(api):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ with pytest.raises(Exception, match="Can't register; API not running"):
+ api_thread.register(CREDENTIALS)
+
+
+def test_register_communicate(api, is_running, process):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._process = process
+
+ api_thread.register(CREDENTIALS)
+
+ process.communicate.assert_called_once_with(
+ (CREDENTIALS[0] + '\n' +
+ CREDENTIALS[1] + '\n' +
+ CREDENTIALS[1] + '\n').encode("utf-8")
+ )
+
+
+def test_stop(api, is_running, process):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._process = process
+
+ api_thread.stop()
+
+ process.terminate.assert_called_once()
+
+
+def test_stop_not_running(api):
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ with pytest.raises(Exception, match='Nothing to stop'):
+ api_thread.stop()
+
+
+def test_watch_thread(api, mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.side_effect = [True, False]
+
+ urlopen_mock = mocker.patch(MODULE % "urlopen")
+ sleep_mock = mocker.patch(MODULE % "sleep")
+
+ callback_mock = mocker.MagicMock()
+
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._watch_thread(callback_mock)
+
+ assert is_running_mock.called is True
+ urlopen_mock.assert_called_once_with("http://localhost:7000/v1/")
+ callback_mock.assert_called_once_with(False)
+
+
+def test_watch_thread_sleep(api, mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.side_effect = [True, False]
+
+ urlopen_mock = mocker.patch(MODULE % "urlopen")
+ sleep_mock = mocker.patch(MODULE % "sleep")
+
+ callback_mock = mocker.MagicMock()
+
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._watch_thread(callback_mock)
+
+ sleep_mock.assert_called_once_with(0.1)
+
+
+def test_watch_thread_not_running(api, mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.return_value = False
+
+ urlopen_mock = mocker.patch(MODULE % "urlopen")
+
+ callback_mock = mocker.MagicMock()
+
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._watch_thread(callback_mock)
+
+ is_running_mock.assert_called_once()
+ assert urlopen_mock.called is False
+ callback_mock.assert_called_once_with(False)
+
+
+def test_watch_thread_404(api, mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.return_value = True
+
+ urlopen_mock = mocker.patch(MODULE % "urlopen")
+ urlopen_mock.side_effect = HTTPError(code=404, msg=None, hdrs=None,
+ fp=None, url=None)
+
+ callback_mock = mocker.MagicMock()
+
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._watch_thread(callback_mock)
+
+ api.on_successful_startup.assert_called_once_with(callback_mock)
+
+
+def test_watch_thread_connection_refused(api, mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.side_effect = [True, False]
+
+ urlopen_mock = mocker.patch(MODULE % "urlopen")
+ urlopen_mock.side_effect = URLError(reason=ConnectionRefusedError())
+
+ callback_mock = mocker.MagicMock()
+
+ api_thread = ApiThread(api, HEADLESS_JAR)
+ api_thread._watch_thread(callback_mock)
+
+ callback_mock.assert_called_once_with(False)
+
+
+def test_watch_thread_url_error(api, mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.return_value = True
+
+ urlopen_mock = mocker.patch(MODULE % "urlopen")
+ urlopen_mock.side_effect = URLError(reason=None)
+
+ api_thread = ApiThread(api, HEADLESS_JAR)
+
+ with pytest.raises(URLError):
+ api_thread._watch_thread(None)
+
+
+@pytest.fixture
+def api(mocker):
+ return mocker.MagicMock()
+
+
+@pytest.fixture
+def is_running(mocker):
+ is_running_mock = mocker.patch(MODULE % 'ApiThread.is_running')
+ is_running_mock.return_value = True
+ return is_running_mock
+
+
+@pytest.fixture
+def process(mocker):
+ return mocker.MagicMock()
diff --git a/tools/tests/test-pytest.sh b/tools/tests/test-pytest.sh
index d175030..d7bfaad 100755
--- a/tools/tests/test-pytest.sh
+++ b/tools/tests/test-pytest.sh
@@ -3,4 +3,4 @@
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md
-PYTHONPATH=briar_wrapper pytest --cov=briar_wrapper tests/
+PYTHONPATH=briar_wrapper pytest --cov-report term-missing --cov=briar_wrapper tests/
--
GitLab