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