diff --git a/README.md b/README.md
index 2397319a6..ec5aadaaa 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 b11b41a9c..000000000
--- 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 8f0db9c1b..07f4c0a98 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 000000000..36337bfe7
--- /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 2075624ee..06125d658 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 89ba703b4..bd39a2b0b 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 4291ddf4f..7dea58ea5 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 8c8c0c468..3d4bc8f08 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 1e892efde..800608405 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 000000000..c42e0e2ca
--- /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