...
 
Commits (4)
......@@ -54,7 +54,7 @@ class Api:
return self.notifications.get_all_rendered(media_type, lang)
def get_rendered_notification(self, msgid, media_type, lang, force_media_type=False):
def get_rendered_notification(self, msgid, media_type='plain', lang='en', force_media_type=False):
"""Get rendered notification of specific media type by id"""
self.notifications.delete_invalid_messages()
......
......@@ -23,14 +23,14 @@ class Supervisor:
return
except OSError as e:
logger.error("fork #1 failed: %d (%s)", e.errno, e.strerror)
sys.exit(1)
os._exit(1)
os.setsid()
try:
pid = os.fork()
if pid > 0:
sys.exit(0)
os._exit(0)
except OSError as e:
logger.error("fork #2 failed: %d (%s)", e.errno, e.strerror)
sys.exit(1)
......@@ -43,7 +43,14 @@ class Supervisor:
logger.error("stdout: %s", self.process.stdout.readline())
logger.error("stderr: %s", self.process.stderr.readline())
sys.exit(0)
# dettach stdin/out/err and close them
devnull = os.open(os.devnull, os.O_WRONLY)
os.dup2(devnull, 0)
os.dup2(devnull, 1)
os.dup2(devnull, 2)
os.close(devnull)
os._exit(0)
def run_proc(self):
try:
......
......@@ -2,25 +2,41 @@ import json
import pytest
from notifylib.api import Api
from notifylib.exceptions import (
MediaTypeNotAvailableException,
NoSuchNotificationException,
NotificationNotDismissibleException,
)
@pytest.fixture
def volatile_dir(tmpdir):
return tmpdir.mkdir('notify').mkdir('volatile')
def tmp_dir(tmpdir):
return tmpdir.mkdir('notify')
@pytest.fixture
def config(volatile_dir):
def volatile_dir(tmp_dir):
return tmp_dir.mkdir('volatile')
@pytest.fixture
def persistent_dir(tmp_dir):
return tmp_dir.mkdir('persistent')
@pytest.fixture
def config(volatile_dir, persistent_dir):
return {
'settings': {
'volatile_dir': volatile_dir,
'persistent_dir': persistent_dir,
}
}
@pytest.mark.xfail
def test_create_notification(tmpdir, volatile_dir, config):
content = '''{
@pytest.fixture
def persistent_notification():
return """{
"notif_id": "8c3ad32ca1774eed985604cc1a328773",
"api_version": 1,
"timestamp": 1549381864,
......@@ -59,30 +75,14 @@ def test_create_notification(tmpdir, volatile_dir, config):
"html": "\\n<h3>Simple message</h3>\\n<p>\\n HTML rendered message: egg egg spam\\n</p>\\n"
},
"valid": true,
"explicit_dismiss": true,
"explicit_dismiss": false,
"default_action": "dismiss"
}'''
# TODO: figure out what to do with timestamp and notification id
api = Api(confdict=config)
opts = {
'skel_id': 'simple.simple',
'data': json.loads('{"message": "egg spam spam"}'),
}
}"""
nid = api.create(**opts)
assert nid.isalnum() and len(nid) == 32
f = volatile_dir.join("{}.json".format(nid))
result = f.read()
assert result == content
def test_get_notification(tmpdir, volatile_dir, config):
content = '''{
@pytest.fixture
def dummy_notification():
return """{
"notif_id": "8c3ad32ca1774eed985604cc1a328773",
"api_version": 1,
"timestamp": 1549381864,
......@@ -123,13 +123,35 @@ def test_get_notification(tmpdir, volatile_dir, config):
"valid": true,
"explicit_dismiss": true,
"default_action": "dismiss"
}'''
}"""
@pytest.mark.xfail
def test_create_notification(tmpdir, volatile_dir, config, dummy_notification):
# TODO: figure out what to do with timestamp and notification id
api = Api(confdict=config)
opts = {
'skel_id': 'simple.simple',
'data': json.loads('{"message": "egg spam spam"}'),
}
nid = api.create(**opts)
assert nid.isalnum() and len(nid) == 32
f = volatile_dir.join("{}.json".format(nid))
result = f.read()
assert result == dummy_notification
def test_get_notification(volatile_dir, config, dummy_notification):
"""Test notification lookup with full lenght id"""
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(content)
f.write(dummy_notification)
api = Api(confdict=config)
n = api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773', 'plain', 'en')
n = api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773')
assert isinstance(n, dict)
......@@ -142,3 +164,129 @@ def test_get_notification(tmpdir, volatile_dir, config):
assert n['message'] == '\n= Simple message =\nMessage: egg egg spam\n'
assert set(('dummy', 'dismiss')).issubset(n['actions'])
assert set(('persistent', 'timestamp', 'severity', 'default_action')).issubset(n['metadata'])
def test_get_notification_shortid(volatile_dir, config, dummy_notification):
"""Test notification lookup with short id"""
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
n = api.get_rendered_notification('8c3ad32c')
assert isinstance(n, dict)
# TODO: check json schema?
assert 'message' in n
assert 'actions' in n
assert 'metadata' in n
assert n['message'] == '\n= Simple message =\nMessage: egg egg spam\n'
assert set(('dummy', 'dismiss')).issubset(n['actions'])
assert set(('persistent', 'timestamp', 'severity', 'default_action')).issubset(n['metadata'])
def test_get_notification_try_media_type(volatile_dir, config, dummy_notification):
"""Try existing media type and it should return it"""
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
n = api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773', 'html')
assert isinstance(n, dict)
# TODO: check json schema?
assert 'message' in n
assert 'actions' in n
assert 'metadata' in n
assert n['message'] == '\n<h3>Simple message</h3>\n<p>\n HTML rendered message: egg egg spam\n</p>\n'
assert set(('dummy', 'dismiss')).issubset(n['actions'])
assert set(('persistent', 'timestamp', 'severity', 'default_action')).issubset(n['metadata'])
def test_get_notification_try_media_type_failure(volatile_dir, config, dummy_notification):
"""Try non-existing media type and it should return default (plain)"""
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
n = api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773', 'dummy')
assert isinstance(n, dict)
# TODO: check json schema?
assert 'message' in n
assert 'actions' in n
assert 'metadata' in n
assert n['message'] == '\n= Simple message =\nMessage: egg egg spam\n'
assert set(('dummy', 'dismiss')).issubset(n['actions'])
assert set(('persistent', 'timestamp', 'severity', 'default_action')).issubset(n['metadata'])
def test_get_notification_force_media_type(volatile_dir, config, dummy_notification):
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
n = api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773', 'html', 'en', force_media_type=True)
assert isinstance(n, dict)
# TODO: check json schema?
assert 'message' in n
assert 'actions' in n
assert 'metadata' in n
assert n['message'] == '\n<h3>Simple message</h3>\n<p>\n HTML rendered message: egg egg spam\n</p>\n'
assert set(('dummy', 'dismiss')).issubset(n['actions'])
assert set(('persistent', 'timestamp', 'severity', 'default_action')).issubset(n['metadata'])
def test_get_notification_force_media_type_failure(volatile_dir, config, dummy_notification):
"""Try to force non-existing media type"""
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
with pytest.raises(MediaTypeNotAvailableException):
api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773', 'dummy', 'en', force_media_type=True)
def test_dismiss_notification(volatile_dir, config, dummy_notification):
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
api.call_action('8c3ad32ca1774eed985604cc1a328773', 'dismiss')
with pytest.raises(NoSuchNotificationException):
api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773')
def test_dismiss_nondismisable_notification(persistent_dir, config, persistent_notification):
f = persistent_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(persistent_notification)
api = Api(confdict=config)
with pytest.raises(NotificationNotDismissibleException):
api.call_action('8c3ad32ca1774eed985604cc1a328773', 'dismiss')
def test_call_action_shortid(volatile_dir, config, dummy_notification):
f = volatile_dir.join('8c3ad32ca1774eed985604cc1a328773.json')
f.write(dummy_notification)
api = Api(confdict=config)
api.call_action('8c3ad32ca1774eed985604cc1a328773', 'dummy')
with pytest.raises(NoSuchNotificationException):
api.get_rendered_notification('8c3ad32ca1774eed985604cc1a328773')