510 lines
19 KiB
Python
510 lines
19 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import pytest
|
|
import logging
|
|
import uuid
|
|
import time
|
|
from flask_testing import TestCase
|
|
from unittest import mock
|
|
|
|
# Add the project directory to sys.path
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from app import create_app, initialize_storage_structure
|
|
from services.project_service import (
|
|
create_project,
|
|
update_project,
|
|
get_project,
|
|
delete_project,
|
|
get_all_projects,
|
|
get_project_children,
|
|
get_project_document_count,
|
|
filter_projects,
|
|
find_project_directory,
|
|
archive_project,
|
|
)
|
|
|
|
|
|
class AppTestCase(TestCase):
|
|
def create_app(self):
|
|
# Configurar la aplicación para pruebas
|
|
os.environ["FLASK_ENV"] = "testing"
|
|
app = create_app("testing")
|
|
return app
|
|
|
|
def setUp(self):
|
|
# Inicializar la estructura de almacenamiento para pruebas
|
|
initialize_storage_structure(self.app)
|
|
|
|
# Create a test schema for projects
|
|
schemas_dir = os.path.join(self.app.config["STORAGE_PATH"], "schemas")
|
|
os.makedirs(schemas_dir, exist_ok=True)
|
|
with open(os.path.join(schemas_dir, "schema.json"), "w") as f:
|
|
json.dump(
|
|
{
|
|
"SCHEMA1": {
|
|
"name": "Test Schema",
|
|
"descripcion": "Schema for testing",
|
|
},
|
|
"SCHEMA2": {
|
|
"name": "Another Schema",
|
|
"descripcion": "Another schema for testing",
|
|
},
|
|
},
|
|
f,
|
|
)
|
|
|
|
# Generar un identificador único para esta sesión de prueba
|
|
self.test_session_id = str(uuid.uuid4())[:8]
|
|
|
|
def tearDown(self):
|
|
# Cerrar los manejadores de logging del app logger
|
|
for handler in self.app.logger.handlers[:]:
|
|
self.app.logger.removeHandler(handler)
|
|
handler.close()
|
|
|
|
# Limpiar después de cada prueba
|
|
storage_path = self.app.config["STORAGE_PATH"]
|
|
if os.path.exists(storage_path):
|
|
import shutil
|
|
import time
|
|
|
|
# Retry logic to handle file in use errors
|
|
for _ in range(5):
|
|
try:
|
|
shutil.rmtree(storage_path)
|
|
break
|
|
except PermissionError:
|
|
time.sleep(1)
|
|
|
|
def get_unique_name(self, base_name):
|
|
"""Genera un nombre único para evitar conflictos en pruebas"""
|
|
timestamp = int(time.time() * 1000) % 10000
|
|
return f"{base_name}_{self.test_session_id}_{timestamp}"
|
|
|
|
def clean_existing_projects(self):
|
|
"""Limpia proyectos existentes que puedan interferir con las pruebas"""
|
|
with self.app.app_context():
|
|
projects = get_all_projects(include_inactive=True)
|
|
for project in projects:
|
|
project_id = int(project["codigo"].replace("PROJ", ""))
|
|
delete_project(project_id)
|
|
|
|
def test_logging_setup(self):
|
|
# Verificar que el sistema de logging se haya configurado correctamente
|
|
log_path = os.path.join(self.app.config["STORAGE_PATH"], "logs")
|
|
assert os.path.exists(log_path), "El directorio de logs no existe"
|
|
|
|
def test_project_creation(self):
|
|
# Verificar la creación de un proyecto de prueba
|
|
project_path = os.path.join(
|
|
self.app.config["STORAGE_PATH"], "projects", "test_project"
|
|
)
|
|
os.makedirs(project_path, exist_ok=True)
|
|
assert os.path.exists(
|
|
project_path
|
|
), "El proyecto de prueba no se creó correctamente"
|
|
|
|
def test_project_service_creation(self):
|
|
"""Test creating a project using project service"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
project_data = {
|
|
"descripcion": self.get_unique_name("Test Project Service"),
|
|
"cliente": "Test Client",
|
|
"esquema": "SCHEMA1",
|
|
"destinacion": "Test Destination",
|
|
}
|
|
|
|
success, message, project_id = create_project(project_data, "test_user")
|
|
|
|
# Verify project creation success
|
|
assert success, f"Project creation failed: {message}"
|
|
assert project_id is not None
|
|
assert message == "Proyecto creado correctamente."
|
|
|
|
# Verify project directory was created with correct format
|
|
project_dir = find_project_directory(project_id)
|
|
assert project_dir is not None
|
|
assert os.path.exists(project_dir)
|
|
|
|
# Verify project metadata file was created
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
assert os.path.exists(meta_file)
|
|
|
|
# Verify project metadata content
|
|
with open(meta_file, "r") as f:
|
|
metadata = json.load(f)
|
|
assert metadata["descripcion"] == "Test Project Service"
|
|
assert metadata["cliente"] == "Test Client"
|
|
assert metadata["esquema"] == "SCHEMA1"
|
|
assert metadata["creado_por"] == "test_user"
|
|
assert metadata["estado"] == "activo"
|
|
|
|
def test_project_retrieval(self):
|
|
"""Test retrieving project information"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# First create a project
|
|
project_data = {
|
|
"descripcion": self.get_unique_name("Retrieval Test Project"),
|
|
"cliente": "Retrieval Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
|
|
success, _, project_id = create_project(project_data, "test_user")
|
|
assert success
|
|
|
|
# Now retrieve it
|
|
project = get_project(project_id)
|
|
|
|
# Verify retrieval
|
|
assert project is not None
|
|
assert project["descripcion"] == "Retrieval Test Project"
|
|
assert project["cliente"] == "Retrieval Client"
|
|
assert "directory" in project, "Directory information should be included"
|
|
|
|
def test_project_update(self):
|
|
"""Test updating project information"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# First create a project
|
|
project_data = {
|
|
"descripcion": self.get_unique_name("Original Project"),
|
|
"cliente": "Original Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
|
|
success, _, project_id = create_project(project_data, "test_user")
|
|
assert success
|
|
|
|
# Update the project
|
|
updated_data = {
|
|
"descripcion": "Updated Project",
|
|
"cliente": "Updated Client",
|
|
"destinacion": "Updated Destination",
|
|
}
|
|
|
|
update_success, update_message = update_project(
|
|
project_id, updated_data, "modifier_user"
|
|
)
|
|
|
|
# Verify update success
|
|
assert update_success, f"Update failed: {update_message}"
|
|
assert update_message == "Proyecto actualizado correctamente."
|
|
|
|
# Verify the changes were applied
|
|
updated_project = get_project(project_id)
|
|
assert updated_project["descripcion"] == "Updated Project"
|
|
assert updated_project["cliente"] == "Updated Client"
|
|
assert updated_project["destinacion"] == "Updated Destination"
|
|
assert updated_project["modificado_por"] == "modifier_user"
|
|
# Schema should remain unchanged
|
|
assert updated_project["esquema"] == "SCHEMA1"
|
|
|
|
def test_project_deletion(self):
|
|
"""Test project soft deletion"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# First create a project
|
|
project_data = {
|
|
"descripcion": self.get_unique_name("Project to Delete"),
|
|
"cliente": "Delete Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
|
|
success, _, project_id = create_project(project_data, "test_user")
|
|
assert success
|
|
|
|
# Delete the project (mark as inactive)
|
|
delete_success, delete_message = delete_project(project_id)
|
|
|
|
# Verify deletion success
|
|
assert delete_success, f"Deletion failed: {delete_message}"
|
|
assert delete_message == "Proyecto marcado como inactivo."
|
|
|
|
# Verify the project is marked as inactive
|
|
inactive_project = get_project(project_id)
|
|
assert inactive_project["estado"] == "inactivo"
|
|
|
|
# Verify the project directory still exists (soft delete)
|
|
project_dir = find_project_directory(project_id)
|
|
assert os.path.exists(project_dir)
|
|
|
|
def test_get_all_projects(self):
|
|
"""Test retrieving all projects"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# Create multiple projects
|
|
projects_data = [
|
|
{
|
|
"descripcion": self.get_unique_name("Project 1"),
|
|
"cliente": "Client 1",
|
|
"esquema": "SCHEMA1",
|
|
},
|
|
{
|
|
"descripcion": self.get_unique_name("Project 2"),
|
|
"cliente": "Client 2",
|
|
"esquema": "SCHEMA1",
|
|
},
|
|
{
|
|
"descripcion": self.get_unique_name("Project 3"),
|
|
"cliente": "Client 1",
|
|
"esquema": "SCHEMA2",
|
|
},
|
|
]
|
|
|
|
project_ids = []
|
|
for data in projects_data:
|
|
success, _, project_id = create_project(data, "test_user")
|
|
assert success
|
|
project_ids.append(project_id)
|
|
|
|
# Mark one project as inactive
|
|
delete_project(project_ids[1])
|
|
|
|
# Get all active projects
|
|
all_active_projects = get_all_projects(include_inactive=False)
|
|
|
|
# Should only return 2 active projects
|
|
assert len(all_active_projects) == 2
|
|
|
|
# Get all projects including inactive
|
|
all_projects = get_all_projects(include_inactive=True)
|
|
|
|
# Should return all 3 projects
|
|
assert len(all_projects) == 3
|
|
|
|
def test_project_filtering(self):
|
|
"""Test project filtering functionality"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# Create projects with different characteristics
|
|
projects_data = [
|
|
{
|
|
"descripcion": self.get_unique_name("Web Development"),
|
|
"cliente": "Client A",
|
|
"esquema": "SCHEMA1",
|
|
},
|
|
{
|
|
"descripcion": self.get_unique_name("Mobile App"),
|
|
"cliente": "Client B",
|
|
"esquema": "SCHEMA1",
|
|
},
|
|
{
|
|
"descripcion": self.get_unique_name("Desktop Application"),
|
|
"cliente": "Client A",
|
|
"esquema": "SCHEMA2",
|
|
},
|
|
]
|
|
|
|
for data in projects_data:
|
|
success, _, _ = create_project(data, "test_user")
|
|
assert success
|
|
|
|
# Test filtering by client
|
|
client_filter = {"cliente": "Client A"}
|
|
client_results = filter_projects(client_filter)
|
|
assert len(client_results) == 2
|
|
assert all(p["cliente"] == "Client A" for p in client_results)
|
|
|
|
# Test filtering by description
|
|
desc_filter = {"descripcion": "Web"}
|
|
desc_results = filter_projects(desc_filter)
|
|
assert len(desc_results) == 1
|
|
assert desc_results[0]["descripcion"] == "Web Development"
|
|
|
|
def test_project_hierarchy(self):
|
|
"""Test parent-child project relationships"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# Create parent project
|
|
parent_data = {
|
|
"descripcion": self.get_unique_name("Parent Project"),
|
|
"cliente": "Parent Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
|
|
success, _, parent_id = create_project(parent_data, "test_user")
|
|
assert success
|
|
|
|
# Get the parent project code
|
|
parent_project = get_project(parent_id)
|
|
parent_code = parent_project["codigo"]
|
|
|
|
# Create child projects
|
|
child_data = [
|
|
{
|
|
"descripcion": "Child Project 1",
|
|
"cliente": "Child Client",
|
|
"esquema": "SCHEMA1",
|
|
"proyecto_padre": parent_code,
|
|
},
|
|
{
|
|
"descripcion": "Child Project 2",
|
|
"cliente": "Child Client",
|
|
"esquema": "SCHEMA1",
|
|
"proyecto_padre": parent_code,
|
|
},
|
|
]
|
|
|
|
for data in child_data:
|
|
success, _, _ = create_project(data, "test_user")
|
|
assert success
|
|
|
|
# Get children of parent project
|
|
children = get_project_children(parent_id)
|
|
|
|
# Verify children
|
|
assert len(children) == 2
|
|
child_descriptions = [child["descripcion"] for child in children]
|
|
assert "Child Project 1" in child_descriptions
|
|
assert "Child Project 2" in child_descriptions
|
|
|
|
def test_document_count(self):
|
|
"""Test counting documents in a project"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# Create a test project
|
|
project_data = {
|
|
"descripcion": self.get_unique_name("Document Test Project"),
|
|
"cliente": "Document Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
|
|
success, _, project_id = create_project(project_data, "test_user")
|
|
assert success
|
|
|
|
# Find project directory
|
|
project_dir = find_project_directory(project_id)
|
|
documents_dir = os.path.join(project_dir, "documents")
|
|
|
|
# Initially should have no documents
|
|
initial_count = get_project_document_count(project_id)
|
|
assert initial_count == 0
|
|
|
|
# Create some mock document directories
|
|
os.makedirs(os.path.join(documents_dir, "@001_doc1"), exist_ok=True)
|
|
os.makedirs(os.path.join(documents_dir, "@002_doc2"), exist_ok=True)
|
|
|
|
# Create a non-document directory (should be ignored)
|
|
os.makedirs(os.path.join(documents_dir, "not_a_document"), exist_ok=True)
|
|
|
|
# Count should now be 2
|
|
updated_count = get_project_document_count(project_id)
|
|
assert updated_count == 2
|
|
|
|
def test_project_routes(self):
|
|
"""Test project routes and API endpoints"""
|
|
with self.app.app_context():
|
|
# Test project list route
|
|
response = self.client.get("/projects/")
|
|
self.assertStatus(
|
|
response, 302
|
|
) # Cambiar a esperar redirección en lugar de 200
|
|
|
|
# Test project API list route
|
|
api_response = self.client.get("/projects/api/list")
|
|
self.assert200(api_response)
|
|
data = json.loads(api_response.data)
|
|
assert isinstance(data, list)
|
|
|
|
# Test project creation route (GET - form)
|
|
create_response = self.client.get("/projects/create")
|
|
self.assert200(create_response)
|
|
|
|
# Setup mock for testing POST with CSRF token
|
|
with mock.patch("flask_wtf.csrf.validate_csrf", return_value=True):
|
|
# Test project creation route (POST)
|
|
post_data = {
|
|
"descripcion": "Route Test Project",
|
|
"cliente": "Route Test Client",
|
|
"esquema": "SCHEMA1",
|
|
"destinacion": "Route Test Destination",
|
|
"proyecto_padre": "",
|
|
}
|
|
|
|
# Mock project_service.create_project to return success
|
|
with mock.patch(
|
|
"routes.project_routes.create_project",
|
|
return_value=(True, "Proyecto creado correctamente.", 1),
|
|
):
|
|
post_response = self.client.post("/projects/create", data=post_data)
|
|
self.assertStatus(post_response, 302) # Redirect on success
|
|
|
|
# Test project view route
|
|
# First create a real project to view
|
|
project_data = {
|
|
"descripcion": "View Test Project",
|
|
"cliente": "View Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
success, _, project_id = create_project(project_data, "test_user")
|
|
assert success
|
|
|
|
view_response = self.client.get(f"/projects/{project_id}")
|
|
self.assert200(view_response)
|
|
|
|
def test_project_archival(self):
|
|
"""Test project archival functionality"""
|
|
with self.app.app_context():
|
|
# Limpiar proyectos existentes primero
|
|
self.clean_existing_projects()
|
|
|
|
# First create a project
|
|
project_data = {
|
|
"descripcion": self.get_unique_name("Project to Archive"),
|
|
"cliente": "Archive Client",
|
|
"esquema": "SCHEMA1",
|
|
}
|
|
|
|
success, _, project_id = create_project(project_data, "test_user")
|
|
assert success
|
|
|
|
# Archive the project
|
|
archive_success, archive_message = archive_project(
|
|
project_id, "archiver_user"
|
|
)
|
|
|
|
# Verify archival success
|
|
assert archive_success, f"Archival failed: {archive_message}"
|
|
assert archive_message == "Proyecto archivado correctamente."
|
|
|
|
# Verify the project is marked as archived
|
|
archived_project = get_project(project_id)
|
|
assert archived_project["estado"] == "archivado"
|
|
|
|
# Verify the project directory still exists (soft archive)
|
|
project_dir = find_project_directory(project_id)
|
|
assert os.path.exists(project_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Ejecutar pruebas y exportar resultados a resultados_test.json
|
|
result = pytest.main(
|
|
[
|
|
"-v",
|
|
"--tb=short",
|
|
"--disable-warnings",
|
|
"--json-report",
|
|
"--json-report-file=resultados_test.json",
|
|
]
|
|
)
|