1048 lines
57 KiB
Python
1048 lines
57 KiB
Python
#!/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
|
|
|
|
|
|
# --------------------------------------------------------------------------------------------
|
|
# 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"
|
|
|
|
|
|
@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".
|
|
"""
|
|
print("Testing endpoint 'Convertir imagen virtual'...")
|
|
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"
|
|
(en caso de error "500").
|
|
"""
|
|
print("Testing endpoint 'Convertir imagen virtual' (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"
|
|
(en caso de error "400", por imagen virtual inexistente).
|
|
"""
|
|
print("Testing endpoint 'Convertir imagen virtual' (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"
|
|
(en caso de error "400", por nombre incorrecto).
|
|
"""
|
|
print("Testing endpoint 'Convertir imagen virtual' (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"
|
|
(en caso de error "400", por falta de espacio en disco).
|
|
"""
|
|
print("Testing endpoint 'Convertir imagen virtual' (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 "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()
|
|
|
|
|
|
# --------------------------------------------------------------------------------------------
|