#!/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 cada uno de 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 creada por el método "setUp" (al finalizar cada una de los tests). 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, AsyncMock from flask import json import os from repo_api import app, get_image_params, search_process, check_remote_connection, check_remote_image, check_lock_local, check_aux_files, check_file_exists, check_remote_backup, check_virtual_image_conversion, check_free_space, check_virtual_image_reconversion # -------------------------------------------------------------------------------------------- # VARIABLES # -------------------------------------------------------------------------------------------- repo_path = '/opt/opengnsys/ogrepository/images' trash_path = '/opt/opengnsys/ogrepository/images_trash' vm_path = '/opt/opengnsys/ogrepository/images_virtual' # -------------------------------------------------------------------------------------------- # 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". """ # ------------------------------------------------------------------- FUNCTIONS def setUp(self): """ Configura el cliente de prueba de Flask y habilita el modo de prueba (antes de cada test). Esto permite simular peticiones HTTP a la API sin necesidad de un servidor en ejecución. Además, crea los archivos "test4unittest.img" y "test4unittest.vdi", para realizar las pruebas. """ self.app = app.test_client() self.app.testing = True # Creamos las imágenes de prueba ("img" y "vdi"): with open(f"{repo_path}/test4unittest.img", 'w') as test_image: test_image.write(' ') with open(f"{vm_path}/test4unittestvm.vdi", 'w') as vtest_image: vtest_image.write(' ') def tearDown(self): """ Limpia el entorno de pruebas (después de cada test). En este caso, elimina el archivo "test4unittest.img" (de "repo_path" y de "trash_path"), y el archivo "test4unittestvm.vdi" (de "vm_path"). """ if os.path.exists(f"{repo_path}/test4unittest.img"): os.remove(f"{repo_path}/test4unittest.img") if os.path.exists(f"{trash_path}/test4unittest.img"): os.remove(f"{trash_path}/test4unittest.img") if os.path.exists(f"{vm_path}/test4unittestvm.vdi"): os.remove(f"{vm_path}/test4unittestvm.vdi") 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 def mock_search_process_error(process, string_to_search): """ Esta función simula la respuesta de la función "search_process" de la API. Es necesaria para el método "test_send_p2p_error500", porque de alguna forma se asocia el valor esperado ("False") a los parámetros que se pasan al script al que se llama desde el endpoint a testear. """ return False # ------------------------------------------------------------------- Tests "Obtener Información de Estado de ogRepository" @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_status_error(self, mock_subprocess): """ Método de prueba del endpoint "Obtener Información de Estado de ogRepository" (en caso de error "500"). """ print("Testing endpoint 'Obtener Información de Estado de ogRepository' (error 500)...") mock_subprocess.side_effect = Exception("Error al obtener el estado del repositorio") response = self.app.get('/ogrepository/v1/status') self.assertEqual(response.status_code, 500) self.assertIn('Error al obtener el estado del repositorio', response.data.decode()) # ------------------------------------------------------------------- Tests "Obtener Información de todas las Imágenes" @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.subprocess.run') def test_get_repo_info_error(self, mock_subprocess): """ Método de prueba del endpoint "Obtener Información de todas las Imágenes" (en caso de error "500"). """ print("Testing endpoint 'Obtener Información de todas las Imágenes' (error 500)...") mock_subprocess.side_effect = Exception("Error al obtener informacion de todas las imagenes") response = self.app.get('/ogrepository/v1/images') self.assertEqual(response.status_code, 500) self.assertIn('Error al obtener informacion de todas las imagenes', response.data.decode()) # ------------------------------------------------------------------- Tests "Obtener Información de una Imagen concreta" @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.get_image_params') @patch('repo_api.subprocess.run') def test_get_repo_image_info_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Obtener Información de una Imagen concreta" (en caso de error "500"). """ print("Testing endpoint 'Obtener Información de una Imagen concreta' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al obtener informacion de una imagen concreta") response = self.app.get('/ogrepository/v1/images/test_image_id') self.assertEqual(response.status_code, 500) self.assertIn('Error al obtener informacion de una imagen concreta', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_get_repo_image_info_error400(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Obtener Información de una Imagen concreta" (en caso de error "400"). """ print("Testing endpoint 'Obtener Información de una Imagen concreta' (error 400)...") mock_get_image_params.return_value = None response = self.app.get('/ogrepository/v1/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('Image not found', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Actualizar Información del Repositorio" @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.subprocess.run') def test_update_repo_info_error(self, mock_subprocess): """ Método de prueba del endpoint "Actualizar Información del Repositorio" (en caso de error "500"). """ print("Testing endpoint 'Actualizar Información del Repositorio' (error 500)...") mock_subprocess.side_effect = Exception("Error al actualizar la informacion del repositorio") response = self.app.put('/ogrepository/v1/images') self.assertEqual(response.status_code, 500) self.assertIn('Error al actualizar la informacion del repositorio', response.data.decode()) # ------------------------------------------------------------------- Tests "Chequear Integridad de Imagen" @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", "Image file didn't pass the Integrity Check"]) response = self.app.get('/ogrepository/v1/status/images/test_image_id') self.assertEqual(response.status_code, 200) self.assertTrue( "Image file passed the Integrity Check correctly" in json.loads(response.data)['output'] or "Image file didn't pass the Integrity Check" in json.loads(response.data)['output'] ) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_check_image_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Chequear Integridad de Imagen" (en caso de error "500"). """ print("Testing endpoint 'Chequear Integridad de Imagen' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al chequear la integridad de la imagen") response = self.app.get('/ogrepository/v1/status/images/test_image_id') self.assertEqual(response.status_code, 500) self.assertIn('Error al chequear la integridad de la imagen', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_check_image_error400(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Chequear Integridad de Imagen" (en caso de error "400"). """ print("Testing endpoint 'Chequear Integridad de Imagen' (error 400)...") mock_get_image_params.return_value = None response = self.app.get('/ogrepository/v1/status/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('Image not found (inexistent or deleted)', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Eliminar una Imagen" @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_delete_image_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Eliminar una Imagen" (en caso de error "500"). """ print("Testing endpoint 'Eliminar una Imagen' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al eliminar la imagen") response = self.app.delete('/ogrepository/v1/images/test_image_id?method=trash') self.assertEqual(response.status_code, 500) self.assertIn('Error al eliminar la imagen', response.data.decode()) @patch('repo_api.get_image_params') def test_delete_image_error400_image(self, mock_get_image_params): """ Método de prueba del endpoint "Eliminar una Imagen" (en caso de error "400", por imagen inexistente). """ print("Testing endpoint 'Eliminar una Imagen' (error 400, no image)...") mock_get_image_params.return_value = None response = self.app.delete('/ogrepository/v1/images/test_image_id?method=trash') self.assertEqual(response.status_code, 400) self.assertIn("Image not found (inexistent or deleted)", json.loads(response.data)['error']) @patch('repo_api.get_image_params') def test_delete_image_error400_incorrect_method(self, mock_get_image_params): """ Método de prueba del endpoint "Eliminar una Imagen" (en caso de error "400", por método incorrecto). """ print("Testing endpoint 'Eliminar una Imagen' (error 400, incorrect method)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} response = self.app.delete('/ogrepository/v1/images/test_image_id?method=incorrect') self.assertEqual(response.status_code, 400) self.assertIn("Incorrect method (must be 'permanent' or 'trash')", json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Recuperar una Imagen" @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_recover_image_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Recuperar una Imagen" (en caso de error "500"). """ print("Testing endpoint 'Recuperar una Imagen' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al recuperar la imagen") 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, 500) self.assertIn('Error al recuperar la imagen', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_recover_image_error400(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Recuperar una Imagen" (en caso de error "400"). """ print("Testing endpoint 'Recuperar una Imagen' (error 400)...") mock_get_image_params.return_value = None 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, 400) self.assertIn('Image not found (inexistent or recovered previously)', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Eliminar una Imagen de la Papelera" @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.get_image_params') @patch('repo_api.subprocess.run') def test_delete_trash_image_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Eliminar una Imagen de la Papelera" (en caso de error "500"). """ print("Testing endpoint 'Eliminar una Imagen de la Papelera' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al eliminar la imagen de la papelera") response = self.app.delete('/ogrepository/v1/trash/images/test_image_id') self.assertEqual(response.status_code, 500) self.assertIn('Error al eliminar la imagen de la papelera', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_delete_trash_image_error400(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Eliminar una Imagen de la Papelera" (en caso de error "400"). """ print("Testing endpoint 'Eliminar una Imagen de la Papelera' (error 400)...") mock_get_image_params.return_value = None response = self.app.delete('/ogrepository/v1/trash/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('Image not found at trash', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Importar una Imagen" @patch('repo_api.check_remote_connection') @patch('repo_api.check_lock_local') @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, mock_check_lock_local): """ 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) mock_check_lock_local.return_value = AsyncMock(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_local') @patch('repo_api.check_remote_image') @patch('repo_api.subprocess.Popen') def test_import_image_error500(self, mock_popen, mock_check_remote_image, mock_check_remote_connection, mock_check_lock_local): """ Método de prueba del endpoint "Importar una Imagen" (en caso de error "500"). """ print("Testing endpoint 'Importar una Imagen' (error 500)...") mock_check_remote_connection.return_value = True mock_check_remote_image.return_value = None mock_popen.side_effect = Exception("Error al importar la imagen") mock_check_lock_local.return_value = AsyncMock(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, 500) self.assertIn('Error al importar la imagen', response.data.decode()) @patch('repo_api.check_remote_connection') def test_import_image_error400_connection(self, mock_check_remote_connection): """ Método de prueba del endpoint "Importar una Imagen" (en caso de error "400", por fallo en la conexión). """ print("Testing endpoint 'Importar una Imagen' (error 400, no connection)...") mock_check_remote_connection.return_value = False 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, 400) self.assertIn("Can't connect to remote server", json.loads(response.data)['exception']) @patch('repo_api.check_remote_connection') @patch('repo_api.check_remote_image') def test_import_image_error400_image(self, mock_check_remote_image, mock_check_remote_connection): """ Método de prueba del endpoint "Importar una Imagen" (en caso de error "400", por imagen inexistente o bloqueada). """ print("Testing endpoint 'Importar una Imagen' (error 400, no image or image locked)...") mock_check_remote_connection.return_value = True mock_check_remote_image.side_effect = ["Remote image not found", "Remote image is locked"] 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, 400) self.assertTrue( 'Remote image not found' in json.loads(response.data)['exception'] or 'Remote image is locked' in json.loads(response.data)['exception'] ) # ------------------------------------------------------------------- Tests "Hacer backup de una Imagen" @patch('repo_api.check_remote_connection') @patch('repo_api.check_remote_backup') @patch('repo_api.get_image_params') @patch('repo_api.subprocess.Popen') def test_backup_image(self, mock_popen, mock_get_image_params, mock_check_remote_connection, mock_check_remote_backup): """ Método de prueba del endpoint "Hacer backup de una Imagen". """ print("Testing endpoint 'Hacer backup de una Imagen'...") mock_check_remote_connection.return_value = True mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_popen.return_value = MagicMock(returncode=None) mock_check_remote_backup.return_value = AsyncMock(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", "remote_path": "/tmp"}), content_type='application/json') self.assertEqual(response.status_code, 200) self.assertIn('Making image backup...', json.loads(response.data)['output']) @patch('repo_api.check_remote_connection') @patch('repo_api.check_remote_backup') @patch('repo_api.get_image_params') @patch('repo_api.subprocess.Popen') def test_backup_image_error500(self, mock_popen, mock_get_image_params, mock_check_remote_connection, mock_check_remote_backup): """ Método de prueba del endpoint "Hacer backup de una Imagen" (en caso de error "500"). """ print("Testing endpoint 'Hacer backup de una Imagen' (error 500)...") mock_check_remote_connection.return_value = True mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_popen.side_effect = Exception("Error al exportar la imagen") mock_check_remote_backup.return_value = AsyncMock(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", "remote_path": "/tmp"}), content_type='application/json') self.assertEqual(response.status_code, 500) self.assertIn('Error al exportar la imagen', response.data.decode()) @patch('repo_api.check_remote_connection') @patch('repo_api.get_image_params') def test_backup_image_error400_connection(self, mock_get_image_params, mock_check_remote_connection): """ Método de prueba del endpoint "Hacer backup de una Imagen" (en caso de error "400", por fallo en la conexión). """ print("Testing endpoint 'Hacer backup de una Imagen' (error 400, no connection)...") mock_check_remote_connection.return_value = False mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} 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", "remote_path": "/tmp"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn("Can't connect to remote host", json.loads(response.data)['exception']) @patch('repo_api.get_image_params') def test_backup_image_error400_image_inexistent(self, mock_get_image_params): """ Método de prueba del endpoint "Hacer backup de una Imagen" (en caso de error "400", por imagen inexistente). """ print("Testing endpoint 'Hacer backup de una Imagen' (error 400, no image)...") mock_get_image_params.return_value = 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", "remote_path": "/tmp"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn("Image not found", json.loads(response.data)['error']) @patch('repo_api.check_file_exists') @patch('repo_api.get_image_params') def test_backup_image_error400_image_locked(self, mock_get_image_params, mock_check_file_exists): """ Método de prueba del endpoint "Hacer backup de una Imagen" (en caso de error "400", por imagen bloqueada). """ print("Testing endpoint 'Hacer backup de una Imagen' (error 400, image locked)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_check_file_exists.return_value = True 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", "remote_path": "/tmp"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn("Image is locked", json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Crear archivos auxiliares" @patch('repo_api.subprocess.Popen') @patch('repo_api.check_aux_files') def test_create_torrent_sum(self, mock_popen, mock_check_aux_files): """ Método de prueba del endpoint "Crear archivos auxiliares". """ print("Testing endpoint 'Crear archivos auxiliares'...") mock_popen.return_value = MagicMock(returncode=None) mock_check_aux_files.return_value = AsyncMock(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.Popen') def test_create_torrent_sum_error500(self, mock_popen): """ Método de prueba del endpoint "Crear archivos auxiliares" (en caso de error "500"). """ print("Testing endpoint 'Crear archivos auxiliares' (error 500)...") mock_popen.side_effect = Exception("Error al crear los archivos auxiliares") response = self.app.post('/ogrepository/v1/images/torrentsum', data=json.dumps({"image": "test4unittest.img"}), content_type='application/json') self.assertEqual(response.status_code, 500) self.assertIn('Error al crear los archivos auxiliares', response.data.decode()) @patch('repo_api.check_file_exists') def test_create_torrent_sum_error400_no_image(self, mock_check_file_exists): """ Método de prueba del endpoint "Crear archivos auxiliares" (en caso de error "400", por imagen inexistente). """ print("Testing endpoint 'Crear archivos auxiliares' (error 400, no image)...") mock_check_file_exists.return_value = False response = self.app.post('/ogrepository/v1/images/torrentsum', data=json.dumps({"image": "test4unittest.img"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn('Image not found', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Convertir imagen virtual a imagen OpenGnsys" @patch('repo_api.subprocess.Popen') @patch('repo_api.check_virtual_image_conversion') def test_convert_virtual_image(self, mock_popen, mock_check_virtual_image_conversion): """ Método de prueba del endpoint "Convertir imagen virtual a imagen OpenGnsys". """ print("Testing endpoint 'Convertir imagen virtual a imagen OpenGnsys'...") mock_popen.return_value = MagicMock(returncode=None) mock_check_virtual_image_conversion.return_value = AsyncMock(returncode=None) response = self.app.post('/ogrepository/v1/images/virtual', data=json.dumps({"virtual_image":"test4unittestvm.vdi", "filesystem":"ext4"}), content_type='application/json') self.assertEqual(response.status_code, 200) self.assertIn('Converting virtual image...', json.loads(response.data)['output']) @patch('repo_api.check_virtual_image_conversion') @patch('repo_api.check_file_exists') @patch('repo_api.subprocess.Popen') def test_convert_virtual_image_error500(self, mock_popen, mock_check_virtual_image_conversion, mock_check_file_exists): """ Método de prueba del endpoint "Convertir imagen virtual a imagen OpenGnsys" (en caso de error "500"). """ print("Testing endpoint 'Convertir imagen virtual a imagen OpenGnsys' (error 500)...") mock_check_file_exists.return_value = True mock_popen.side_effect = Exception("Error al convertir la imagen virtual") mock_check_virtual_image_conversion.return_value = AsyncMock(returncode=None) response = self.app.post('/ogrepository/v1/images/virtual', data=json.dumps({"virtual_image":"test4unittestvm.vdi", "filesystem":"ext4"}), content_type='application/json') self.assertEqual(response.status_code, 500) self.assertIn('Error al convertir la imagen virtual', response.data.decode()) @patch('repo_api.check_file_exists') def test_convert_virtual_image_error400_no_virtual_image(self, mock_check_file_exists): """ Método de prueba del endpoint "Convertir imagen virtual a imagen OpenGnsys" (en caso de error "400", por imagen virtual inexistente). """ print("Testing endpoint 'Convertir imagen virtual a imagen OpenGnsys' (error 400, no virtual image)...") mock_check_file_exists.return_value = False response = self.app.post('/ogrepository/v1/images/virtual', data=json.dumps({"virtual_image":"test4unittestvm.vdi", "filesystem":"ext4"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn('Virtual image not found', json.loads(response.data)['exception']) @patch('repo_api.check_file_exists') def test_convert_virtual_image_error400_name_incorrect(self, mock_check_file_exists): """ Método de prueba del endpoint "Convertir imagen virtual a imagen OpenGnsys" (en caso de error "400", por nombre incorrecto). """ print("Testing endpoint 'Convertir imagen virtual a imagen OpenGnsys' (error 400, name incorrect)...") mock_check_file_exists.return_value = True response = self.app.post('/ogrepository/v1/images/virtual', data=json.dumps({"virtual_image":"test4unittestvm.vdi", "filesystem":"ext4"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn('There is an image with the same name as the virtual image', json.loads(response.data)['exception']) @patch('repo_api.check_free_space') def test_convert_virtual_image_error400_not_enough_space(self, mock_check_free_space): """ Método de prueba del endpoint "Convertir imagen virtual a imagen OpenGnsys" (en caso de error "400", por falta de espacio en disco). """ print("Testing endpoint 'Convertir imagen virtual a imagen OpenGnsys' (error 400, not enough space)...") mock_check_free_space.return_value = False response = self.app.post('/ogrepository/v1/images/virtual', data=json.dumps({"virtual_image":"test4unittestvm.vdi", "filesystem":"ext4"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn('There is not enough free disk space', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Convertir imagen OpenGnsys a imagen virtual" @patch('repo_api.subprocess.Popen') @patch('repo_api.get_image_params') @patch('repo_api.check_virtual_image_reconversion') def test_convert_image_to_virtual(self, mock_popen, mock_get_image_params, mock_check_virtual_image_reconversion): """ Método de prueba del endpoint "Convertir imagen OpenGnsys a imagen virtual". """ print("Testing endpoint 'Convertir imagen OpenGnsys a imagen virtual'...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_popen.return_value = MagicMock(returncode=None) mock_check_virtual_image_reconversion.return_value = AsyncMock(returncode=None) response = self.app.put('/ogrepository/v1/images/virtual', data=json.dumps({"ID_img": "test_image_id", "vm_extension":"vdi"}), content_type='application/json') self.assertEqual(response.status_code, 200) self.assertIn('Converting image to virtual...', json.loads(response.data)['output']) @patch('repo_api.check_virtual_image_reconversion') @patch('repo_api.get_image_params') @patch('repo_api.subprocess.Popen') def test_convert_image_to_virtual_error500(self, mock_popen, mock_get_image_params, mock_check_virtual_image_reconversion): """ Método de prueba del endpoint "Convertir imagen OpenGnsys a imagen virtual" (en caso de error "500"). """ print("Testing endpoint 'Convertir imagen OpenGnsys a imagen virtual' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_popen.side_effect = Exception("Error al convertir la imagen a virtual") mock_check_virtual_image_reconversion.return_value = AsyncMock(returncode=None) response = self.app.put('/ogrepository/v1/images/virtual', data=json.dumps({"ID_img": "test_image_id", "vm_extension":"vdi"}), content_type='application/json') self.assertEqual(response.status_code, 500) self.assertIn('Error al convertir la imagen a virtual', response.data.decode()) @patch('repo_api.get_image_params') def test_convert_image_to_virtual_error400_no_image(self, mock_get_image_params): """ Método de prueba del endpoint "Convertir imagen OpenGnsys a imagen virtual" (en caso de error "400", por imagen virtual inexistente). """ print("Testing endpoint 'Convertir imagen OpenGnsys a imagen virtual' (error 400, no image)...") mock_get_image_params.return_value = None response = self.app.put('/ogrepository/v1/images/virtual', data=json.dumps({"ID_img": "test_image_id", "vm_extension":"vdi"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn("Image not found", json.loads(response.data)['error']) @patch('repo_api.check_file_exists') @patch('repo_api.get_image_params') def test_convert_image_to_virtual_error400_name_incorrect(self, mock_get_image_params, mock_check_file_exists): """ Método de prueba del endpoint "Convertir imagen OpenGnsys a imagen virtual" (en caso de error "400", por nombre incorrecto). """ print("Testing endpoint 'Convertir imagen OpenGnsys a imagen virtual' (error 400, name incorrect)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_check_file_exists.return_value = True response = self.app.put('/ogrepository/v1/images/virtual', data=json.dumps({"ID_img": "test_image_id", "vm_extension":"vdi"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn('There is an exported virtual image with the same name as the image', json.loads(response.data)['exception']) @patch('repo_api.check_free_space') @patch('repo_api.get_image_params') def test_convert_image_to_virtual_error400_not_enough_space(self, mock_get_image_params, mock_check_free_space): """ Método de prueba del endpoint "Convertir imagen OpenGnsys a imagen virtual" (en caso de error "400", por falta de espacio en disco). """ print("Testing endpoint 'Convertir imagen OpenGnsys a imagen virtual' (error 400, not enough space)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_check_free_space.return_value = False response = self.app.put('/ogrepository/v1/images/virtual', data=json.dumps({"ID_img": "test_image_id", "vm_extension":"vdi"}), content_type='application/json') self.assertEqual(response.status_code, 400) self.assertIn('There is not enough free disk space', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Enviar paquete Wake On Lan" @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.subprocess.run') def test_send_wakeonlan_error500(self, mock_subprocess): """ Método de prueba del endpoint "Enviar paquete Wake On Lan" (en caso de error "500"). """ print("Testing endpoint 'Enviar paquete Wake On Lan' (error 500)...") mock_subprocess.side_effect = Exception("Error al enviar el paquete Wake On Lan") 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, 500) self.assertIn('Error al enviar el paquete Wake On Lan', response.data.decode()) # ------------------------------------------------------------------- Tests "Enviar una Imagen mediante UDPcast" @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.subprocess.Popen') def test_send_udpcast_error500(self, mock_popen, mock_get_image_params): """ Método de prueba del endpoint "Enviar una Imagen mediante UDPcast" (en caso de error "500"). """ print("Testing endpoint 'Enviar una Imagen mediante UDPcast' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_popen.side_effect = Exception("Error al enviar la imagen mediante UDPcast") 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, 500) self.assertIn('Error al enviar la imagen mediante UDPcast', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.Popen') def test_send_udpcast_error400(self, mock_popen, mock_get_image_params): """ Método de prueba del endpoint "Enviar una Imagen mediante UDPcast" (en caso de error "400"). """ print("Testing endpoint 'Enviar una Imagen mediante UDPcast' (error 400)...") mock_get_image_params.return_value = 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, 400) self.assertIn('Image not found', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Enviar una Imagen mediante UFTP" @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.subprocess.Popen') def test_send_uftp_error500(self, mock_popen, mock_get_image_params): """ Método de prueba del endpoint "Enviar una Imagen mediante UFTP" (en caso de error "500"). """ print("Testing endpoint 'Enviar una Imagen mediante UFTP' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_popen.side_effect = Exception("Error al enviar la imagen mediante UFTP") 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, 500) self.assertIn('Error al enviar la imagen mediante UFTP', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.Popen') def test_send_uftp_error400(self, mock_popen, mock_get_image_params): """ Método de prueba del endpoint "Enviar una Imagen mediante UFTP" (en caso de error "400"). """ print("Testing endpoint 'Enviar una Imagen mediante UFTP' (error 400)...") mock_get_image_params.return_value = 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, 400) self.assertIn('Image not found', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Enviar una Imagen mediante P2P" @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.get_image_params') @patch('repo_api.search_process', side_effect=mock_search_process_error) @patch('repo_api.subprocess.Popen') def test_send_p2p_error500(self, mock_popen, mock_get_image_params, mock_search_process): """ Método de prueba del endpoint "Enviar una Imagen mediante P2P" (en caso de error "500"). """ print("Testing endpoint 'Enviar una Imagen mediante P2P' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} 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, 500) self.assertIn("Tracker or Seeder (or both) not running", json.loads(response.data)['error']) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.Popen') def test_send_p2p_error400(self, mock_popen, mock_get_image_params): """ Método de prueba del endpoint "Enviar una Imagen mediante P2P" (en caso de error "400"). """ print("Testing endpoint 'Enviar una Imagen mediante P2P' (error 400)...") mock_get_image_params.return_value = 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, 400) self.assertIn('Image not found', json.loads(response.data)['error']) # ------------------------------------------------------------------- Tests "Ver Estado de Transmisiones UDPcast" @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_udpcast_info_error500(self, mock_subprocess): """ Método de prueba del endpoint "Ver Estado de Transmisiones UDPcast" (en caso de error "500"). """ print("Testing endpoint 'Ver Estado de Transmisiones UDPcast' (error 500)...") mock_subprocess.side_effect = Exception("Error al comprobar las transmisiones UDPcast") response = self.app.get('/ogrepository/v1/udpcast') self.assertEqual(response.status_code, 500) self.assertIn('Error al comprobar las transmisiones UDPcast', response.data.decode()) @patch('repo_api.subprocess.run') def test_get_udpcast_info_error400(self, mock_subprocess): """ Método de prueba del endpoint "Ver Estado de Transmisiones UDPcast" (en caso de error "400"). """ print("Testing endpoint 'Ver Estado de Transmisiones UDPcast' (error 400)...") mock_subprocess.side_effect = Exception("exit status 1") response = self.app.get('/ogrepository/v1/udpcast') self.assertEqual(response.status_code, 400) self.assertIn('No UDPCast active transmissions', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Ver Estado de Transmisiones UFTP" @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.subprocess.run') def test_get_uftp_info_error500(self, mock_subprocess): """ Método de prueba del endpoint "Ver Estado de Transmisiones UFTP" (en caso de error "500"). """ print("Testing endpoint 'Ver Estado de Transmisiones UFTP' (error 500)...") mock_subprocess.side_effect = Exception("Error al comprobar las transmisiones UFTP") response = self.app.get('/ogrepository/v1/uftp') self.assertEqual(response.status_code, 500) self.assertIn('Error al comprobar las transmisiones UFTP', response.data.decode()) @patch('repo_api.subprocess.run') def test_get_uftp_info_error400(self, mock_subprocess): """ Método de prueba del endpoint "Ver Estado de Transmisiones UFTP" (en caso de error "400"). """ print("Testing endpoint 'Ver Estado de Transmisiones UFTP' (error 400)...") mock_subprocess.side_effect = Exception("exit status 1") response = self.app.get('/ogrepository/v1/uftp') self.assertEqual(response.status_code, 400) self.assertIn('No UFTP active transmissions', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Cancelar Transmisión UDPcast" @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_udpcast_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Cancelar Transmisión UDPcast" (en caso de error "500"). """ print("Testing endpoint 'Cancelar Transmisión UDPcast' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al cancelar la transmision UDPcast") response = self.app.delete('/ogrepository/v1/udpcast/images/test_image_id') self.assertEqual(response.status_code, 500) self.assertIn('Error al cancelar la transmision UDPcast', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_stop_udpcast_error400_image(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Cancelar Transmisión UDPcast" (en caso de error "400", por imagen inexistente). """ print("Testing endpoint 'Cancelar Transmisión UDPcast' (error 400, no image)...") mock_get_image_params.return_value = None response = self.app.delete('/ogrepository/v1/udpcast/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('Image not found', json.loads(response.data)['error']) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_stop_udpcast_error400_transmission(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Cancelar Transmisión UDPcast" (en caso de error "400", por transmisión inexistente). """ print("Testing endpoint 'Cancelar Transmisión UDPcast' (error 400, no transmission)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("exit status 3") response = self.app.delete('/ogrepository/v1/udpcast/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('No UDPCast active transmissions for specified image', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Cancelar Transmisión UFTP" @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.get_image_params') @patch('repo_api.subprocess.run') def test_stop_uftp_error500(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Cancelar Transmisión UFTP" (en caso de error "500"). """ print("Testing endpoint 'Cancelar Transmisión UFTP' (error 500)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("Error al cancelar la transmision UFTP") response = self.app.delete('/ogrepository/v1/uftp/images/test_image_id') self.assertEqual(response.status_code, 500) self.assertIn('Error al cancelar la transmision UFTP', response.data.decode()) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_stop_uftp_error400_image(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Cancelar Transmisión UFTP" (en caso de error "400", por imagen inexistente). """ print("Testing endpoint 'Cancelar Transmisión UFTP' (error 400, no image)...") mock_get_image_params.return_value = None response = self.app.delete('/ogrepository/v1/uftp/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('Image not found', json.loads(response.data)['error']) @patch('repo_api.get_image_params') @patch('repo_api.subprocess.run') def test_stop_uftp_error400_transmission(self, mock_subprocess, mock_get_image_params): """ Método de prueba del endpoint "Cancelar Transmisión UFTP" (en caso de error "400", por transmisión inexistente). """ print("Testing endpoint 'Cancelar Transmisión UFTP' (error 400, no transmission)...") mock_get_image_params.return_value = {'name': 'test4unittest', 'extension': 'img'} mock_subprocess.side_effect = Exception("exit status 3") response = self.app.delete('/ogrepository/v1/uftp/images/test_image_id') self.assertEqual(response.status_code, 400) self.assertIn('No UFTP active transmissions for specified image', json.loads(response.data)['exception']) # ------------------------------------------------------------------- Tests "Cancelar Transmisiones P2P" @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']) @patch('repo_api.subprocess.run') def test_stop_p2p_error500(self, mock_subprocess): """ Método de prueba del endpoint "Cancelar Transmisiones P2P" (en caso de error "500"). """ print("Testing endpoint 'Cancelar Transmisiones P2P' (error 500)...") mock_subprocess.side_effect = Exception("Error al cancelar las transmisiones P2P") response = self.app.delete('/ogrepository/v1/p2p') self.assertEqual(response.status_code, 500) self.assertIn('Error al cancelar las transmisiones P2P', response.data.decode()) # -------------------------------------------------------------------------------------------- # Al ejecutar el archivo se llama a "unittest.main()", para lanzar las pruebas definidas en la clase "RepoApiTestCase": if __name__ == '__main__': unittest.main() # --------------------------------------------------------------------------------------------