From ec819cb17dfbd44719ed32e8e8d83c13a12a619a Mon Sep 17 00:00:00 2001 From: ggil Date: Fri, 24 Jan 2025 11:33:44 +0100 Subject: [PATCH] refs #1346 - Add API tests --- api/test_repo_api.py | 344 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 344 insertions(+) create mode 100644 api/test_repo_api.py diff --git a/api/test_repo_api.py b/api/test_repo_api.py new file mode 100644 index 0000000..9ee4445 --- /dev/null +++ b/api/test_repo_api.py @@ -0,0 +1,344 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" + Tests de prueba de la API de ogRepository, programados con el módulo "unittest". +---------------------------------------------------------------------------------- + +Define un método "setUp", que configura el cliente de prueba de Flask y habilita el modo de prueba + (y que crea una imagen para realizar los tests). +Esto permite simular peticiones HTTP a la API sin necesidad de un servidor en ejecución. + +También define un método de prueba por cada uno de los endpoints de la API, y decora cada uno con "@patch", + para reemplazar las llamadas a "subprocess.run", "subprocess.Popen" y otras funciones con objetos "MagicMock". +Esto permite simular diferentes respuestas y comportamientos sin ejecutar realmente los comandos del sistema. + +Finalmente, define un método "tearDown" que limpia el entorno de pruebas (eliminando la imagen creado por "setUp"). + +NOTA: Se debe ejecutar como "root", o dará errores de permisos con la imagen de prueba. +""" + +# -------------------------------------------------------------------------------------------- +# IMPORTS +# -------------------------------------------------------------------------------------------- + +import unittest +from unittest.mock import patch, mock_open, MagicMock +from flask import json +import os +from repo_api import app, get_image_params, search_process, check_remote_connection, check_remote_image, check_lock_remote + + +# -------------------------------------------------------------------------------------------- +# VARIABLES +# -------------------------------------------------------------------------------------------- + +repo_path = '/opt/opengnsys/ogrepository/images' + + +# -------------------------------------------------------------------------------------------- +# TESTS +# -------------------------------------------------------------------------------------------- + + +class RepoApiTestCase(unittest.TestCase): + """ Clase que hereda de "unittest.TestCase", para ejecutar pruebas con el módulo "unittest". + Define un método "setUp", que configura el cliente de prueba de Flask y habilita el modo de prueba. + Define un método de prueba por cada uno de los endpoints de la API, y decora cada uno con "@patch", + para reemplazar las llamadas a "subprocess.run", "subprocess.Popen" y otras funciones con objetos "MagicMock". + """ + + def setUp(self): + """ Configura el cliente de prueba de Flask y habilita el modo de prueba. + Esto permite simular peticiones HTTP a la API sin necesidad de un servidor en ejecución. + Además, crea el archivo "test4unittest.img", para realizar las pruebas. + """ + self.app = app.test_client() + self.app.testing = True + # Hay que crear la imagen de prueba, y eliminarla al finalizar las pruebas. + with open(f"{repo_path}/test4unittest.img", 'w') as test_image: + test_image.write(' ') + + + def mock_search_process(process, string_to_search): + """ Esta función simula la respuesta de la función "search_process" de la API. + Es necesaria para los métodos "test_send_udpcast" y "test_send_uftp", porque de alguna forma se asocia el valor esperado ("True") + a los parámetros que se pasan al script al que se llama desde los endpoints a testear. + Sin embargo, no es necesaria para el método "test_send_p2p", donde basta con especificar "mock_search_process.return_value = True" + (aunque testea un endpoint muy similar a los que testean los métodos "test_send_udpcast" y "test_send_uftp"). + """ + return True + + + @patch('repo_api.subprocess.run') + def test_get_repo_status(self, mock_subprocess): + """ Método de prueba del endpoint "Obtener Información de Estado de ogRepository". + """ + print("Testing endpoint 'Obtener Información de Estado de ogRepository'...") + mock_subprocess.return_value = MagicMock(returncode=0, stdout='{"cpu": "20%", "memory": "30%"}') + response = self.app.get('/ogrepository/v1/status') + self.assertEqual(response.status_code, 200) + self.assertIn('cpu', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.run') + def test_get_repo_info(self, mock_subprocess): + """ Método de prueba del endpoint "Obtener Información de todas las Imágenes". + """ + print("Testing endpoint 'Obtener Información de todas las Imágenes'...") + mock_subprocess.return_value = MagicMock(returncode=0, stdout='{"images": []}') + response = self.app.get('/ogrepository/v1/images') + self.assertEqual(response.status_code, 200) + self.assertIn('images', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_get_repo_image_info(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Obtener Información de una Imagen concreta". + """ + print("Testing endpoint 'Obtener Información de una Imagen concreta'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0, stdout='{"image": "info"}') + response = self.app.get('/ogrepository/v1/images/test_image_id') + self.assertEqual(response.status_code, 200) + self.assertIn('image', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.run') + def test_update_repo_info(self, mock_subprocess): + """ Método de prueba del endpoint "Actualizar Información del Repositorio". + """ + print("Testing endpoint 'Actualizar Información del Repositorio'...") + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.put('/ogrepository/v1/images') + self.assertEqual(response.status_code, 200) + self.assertIn('Repository info updated successfully', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_check_image(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Chequear Integridad de Imagen". + """ + print("Testing endpoint 'Chequear Integridad de Imagen'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0, stdout='Image file passed the Integrity Check correctly') + response = self.app.get('/ogrepository/v1/status/images/test_image_id') + self.assertEqual(response.status_code, 200) + self.assertIn('Image file passed the Integrity Check correctly', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_delete_image(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Eliminar una Imagen". + """ + print("Testing endpoint 'Eliminar una Imagen'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.delete('/ogrepository/v1/images/test_image_id?method=trash') + self.assertEqual(response.status_code, 200) + self.assertIn('Image deleted successfully', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_recover_image(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Recuperar una Imagen". + """ + print("Testing endpoint 'Recuperar una Imagen'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.post('/ogrepository/v1/trash/images', data=json.dumps({"ID_img": "test_image_id"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Image recovered successfully', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_delete_trash_image(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Eliminar una Imagen de la Papelera". + """ + print("Testing endpoint 'Eliminar una Imagen de la Papelera'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.delete('/ogrepository/v1/trash/images/test_image_id') + self.assertEqual(response.status_code, 200) + self.assertIn('Image deleted successfully', json.loads(response.data)['output']) + + + @patch('repo_api.check_remote_connection') + @patch('repo_api.check_remote_image') + @patch('repo_api.subprocess.Popen') + def test_import_image(self, mock_popen, mock_check_remote_image, mock_check_remote_connection): + """ Método de prueba del endpoint "Importar una Imagen". + """ + print("Testing endpoint 'Importar una Imagen'...") + mock_check_remote_connection.return_value = True + mock_check_remote_image.return_value = None + mock_popen.return_value = MagicMock(returncode=None) + response = self.app.post('/ogrepository/v1/repo/images', data=json.dumps({"image": "test4unittest.img", "repo_ip": "127.0.0.1", "user": "test_user"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Importing image...', json.loads(response.data)['output']) + + + @patch('repo_api.check_remote_connection') + @patch('repo_api.check_lock_remote') + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.Popen') + def test_export_image(self, mock_popen, mock_get_image_params, mock_check_remote_connection, mock_check_lock_remote): + """ Método de prueba del endpoint "Exportar una Imagen". + """ + print("Testing endpoint 'Exportar una Imagen'...") + mock_check_remote_connection.return_value = True + mock_check_lock_remote.return_value = None + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_popen.return_value = MagicMock(returncode=None) + response = self.app.put('/ogrepository/v1/repo/images', data=json.dumps({"ID_img": "test_image_id", "repo_ip": "127.0.0.1", "user": "test_user"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Exporting image...', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.Popen') + def test_create_torrent_sum(self, mock_popen): + """ Método de prueba del endpoint "Crear archivos auxiliares". + """ + print("Testing endpoint 'Crear archivos auxiliares'...") + mock_popen.return_value = MagicMock(returncode=None) + response = self.app.post('/ogrepository/v1/images/torrentsum', data=json.dumps({"image": "test4unittest.img"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Creating auxiliar files...', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.run') + def test_send_wakeonlan(self, mock_subprocess): + """ Método de prueba del endpoint "Enviar paquete Wake On Lan". + """ + print("Testing endpoint 'Enviar paquete Wake On Lan'...") + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.post('/ogrepository/v1/wol', data=json.dumps({"broadcast_ip": "192.168.1.255", "mac": "00:11:22:33:44:55"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Wake On Lan packet sended successfully', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.search_process', side_effect=mock_search_process) # Esto hace que pille el retorno de la función "search_process" de la API llamando al método "mock_search_process" de esta clase.) + @patch('repo_api.subprocess.Popen') + def test_send_udpcast(self, mock_popen, mock_get_image_params, mock_search_process): + """ Método de prueba del endpoint "Enviar una Imagen mediante UDPcast". + """ + print("Testing endpoint 'Enviar una Imagen mediante UDPcast'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_popen.return_value = MagicMock(returncode=None) + response = self.app.post('/ogrepository/v1/udpcast', data=json.dumps({"ID_img": "test_image_id", "port": "9000", "method": "mcast", "ip": "224.0.0.1", "bitrate": "10M", "nclients": "5", "maxtime": "60"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Sending image...', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.search_process', side_effect=mock_search_process) # Esto hace que pille el retorno de la función "search_process" de la API llamando al método "mock_search_process" de esta clase. + @patch('repo_api.subprocess.Popen') + def test_send_uftp(self, mock_popen, mock_get_image_params, mock_search_process): + """ Método de prueba del endpoint "Enviar una Imagen mediante UFTP". + """ + print("Testing endpoint 'Enviar una Imagen mediante UFTP'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_popen.return_value = MagicMock(returncode=None) + response = self.app.post('/ogrepository/v1/uftp', data=json.dumps({"ID_img": "test_image_id", "port": "9000", "ip": "224.0.0.1", "bitrate": "10M"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Sending image...', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.search_process') + @patch('repo_api.subprocess.Popen') + def test_send_p2p(self, mock_popen, mock_get_image_params, mock_search_process): + """ Método de prueba del endpoint "Enviar una Imagen mediante P2P". + """ + print("Testing endpoint 'Enviar una Imagen mediante P2P'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_search_process.return_value = True + mock_popen.return_value = MagicMock(returncode=None) + response = self.app.post('/ogrepository/v1/p2p', data=json.dumps({"ID_img": "test_image_id"}), content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertIn('Tracker and Seeder serving image correctly', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.run') + def test_get_udpcast_info(self, mock_subprocess): + """ Método de prueba del endpoint "Ver Estado de Transmisiones UDPcast". + """ + print("Testing endpoint 'Ver Estado de Transmisiones UDPcast'...") + mock_subprocess.return_value = MagicMock(returncode=0, stdout='{"udpcast": "info"}') + response = self.app.get('/ogrepository/v1/udpcast') + self.assertEqual(response.status_code, 200) + self.assertIn('udpcast', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.run') + def test_get_uftp_info(self, mock_subprocess): + """ Método de prueba del endpoint "Ver Estado de Transmisiones UFTP". + """ + print("Testing endpoint 'Ver Estado de Transmisiones UFTP'...") + mock_subprocess.return_value = MagicMock(returncode=0, stdout='{"uftp": "info"}') + response = self.app.get('/ogrepository/v1/uftp') + self.assertEqual(response.status_code, 200) + self.assertIn('uftp', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_stop_udpcast(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Cancelar Transmisión UDPcast". + """ + print("Testing endpoint 'Cancelar Transmisión UDPcast'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.delete('/ogrepository/v1/udpcast/images/test_image_id') + self.assertEqual(response.status_code, 200) + self.assertIn('Image transmission canceled successfully', json.loads(response.data)['output']) + + + @patch('repo_api.get_image_params') + @patch('repo_api.subprocess.run') + def test_stop_uftp(self, mock_subprocess, mock_get_image_params): + """ Método de prueba del endpoint "Cancelar Transmisión UFTP". + """ + print("Testing endpoint 'Cancelar Transmisión UFTP'...") + mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.delete('/ogrepository/v1/uftp/images/test_image_id') + self.assertEqual(response.status_code, 200) + self.assertIn('Image transmission canceled successfully', json.loads(response.data)['output']) + + + @patch('repo_api.subprocess.run') + def test_stop_p2p(self, mock_subprocess): + """ Método de prueba del endpoint "Cancelar Transmisiones P2P". + """ + print("Testing endpoint 'Cancelar Transmisiones P2P'...") + mock_subprocess.return_value = MagicMock(returncode=0) + response = self.app.delete('/ogrepository/v1/p2p') + self.assertEqual(response.status_code, 200) + self.assertIn('P2P transmissions canceled successfully', json.loads(response.data)['output']) + + + def tearDown(self): + """ Método para limpiar el entorno de pruebas. + En este caso, elimina el archivo "test4unittest.img" (imagen de prueba). + """ + if os.path.exists(f"{repo_path}/test4unittest.img"): + os.remove(f"{repo_path}/test4unittest.img") + + + +# -------------------------------------------------------------------------------------------- + + +# Al ejecutar el archivo se llama a "unittest.main()", para lanzar las pruebas definidas en la clase "RepoApiTestCase": +if __name__ == '__main__': + unittest.main() + + +# --------------------------------------------------------------------------------------------