From 49ad3261f16263d0e0319b713c7260103c9a962f Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Sat, 6 Mar 2021 16:21:28 +0100 Subject: [PATCH] Refactored tests to use pytest with fixtures instead of unittest.TestCase --- README.md | 2 +- run_tests.sh | 22 --- tests/__init__.py | 131 ------------------ tests/conftest.py | 45 ++++++ ...test_http_config.yaml => config_test.yaml} | 3 + .../test_procedure.yaml} | 7 - tests/test_event_parse.py | 44 +++--- tests/test_http.py | 99 ++++++------- tests/test_procedure.py | 89 ++++++------ tests/utils.py | 93 +++++++++++++ 10 files changed, 257 insertions(+), 278 deletions(-) delete mode 100755 run_tests.sh create mode 100644 tests/conftest.py rename tests/etc/{test_http_config.yaml => config_test.yaml} (77%) rename tests/etc/{test_procedure_config.yaml => include/test_procedure.yaml} (74%) create mode 100644 tests/utils.py diff --git a/README.md b/README.md index 2397319a..ec5aadaa 100644 --- a/README.md +++ b/README.md @@ -437,7 +437,7 @@ platyvdock rm device_id ## Tests -To run the tests run the `run_tests.sh` from the source root folder - no further configuration is required. +To run the tests simply run `pytest` either from the project root folder or the `tests/` folder. --- diff --git a/run_tests.sh b/run_tests.sh deleted file mode 100755 index b11b41a9..00000000 --- a/run_tests.sh +++ /dev/null @@ -1,22 +0,0 @@ -#!/bin/bash - -PYTHON=python -tests_ret=0 - -for testcase in tests/test_*.py -do - $PYTHON -m unittest $testcase - test_ret=$? - - if [[ $test_ret != 0 ]]; then - tests_ret=$test_ret - echo "-------------" >&2 - echo "FAILED: $testcase" >&2 - echo "-------------" >&2 - fi -done - -exit $tests_ret - - -# vim:sw=4:ts=4:et: diff --git a/tests/__init__.py b/tests/__init__.py index 8f0db9c1..07f4c0a9 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,138 +1,7 @@ -import abc import logging import os -import requests import sys -import time -import unittest -from threading import Thread -from typing import Optional logging.basicConfig(level=logging.INFO, stream=sys.stdout) test_dir = os.path.abspath(os.path.dirname(__file__)) -conf_dir = os.path.join(test_dir, 'etc') sys.path.insert(0, os.path.abspath(os.path.join(test_dir, '..'))) - -from platypush import Daemon, Config, Response -from platypush.message import Message -from platypush.utils import set_timeout, clear_timeout - - -class TimeoutException(RuntimeError): - def __init__(self, msg): - self.msg = msg - - -class BaseTest(unittest.TestCase, abc.ABC): - """ - Base class for Platypush tests. - """ - - app_start_timeout = 5 - request_timeout = 10 - config_file = None - db_file = None - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.app: Optional[Daemon] = None - - def setUp(self) -> None: - self.start_daemon() - - def tearDown(self): - try: - self.stop_daemon() - finally: - if self.db_file and os.path.isfile(self.db_file): - logging.info('Removing temporary db file {}'.format(self.db_file)) - os.unlink(self.db_file) - - def start_daemon(self): - logging.info('Starting platypush service') - self.app = Daemon(config_file=self.config_file) - Thread(target=lambda: self.app.run()).start() - logging.info('Sleeping {} seconds while waiting for the daemon to start up'.format(self.app_start_timeout)) - time.sleep(self.app_start_timeout) - - def stop_daemon(self): - if self.app: - logging.info('Stopping platypush service') - self.app.stop_app() - - @staticmethod - def parse_response(response): - response = Message.build(response.json()) - assert isinstance(response, Response), 'Expected Response type, got {}'.format(response.__class__.__name__) - return response - - @staticmethod - def on_timeout(msg): - def _f(): raise TimeoutException(msg) - return _f - - -class BaseHttpTest(BaseTest, abc.ABC): - """ - Base class for Platypush HTTP tests. - """ - - base_url = None - test_user = 'platypush' - test_pass = 'test' - - def setUp(self) -> None: - Config.init(self.config_file) - backends = Config.get_backends() - self.assertTrue('http' in backends, 'Missing HTTP server configuration') - self.base_url = 'http://localhost:{port}'.format(port=backends['http']['port']) - self.db_file = Config.get('main.db')['engine'][len('sqlite:///'):] - super().setUp() - - def register_user(self, username: Optional[str] = None, password: Optional[str] = None): - if not username: - username = self.test_user - password = self.test_pass - - set_timeout(seconds=self.request_timeout, on_timeout=self.on_timeout('User registration response timed out')) - response = requests.post('{base_url}/register?redirect={base_url}/'.format(base_url=self.base_url), data={ - 'username': username, - 'password': password, - 'confirm_password': password, - }) - - clear_timeout() - return response - - def send_request(self, action: str, timeout: Optional[float] = None, args: Optional[dict] = None, - parse_response: bool = True, authenticate: bool = True, **kwargs): - if not timeout: - timeout = self.request_timeout - if not args: - args = {} - - auth = (self.test_user, self.test_pass) if authenticate else kwargs.pop('auth', ()) - set_timeout(seconds=timeout, on_timeout=self.on_timeout('Receiver response timed out')) - response = requests.post( - '{}/execute'.format(self.base_url), - auth=auth, - json={ - 'type': 'request', - 'action': action, - 'args': args, - }, **kwargs - ) - - clear_timeout() - - if parse_response: - response = self.parse_response(response) - return response - - def assertEqual(self, first, second, msg=..., expected=None, actual=None) -> None: - if expected is not None and actual is not None: - if not msg: - msg = '' - msg += '\n\tExpected: {expected}\n\tActual: {actual}'.format(expected=expected, actual=actual) - - super().assertEqual(first, second, msg) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..36337bfe --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,45 @@ +import logging +import os +import pytest +import time +from threading import Thread + +from platypush import Daemon, Config + +from .utils import config_file, set_base_url + +app_start_timeout = 5 + + +@pytest.fixture(scope='session', autouse=True) +def app(): + logging.info('Starting Platypush test service') + + Config.init(config_file) + app = Daemon(config_file=config_file) + Thread(target=lambda: app.run()).start() + logging.info('Sleeping {} seconds while waiting for the daemon to start up'.format(app_start_timeout)) + time.sleep(app_start_timeout) + yield app + + logging.info('Stopping Platypush test service') + app.stop_app() + db_file = (Config.get('main.db') or {}).get('engine', '')[len('sqlite:///'):] + + if db_file and os.path.isfile(db_file): + logging.info('Removing temporary db file {}'.format(db_file)) + os.unlink(db_file) + + +@pytest.fixture(scope='session') +def db_file(): + yield Config.get('main.db')['engine'][len('sqlite:///'):] + + +@pytest.fixture(scope='session') +def base_url(): + backends = Config.get_backends() + assert 'http' in backends, 'Missing HTTP server configuration' + url = 'http://localhost:{port}'.format(port=backends['http']['port']) + set_base_url(url) + yield url diff --git a/tests/etc/test_http_config.yaml b/tests/etc/config_test.yaml similarity index 77% rename from tests/etc/test_http_config.yaml rename to tests/etc/config_test.yaml index 2075624e..06125d65 100644 --- a/tests/etc/test_http_config.yaml +++ b/tests/etc/config_test.yaml @@ -1,3 +1,6 @@ +include: + - include/test_procedure.yaml + main.db: engine: sqlite:////tmp/platypush-test-http.db diff --git a/tests/etc/test_procedure_config.yaml b/tests/etc/include/test_procedure.yaml similarity index 74% rename from tests/etc/test_procedure_config.yaml rename to tests/etc/include/test_procedure.yaml index 89ba703b..bd39a2b0 100644 --- a/tests/etc/test_procedure_config.yaml +++ b/tests/etc/include/test_procedure.yaml @@ -1,10 +1,3 @@ -main.db: - engine: sqlite:////tmp/platypush-test-procedure.db - -backend.http: - port: 8124 - disable_websocket: True - procedure.write_file: - action: file.write args: diff --git a/tests/test_event_parse.py b/tests/test_event_parse.py index 4291ddf4..7dea58ea 100644 --- a/tests/test_event_parse.py +++ b/tests/test_event_parse.py @@ -1,34 +1,28 @@ -import os -import unittest - from platypush.event.hook import EventCondition from platypush.message.event.ping import PingEvent -from . import BaseTest, conf_dir + +condition = EventCondition.build({ + 'type': 'platypush.message.event.ping.PingEvent', + 'message': 'This is (the)? answer: ${answer}' +}) -class TestEventParse(BaseTest): - config_file = os.path.join(conf_dir, 'test_http_config.yaml') - condition = EventCondition.build({ - 'type': 'platypush.message.event.ping.PingEvent', - 'message': 'This is (the)? answer: ${answer}' - }) +def test_event_parse(): + """ + Test for the events/conditions matching logic. + """ + message = "GARBAGE GARBAGE this is the answer: 42" + event = PingEvent(message=message) + result = event.matches_condition(condition) + assert result.is_match + assert 'answer' in result.parsed_args + assert result.parsed_args['answer'] == '42' - def test_event_parse(self): - message = "GARBAGE GARBAGE this is the answer: 42" - event = PingEvent(message=message) - result = event.matches_condition(self.condition) - self.assertTrue(result.is_match) - self.assertTrue('answer' in result.parsed_args) - self.assertEqual(result.parsed_args['answer'], '42') + message = "what is not the answer? 43" + event = PingEvent(message=message) + result = event.matches_condition(condition) + assert not result.is_match - message = "what is not the answer? 43" - event = PingEvent(message=message) - result = event.matches_condition(self.condition) - self.assertFalse(result.is_match) - - -if __name__ == '__main__': - unittest.main() # vim:sw=4:ts=4:et: diff --git a/tests/test_http.py b/tests/test_http.py index 8c8c0c46..3d4bc8f0 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -1,57 +1,62 @@ -import os -import unittest +import pytest -from . import BaseHttpTest, conf_dir +from .utils import register_user, send_request as _send_request -class TestHttp(BaseHttpTest): +@pytest.fixture(scope='module') +def expected_registration_redirect(base_url): + yield '{base_url}/register?redirect={base_url}/execute'.format(base_url=base_url) + + +@pytest.fixture(scope='module') +def expected_login_redirect(base_url): + yield '{base_url}/login?redirect={base_url}/execute'.format(base_url=base_url) + + +def send_request(**kwargs): + return _send_request('shell.exec', args={'cmd': 'echo ping'}, **kwargs) + + +def test_request_with_no_registered_users(base_url, expected_registration_redirect): """ - Tests the full flow of a request/response on the HTTP backend. - Runs a remote command over HTTP via shell.exec plugin and gets the output. + An /execute request performed before any user is registered should redirect to the registration page. """ - - config_file = os.path.join(conf_dir, 'test_http_config.yaml') - - def __init__(self, *args, **kwargs): - super(TestHttp, self).__init__(*args, **kwargs) - - def test_http_flow(self): - # An /execute request performed before any user is registered should redirect to the registration page. - expected_registration_redirect = '{base_url}/register?redirect={base_url}/execute'.format( - base_url=self.base_url) - expected_login_redirect = '{base_url}/login?redirect={base_url}/execute'.format( - base_url=self.base_url) - - response = self.send_request(authenticate=False, parse_response=False) - self.assertEqual(expected_registration_redirect, response.url, - 'No users registered, but the application did not redirect us to the registration page') - - # Emulate a first user registration through form and get the session_token. - response = self.register_user() - self.assertGreater(len(response.history), 0, 'Redirect missing from the history') - self.assertTrue('session_token' in response.history[0].cookies, 'No session_token returned upon registration') - self.assertEqual('{base_url}/'.format(base_url=self.base_url), response.url, - 'The registration form did not redirect to the main panel') - - # After a first user has been registered any unauthenticated call to /execute should redirect to /login. - response = self.send_request(authenticate=False, parse_response=False) - self.assertEqual(expected_login_redirect, response.url, - 'An unauthenticated request after user registration should result in a login redirect') - - # A request authenticated with user/pass should succeed. - response = self.send_request(authenticate=True) - self.assertEqual(response.output.strip(), 'ping', 'The request did not return the expected output') - - # A request with the wrong user/pass should fail. - response = self.send_request(authenticate=False, auth=('wrong', 'wrong'), parse_response=False) - self.assertEqual(expected_login_redirect, response.url, 'A request with wrong credentials should fail') - - def send_request(self, **kwargs): - return super().send_request('shell.exec', args={'cmd': 'echo ping'}, **kwargs) + response = send_request(authenticate=False, parse_json=False) + assert expected_registration_redirect == response.url, \ + 'No users registered, but the application did not redirect us to the registration page' -if __name__ == '__main__': - unittest.main() +def test_first_user_registration(base_url): + """ + Emulate a first user registration through form and get the session_token. + """ + response = register_user() + + assert len(response.history) > 0, 'Redirect missing from the history' + assert 'session_token' in response.history[0].cookies, 'No session_token returned upon registration' + assert '{base_url}/'.format(base_url=base_url) == response.url, \ + 'The registration form did not redirect to the main panel' + + +def test_unauthorized_request_with_registered_user(base_url, expected_login_redirect): + """ + After a first user has been registered any unauthenticated call to /execute should redirect to /login. + """ + response = send_request(authenticate=False, parse_json=False) + assert expected_login_redirect == response.url, \ + 'An unauthenticated request after user registration should result in a login redirect' + + +def test_authorized_request_with_registered_user(base_url): + # A request authenticated with user/pass should succeed. + response = send_request(authenticate=True) + assert response.output.strip() == 'ping', 'The request did not return the expected output' + + +def test_request_with_wrong_credentials(base_url, expected_login_redirect): + # A request with the wrong user/pass should fail. + response = send_request(authenticate=False, auth=('wrong', 'wrong'), parse_json=False) + assert expected_login_redirect == response.url, 'A request with wrong credentials should fail' # vim:sw=4:ts=4:et: diff --git a/tests/test_procedure.py b/tests/test_procedure.py index 1e892efd..80060840 100644 --- a/tests/test_procedure.py +++ b/tests/test_procedure.py @@ -1,60 +1,59 @@ import os +import pytest import tempfile import time -import unittest from platypush.message.event.custom import CustomEvent -from . import BaseHttpTest, conf_dir + +from .utils import register_user, send_request -@unittest.skip('Skipped until I can find a way to properly clean up the environment from the previous tests and start ' - 'a new platform') -class TestProcedure(BaseHttpTest): +@pytest.fixture(scope='module', autouse=True) +def user(*_): + register_user() + + +@pytest.fixture(scope='module') +def tmp_file(*_): + tmp_file = tempfile.NamedTemporaryFile(prefix='platypush-test-procedure-', suffix='.txt', delete=False) + yield tmp_file + if os.path.isfile(tmp_file.name): + os.unlink(tmp_file.name) + + +def check_file_content(expected_content: str, tmp_file): + assert os.path.isfile(tmp_file.name), 'The expected output file was not created' + with open(tmp_file.name, 'r') as f: + content = f.read() + + assert content == expected_content, 'The output file did not contain the expected text' + + +def test_procedure_call(tmp_file): """ - Test the execution of configured procedures. + Test the result of a procedure invoked directly over HTTP. """ + output_text = 'Procedure test' + send_request( + action='procedure.write_file', + args={ + 'file': tmp_file.name, + 'content': output_text, + }) - config_file = os.path.join(conf_dir, 'test_procedure_config.yaml') - - def setUp(self) -> None: - super().setUp() - self.register_user() - self.tmp_file = tempfile.NamedTemporaryFile(prefix='platypush-test-procedure-', suffix='.txt', delete=False) - - def tearDown(self): - if os.path.isfile(self.tmp_file.name): - os.unlink(self.tmp_file.name) - super().tearDown() - - def check_file_content(self, expected_content: str): - self.assertTrue(os.path.isfile(self.tmp_file.name), 'The expected output file was not created') - with open(self.tmp_file.name, 'r') as f: - content = f.read() - - self.assertEqual(content, expected_content, 'The output file did not contain the expected text', - expected=expected_content, actual=content) - - def test_procedure_call(self): - output_text = 'Procedure test' - self.send_request( - action='procedure.write_file', - args={ - 'file': self.tmp_file.name, - 'content': output_text, - }) - - self.check_file_content(expected_content=output_text) - - def test_procedure_from_event(self): - output_text = 'Procedure from event test' - event_type = 'platypush_test_procedure_from_event' - self.app.bus.post(CustomEvent(subtype=event_type, file=self.tmp_file.name, content=output_text)) - time.sleep(3) - self.check_file_content(output_text) + check_file_content(expected_content=output_text, tmp_file=tmp_file) -if __name__ == '__main__': - unittest.main() +def test_procedure_from_event(app, tmp_file): + """ + Test the result of a procedure triggered by an event. + """ + output_text = 'Procedure from event test' + event_type = 'platypush_test_procedure_from_event' + # noinspection PyUnresolvedReferences + app.bus.post(CustomEvent(subtype=event_type, file=tmp_file.name, content=output_text)) + time.sleep(2) + check_file_content(expected_content=output_text, tmp_file=tmp_file) # vim:sw=4:ts=4:et: diff --git a/tests/utils.py b/tests/utils.py new file mode 100644 index 00000000..c42e0e2c --- /dev/null +++ b/tests/utils.py @@ -0,0 +1,93 @@ +import os +import requests +from typing import Optional + +from platypush.message import Message +from platypush.message.response import Response +from platypush.utils import set_timeout, clear_timeout + +from . import test_dir + +# Default configuration folder for tests +conf_dir = os.path.join(test_dir, 'etc') + +# Default configuration file for tests +config_file = os.path.join(conf_dir, 'config_test.yaml') + +# Default request timeout in seconds +request_timeout = 10 + +# Default test user +test_user = 'platypush' + +# Default test password +test_pass = 'test' + +# Base URL +base_url = 'http://localhost:8123' + + +def set_base_url(url: str): + global base_url + base_url = url + + +class TimeoutException(RuntimeError): + """ + Exception raised in case of timeout. + """ + def __init__(self, msg: str = 'Timeout'): + self.msg = msg + + +def send_request(action: str, timeout: Optional[float] = None, args: Optional[dict] = None, + parse_json: bool = True, authenticate: bool = True, **kwargs): + if not timeout: + timeout = request_timeout + if not args: + args = {} + + auth = (test_user, test_pass) if authenticate else kwargs.pop('auth', ()) + set_timeout(seconds=timeout, on_timeout=on_timeout('Receiver response timed out')) + response = requests.post( + '{}/execute'.format(base_url), + auth=auth, + json={ + 'type': 'request', + 'action': action, + 'args': args, + }, **kwargs + ) + + clear_timeout() + + if parse_json: + response = parse_response(response) + return response + + +def register_user(username: Optional[str] = None, password: Optional[str] = None): + if not username: + username = test_user + password = test_pass + + set_timeout(seconds=request_timeout, on_timeout=on_timeout('User registration response timed out')) + response = requests.post('{base_url}/register?redirect={base_url}/'.format(base_url=base_url), data={ + 'username': username, + 'password': password, + 'confirm_password': password, + }) + + clear_timeout() + return response + + +def on_timeout(msg): + def _f(): raise TimeoutException(msg) + return _f + + +def parse_response(response): + response = Message.build(response.json()) + assert isinstance(response, Response), 'Expected Response type, got {}'.format(response.__class__.__name__) + return response