OneSignal MCP Server

by WeirdBrains
Verified
import unittest from unittest.mock import patch, MagicMock import os import sys import json import asyncio # Add the parent directory to sys.path to import the server module sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) # Import the server module import onesignal_server class TestOneSignalServer(unittest.TestCase): """Test cases for the OneSignal MCP server.""" def setUp(self): """Set up test environment before each test.""" # Mock environment variables self.env_patcher = patch.dict('os.environ', { 'ONESIGNAL_APP_ID': 'test-app-id', 'ONESIGNAL_API_KEY': 'test-api-key', 'ONESIGNAL_ORG_API_KEY': 'test-org-api-key' }) self.env_patcher.start() # Reset app configurations for each test onesignal_server.app_configs = {} onesignal_server.current_app_key = None # Initialize with test app onesignal_server.add_app_config('test', 'test-app-id', 'test-api-key', 'Test App') onesignal_server.current_app_key = 'test' def tearDown(self): """Clean up after each test.""" self.env_patcher.stop() def test_app_config(self): """Test AppConfig class.""" app = onesignal_server.AppConfig('app-id', 'api-key', 'App Name') self.assertEqual(app.app_id, 'app-id') self.assertEqual(app.api_key, 'api-key') self.assertEqual(app.name, 'App Name') self.assertEqual(str(app), 'App Name (app-id)') def test_add_app_config(self): """Test adding app configurations.""" onesignal_server.add_app_config('new-app', 'new-app-id', 'new-api-key', 'New App') self.assertIn('new-app', onesignal_server.app_configs) self.assertEqual(onesignal_server.app_configs['new-app'].app_id, 'new-app-id') self.assertEqual(onesignal_server.app_configs['new-app'].api_key, 'new-api-key') self.assertEqual(onesignal_server.app_configs['new-app'].name, 'New App') def test_set_current_app(self): """Test setting the current app.""" # Add a second app onesignal_server.add_app_config('second', 'second-app-id', 'second-api-key') # Test switching to an existing app result = onesignal_server.set_current_app('second') self.assertTrue(result) self.assertEqual(onesignal_server.current_app_key, 'second') # Test switching to a non-existent app result = onesignal_server.set_current_app('non-existent') self.assertFalse(result) self.assertEqual(onesignal_server.current_app_key, 'second') # Should not change def test_get_current_app(self): """Test getting the current app configuration.""" current_app = onesignal_server.get_current_app() self.assertIsNotNone(current_app) self.assertEqual(current_app.app_id, 'test-app-id') self.assertEqual(current_app.api_key, 'test-api-key') # Test with no current app onesignal_server.current_app_key = None current_app = onesignal_server.get_current_app() self.assertIsNone(current_app) @patch('requests.get') def test_make_onesignal_request_get(self, mock_get): """Test making a GET request to the OneSignal API.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = {'success': True} mock_response.text = json.dumps({'success': True}) mock_get.return_value = mock_response # Make the request and run it through the event loop loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( onesignal_server.make_onesignal_request('notifications', 'GET', params={'limit': 10}) ) # Check that the request was made correctly mock_get.assert_called_once() args, kwargs = mock_get.call_args self.assertEqual(kwargs['headers']['Authorization'], 'Key test-api-key') self.assertEqual(kwargs['params']['app_id'], 'test-app-id') self.assertEqual(kwargs['params']['limit'], 10) # Check the result self.assertEqual(result, {'success': True}) finally: loop.close() @patch('requests.post') def test_make_onesignal_request_post(self, mock_post): """Test making a POST request to the OneSignal API.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = {'id': 'notification-id'} mock_response.text = json.dumps({'id': 'notification-id'}) mock_post.return_value = mock_response # Make the request data = {'contents': {'en': 'Test message'}} loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( onesignal_server.make_onesignal_request('notifications', 'POST', data=data) ) # Check that the request was made correctly mock_post.assert_called_once() args, kwargs = mock_post.call_args self.assertEqual(kwargs['headers']['Authorization'], 'Key test-api-key') self.assertEqual(kwargs['json']['app_id'], 'test-app-id') self.assertEqual(kwargs['json']['contents']['en'], 'Test message') # Check the result self.assertEqual(result, {'id': 'notification-id'}) finally: loop.close() @patch('requests.get') @patch('onesignal_server.ONESIGNAL_ORG_API_KEY', 'test-org-api-key') def test_make_onesignal_request_with_org_key(self, mock_get): """Test making a request with the organization API key.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = {'apps': []} mock_response.text = json.dumps({'apps': []}) mock_get.return_value = mock_response # Make the request loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( onesignal_server.make_onesignal_request('apps', 'GET', use_org_key=True) ) # Check that the request was made correctly mock_get.assert_called_once() args, kwargs = mock_get.call_args self.assertEqual(kwargs['headers']['Authorization'], 'Key test-org-api-key') # Check the result self.assertEqual(result, {'apps': []}) finally: loop.close() @patch('requests.get') def test_make_onesignal_request_error_handling(self, mock_get): """Test error handling in make_onesignal_request.""" # Mock a request exception mock_get.side_effect = Exception('Test error') # Make the request loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( onesignal_server.make_onesignal_request('notifications') ) # Check the result self.assertIn('error', result) self.assertEqual(result['error'], 'Unexpected error: Test error') finally: loop.close() if __name__ == '__main__': unittest.main()