Psion-fanatic/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

Refactored tests to use pytest with fixtures instead of unittest.TestCase

Browse source
This commit is contained in:
Fabio Manganiello 2021-03-06 16:21:28 +01:00
parent 73e16fa6b1
commit 49ad3261f1
10 changed files with 257 additions and 278 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -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.
---

22
run_tests.sh
View file

@ -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:

131
tests/__init__.py
View file

@ -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)

45
tests/conftest.py Normal file
View file

@ -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

3
tests/etc/test_http_config.yaml → tests/etc/config_test.yaml
View file

@ -1,3 +1,6 @@
include:
- include/test_procedure.yaml
main.db:
engine: sqlite:////tmp/platypush-test-http.db

7
tests/etc/test_procedure_config.yaml → tests/etc/include/test_procedure.yaml
View file

@ -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:

44
tests/test_event_parse.py
View file

@ -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:

99
tests/test_http.py
View file

@ -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:

89
tests/test_procedure.py
View file

@ -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:

93
tests/utils.py Normal file
View file

@ -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