From a0b3ac0861c1c9517d05cab8e8b1237d23742b95 Mon Sep 17 00:00:00 2001 From: Mark Cotton Date: Mon, 29 May 2023 16:33:44 -0500 Subject: [PATCH] added some additional methods, and ChatGPT tests --- EagleEyev3/__init__.py | 238 +++++++++++++++++++++++++++++++++++++++-- server.py | 3 +- tests.py | 115 ++++++++++++++++++++ 3 files changed, 345 insertions(+), 11 deletions(-) create mode 100644 tests.py diff --git a/EagleEyev3/__init__.py b/EagleEyev3/__init__.py index 4ac22ef..4efb16e 100644 --- a/EagleEyev3/__init__.py +++ b/EagleEyev3/__init__.py @@ -7,7 +7,14 @@ from .settings import * logging.basicConfig(level=logging.INFO) class EagleEyev3(): + """ + Class representing the EagleEyev3 client. + """ + def __init__(self): + """ + Initializes the EagleEyev3 client object. + """ self.client_id = None self.client_secret = None self.access_token = None @@ -21,13 +28,16 @@ class EagleEyev3(): self.users = [] self.bridges = [] self.cameras = [] + self.switches = [] self.users = [] self.accounts = [] self.base_url = None def _load_vars_from_settings(self): - """Load variables from the settings module.""" + """ + Load variables from the settings module. + """ self.client_id = settings.client_id self.client_secret = settings.client_secret self.server_host = settings.server_host @@ -38,12 +48,13 @@ class EagleEyev3(): self.redirect_uri = f"{settings.server_protocol}://{settings.server_host}:{settings.server_port}" def login_tokens(self, code=None, cascade=True): - """Obtains login tokens using the authorization code. - + """ + Obtains login tokens using the authorization code. + Args: code (str): The authorization code. cascade (bool): Indicates whether to cascade and get the base URL and current user information. - + Returns: dict: Dictionary containing the success status, response HTTP status code, data, and current user information. """ @@ -74,11 +85,12 @@ class EagleEyev3(): } def get_base_url(self, cascade=True): - """Obtains the base URL for the user. - + """ + Obtains the base URL for the user. + Args: cascade (bool): Indicates whether to cascade and get the current user information. - + Returns: dict: Dictionary containing the success status, response HTTP status code, and data. """ @@ -110,8 +122,9 @@ class EagleEyev3(): } def get_current_user(self): - """Obtains the information of the current user. - + """ + Obtains the information of the current user. + Returns: dict: Dictionary containing the success status, response HTTP status code, and data. """ @@ -138,3 +151,210 @@ class EagleEyev3(): "response_http_status": response.status_code, "data": response_json } + + def get_list_of_users(self): + """ + Obtains the list of users. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/users" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_users") + + if response.status_code == 200: + success = True + self.users = [i for i in response_json['results']] + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } + + def get_list_of_cameras(self): + """ + Obtains the list of cameras. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/cameras" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_cameras") + + if response.status_code == 200: + success = True + self.cameras = [i for i in response_json['results']] + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } + + def get_list_of_bridges(self): + """ + Obtains the list of bridges. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/bridges" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_bridges") + + if response.status_code == 200: + success = True + self.bridges = [i for i in response_json['results']] + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } + + def get_list_of_switches(self): + """ + Obtains the list of switches. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/switches" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_switches") + + if response.status_code == 200: + success = True + self.switches = [i for i in response_json['results']] + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } + + def get_list_of_available_devices(self, deviceType__in="camera"): + """ + Obtains the list of available devices. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/availableDevices?deviceType__in={deviceType__in}" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_available_devices") + + if response.status_code == 200: + success = True + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } + + def get_list_of_multi_cameras(self): + """ + Obtains the list of multi-cameras. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/multiCameras" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_multi_cameras") + + if response.status_code == 200: + success = True + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } + + def get_list_of_feeds(self): + """ + Obtains the list of feeds. + + Returns: + dict: Dictionary containing the success status, response HTTP status code, and data. + """ + url = f"https://{self.user_base_url}/api/v3.0/feeds" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Accept": "application/json" + } + + response = requests.get(url, headers=headers) + response_json = response.json() + + logging.info(f"{response.status_code} in get_list_of_feeds") + + if response.status_code == 200: + success = True + else: + success = False + + return { + "success": success, + "response_http_status": response.status_code, + "data": response_json + } diff --git a/server.py b/server.py index e1fb080..c7dd322 100644 --- a/server.py +++ b/server.py @@ -1,8 +1,7 @@ import json, requests from flask import Flask, request - -from EagleEyev3 import * +from EagleEyev3 import EagleEyev3 een = EagleEyev3() diff --git a/tests.py b/tests.py new file mode 100644 index 0000000..0f529bd --- /dev/null +++ b/tests.py @@ -0,0 +1,115 @@ +import unittest +from unittest.mock import MagicMock, patch +from EagleEyev3 import EagleEyev3 + +class TestEagleEyev3(unittest.TestCase): + + def setUp(self): + # Create an instance of EagleEyev3 for testing + self.eagle_eye = EagleEyev3() + + @patch('EagleEyev3.requests') + def test_login_tokens_success(self, mock_requests): + # Mock the requests.post method to return a successful response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'access_token': 'token', 'refresh_token': 'refresh_token'} + mock_requests.post.return_value = mock_response + + result = self.eagle_eye.login_tokens(code='authorization_code', cascade=True) + + self.assertTrue(result['success']) + self.assertEqual(result['response_http_status'], 200) + self.assertEqual(result['data']['access_token'], 'token') + self.assertEqual(result['data']['refresh_token'], 'refresh_token') + + @patch('EagleEyev3.requests') + def test_login_tokens_failure(self, mock_requests): + # Mock the requests.post method to return an unsuccessful response + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.json.return_value = {} + mock_requests.post.return_value = mock_response + + result = self.eagle_eye.login_tokens(code='authorization_code', cascade=True) + + self.assertFalse(result['success']) + self.assertEqual(result['response_http_status'], 400) + self.assertEqual(result['data'], {}) + + @patch('EagleEyev3.requests') + def test_get_base_url_success(self, mock_requests): + # Set access_token for testing + self.eagle_eye.access_token = 'access_token' + + # Mock the requests.get method to return a successful response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'httpsBaseUrl': {'hostname': 'base_url'}} + mock_requests.get.return_value = mock_response + + result = self.eagle_eye.get_base_url(cascade=True) + + self.assertTrue(result['success']) + self.assertEqual(result['response_http_status'], 200) + self.assertEqual(self.eagle_eye.user_base_url, 'base_url') + self.assertIsNotNone(result['data']) + + @patch('EagleEyev3.requests') + def test_get_base_url_failure(self, mock_requests): + # Set access_token for testing + self.eagle_eye.access_token = 'access_token' + + # Mock the requests.get method to return an unsuccessful response + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.json.return_value = {} + mock_requests.get.return_value = mock_response + + result = self.eagle_eye.get_base_url(cascade=True) + + self.assertFalse(result['success']) + self.assertEqual(result['response_http_status'], 400) + self.assertEqual(self.eagle_eye.user_base_url, None) + self.assertEqual(result['data'], {}) + + @patch('EagleEyev3.requests') + def test_get_current_user_success(self, mock_requests): + # Set access_token and user_base_url for testing + self.eagle_eye.access_token = 'access_token' + self.eagle_eye.user_base_url = 'base_url' + + # Mock the requests.get method to return a successful response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'user_id': 'user123'} + mock_requests.get.return_value = mock_response + + result = self.eagle_eye.get_current_user() + + self.assertTrue(result['success']) + self.assertEqual(result['response_http_status'], 200) + self.assertEqual(result['data']['user_id'], 'user123') + + @patch('EagleEyev3.requests') + def test_get_current_user_failure(self, mock_requests): + # Set access_token and user_base_url for testing + self.eagle_eye.access_token = 'access_token' + self.eagle_eye.user_base_url = 'base_url' + + # Mock the requests.get method to return an unsuccessful response + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.json.return_value = {} + mock_requests.get.return_value = mock_response + + result = self.eagle_eye.get_current_user() + + self.assertFalse(result['success']) + self.assertEqual(result['response_http_status'], 400) + self.assertEqual(result['data'], {}) + + # Write similar tests for the remaining methods + +if __name__ == '__main__': + unittest.main()