-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
256 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
42 changes: 42 additions & 0 deletions
42
tests/testFiles/ModifiedTelegramNotification/TelegramNotification.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
from pipot.notifications.INotification import INotification | ||
|
||
|
||
class TelegramNotification(INotification): | ||
def __init__(self, config): | ||
super(TelegramNotification, self).__init__(config) | ||
import telegram # Local import necessary due to parsing before | ||
# dependency is installed | ||
self.bot = telegram.Bot(token=config['token']) | ||
|
||
def process(self, message): | ||
for chat_id in self.config['chat_ids']: | ||
self.bot.send_message(chat_id, message) | ||
|
||
@staticmethod | ||
def get_pip_dependencies(): | ||
return ['python-telegram-bot'] | ||
|
||
@staticmethod | ||
def after_install_hook(): | ||
return True | ||
|
||
@staticmethod | ||
def get_apt_dependencies(): | ||
return [] | ||
|
||
def requires_extra_config(self): | ||
return True | ||
|
||
def is_valid_extra_config(self, config): | ||
return 'token' in config and 'chat_ids' in config | ||
|
||
@staticmethod | ||
def get_extra_config_sample(): | ||
return { | ||
'token': '123456:ABC', | ||
'chat_ids': [123456, 1234567] | ||
} | ||
|
||
@staticmethod | ||
def modified_function_for_test(): | ||
pass |
38 changes: 38 additions & 0 deletions
38
tests/testFiles/TelegramNotification/TelegramNotification.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,38 @@ | ||
from pipot.notifications.INotification import INotification | ||
|
||
|
||
class TelegramNotification(INotification): | ||
def __init__(self, config): | ||
super(TelegramNotification, self).__init__(config) | ||
import telegram # Local import necessary due to parsing before | ||
# dependency is installed | ||
self.bot = telegram.Bot(token=config['token']) | ||
|
||
def process(self, message): | ||
for chat_id in self.config['chat_ids']: | ||
self.bot.send_message(chat_id, message) | ||
|
||
@staticmethod | ||
def get_pip_dependencies(): | ||
return ['python-telegram-bot'] | ||
|
||
@staticmethod | ||
def after_install_hook(): | ||
return True | ||
|
||
@staticmethod | ||
def get_apt_dependencies(): | ||
return [] | ||
|
||
def requires_extra_config(self): | ||
return True | ||
|
||
def is_valid_extra_config(self, config): | ||
return 'token' in config and 'chat_ids' in config | ||
|
||
@staticmethod | ||
def get_extra_config_sample(): | ||
return { | ||
'token': '123456:ABC', | ||
'chat_ids': [123456, 1234567] | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,155 @@ | ||
import os | ||
import sys | ||
import mock | ||
import unittest | ||
import json | ||
import codecs | ||
import filecmp | ||
from mock import patch | ||
from functools import wraps | ||
from werkzeug.datastructures import FileStorage | ||
|
||
from flask import request, jsonify | ||
|
||
import tests.authMock | ||
from database import create_session | ||
from mod_config.models import Notification | ||
from tests.testAppBase import TestAppBase | ||
|
||
test_dir = os.path.dirname(os.path.abspath(__file__)) | ||
notification_dir = os.path.join(test_dir, '../pipot/notifications/') | ||
temp_dir = os.path.join(test_dir, 'temp') | ||
if not os.path.exists(temp_dir): | ||
os.makedirs(temp_dir) | ||
|
||
|
||
def _install_notification_service(cls, update=True, description=""): | ||
from mod_config.controllers import g | ||
if not update: | ||
instance = cls(getattr(cls, 'get_extra_config_sample')()) | ||
notification = Notification( | ||
instance.__class__.__name__, description) | ||
g.db.add(notification) | ||
g.db.commit() | ||
return True | ||
|
||
|
||
class TestNotificationManagement(TestAppBase): | ||
|
||
def setUp(self): | ||
super(TestNotificationManagement, self).setUp() | ||
|
||
def tearDown(self): | ||
super(TestNotificationManagement, self).tearDown() | ||
|
||
@patch("mod_config.controllers._install_notification_service", side_effect=_install_notification_service) | ||
def add_notification(self, notification_name, notification_file_name, install_mock): | ||
# upload the notification file | ||
notification_file = codecs.open(os.path.join(test_dir, 'testFiles', notification_name, | ||
notification_file_name), 'rb') | ||
# notification_file = FileStorage(notification_file) | ||
with self.app.test_client() as client: | ||
data = dict( | ||
file=notification_file, | ||
description='test' | ||
) | ||
response = client.post('/notifications', data=data, follow_redirects=False) | ||
self.assertEqual(response.status_code, 200) | ||
install_mock.assert_called_once() | ||
# check backup notification file is removed under temp_path | ||
self.assertFalse(os.path.isfile(os.path.join(notification_dir, 'temp', notification_file_name))) | ||
# check notification file and folder is created under final_path | ||
self.assertTrue(os.path.isfile(os.path.join(notification_dir, notification_name + '.py'))) | ||
# check database | ||
try: | ||
db = create_session(self.app.config['DATABASE_URI'], drop_tables=False) | ||
notification_row = db.query(Notification.id, Notification.name).first() | ||
notification_id = notification_row.id | ||
name = notification_row.name | ||
finally: | ||
db.remove() | ||
self.assertEqual(notification_name, name) | ||
return notification_id | ||
|
||
def remove_notification(self, notification_id, notification_name): | ||
# delete notification file | ||
with self.app.test_client() as client: | ||
data = dict( | ||
id=notification_id | ||
) | ||
response = client.post('/notifications/delete', data=data, follow_redirects=False) | ||
self.assertEqual(response.status_code, 200) | ||
self.assertEqual(response.get_json()['status'], 'success') | ||
# check notification file and folder is removed unser final_path | ||
self.assertFalse(os.path.isfile(os.path.join(notification_dir, notification_name + '.py'))) | ||
# check database | ||
try: | ||
db = create_session(self.app.config['DATABASE_URI'], drop_tables=False) | ||
notification_row = db.query(Notification.id).first() | ||
result = (notification_row is None) | ||
finally: | ||
db.remove() | ||
self.assertTrue(result) | ||
|
||
@patch("mod_config.controllers._install_notification_service", side_effect=_install_notification_service) | ||
def update_notification(self, notification_id, notification_name, notification_file_name, install_mock): | ||
notification_file = codecs.open(os.path.join(test_dir, 'testFiles', notification_file_name), 'rb') | ||
# update notification file | ||
with self.app.test_client() as client: | ||
data = dict( | ||
notificationUpdate_id=notification_id, | ||
notificationUpdate_file=notification_file, | ||
) | ||
response = client.post('/notifications/update', data=data, follow_redirects=False) | ||
self.assertEqual(response.status_code, 200) | ||
install_mock.assert_called_once() | ||
# check backup notification file is removed under temp_path | ||
self.assertFalse(os.path.isfile(os.path.join(notification_dir, 'temp', notification_name + '.py'))) | ||
# check notification file and folder still exist under final path | ||
self.assertTrue(os.path.isfile(os.path.join(notification_dir, notification_name + '.py'))) | ||
return response | ||
|
||
def test_add_and_delete_notification_file(self): | ||
notification_name = 'TelegramNotification' | ||
notification_file_name = notification_name + '.py' | ||
notification_id = self.add_notification(notification_name, notification_file_name) | ||
self.remove_notification(notification_id, notification_name) | ||
|
||
def test_update_with_valid_notification_file(self): | ||
notification_name = 'TelegramNotification' | ||
notification_file_name = notification_name + '.py' | ||
# add a new discription column | ||
modified_notification_file_namae = 'ModifiedTelegramNotification/TelegramNotification.py' | ||
notification_id = self.add_notification(notification_name, notification_file_name) | ||
response = self.update_notification(notification_id, notification_name, modified_notification_file_namae) | ||
self.assertEqual(response.get_json()['status'], 'success') | ||
# check file content | ||
self.assertTrue(filecmp.cmp(os.path.join(notification_dir, notification_file_name), | ||
os.path.join(test_dir, 'testFiles', modified_notification_file_namae))) | ||
# check on database | ||
try: | ||
db = create_session(self.app.config['DATABASE_URI'], drop_tables=False) | ||
notification_row = db.query(Notification.id, Notification.name).first() | ||
notification_id = notification_row.id | ||
name = notification_row.name | ||
finally: | ||
db.remove() | ||
self.assertEqual(notification_name, name) | ||
self.remove_notification(notification_id, notification_name) | ||
|
||
# def test_update_with_invalid_notification_file(self): | ||
# notification_name = 'TelnetService' | ||
# notification_file_name = notification_name + '.py' | ||
# # try to update an invalid notification file | ||
# modified_notification_file_name = 'EmptyTelnetService/TelnetService.py' | ||
# notification_id = self.add_notification(notification_name, notification_file_name) | ||
# response = self.update_notification(notification_id, notification_name, modified_notification_file_name) | ||
# # check the notification file doesn't change | ||
# self.assertTrue(filecmp.cmp(os.path.join(notification_dir, notification_name, notification_name + '.py'), | ||
# os.path.join(test_dir, 'testFiles', notification_file_name))) | ||
# self.assertEqual(response.get_json()['status'], 'error') | ||
# self.remove_notification(notification_id, notification_name) | ||
|
||
|
||
if __name__ == '__main__': | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters