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", ] )