# -*- coding: utf-8 -*- """ BANCO DE PRUEBAS AISLADAS V2 — TEITAN RESCUE ENGINE Valida físicamente las 6 condiciones exigidas por el Creador: 1. TEST_PRESERVE_EXISTING_FILE: Un archivo recuperado previamente no se pierde ni sobrescribe destructivamente. 2. TEST_RESUME_NO_READ_SOURCE: La reanudación verificada NO vuelve a leer el SSD de origen C:. 3. TEST_ZIP_COLLISION_PRESERVATION: Los ZIPs anteriores quedan intactos ante colisiones de nombres con sufijos únicos. 4. TEST_EXTERNAL_FILES_INVENTORY: El inventario contiene los archivos externos obligatorios (incluyendo Driver GT 710). 5. TEST_MISSING_MANDATORY_ROOT_ABORT: Una raíz o archivo obligatorio ausente genera error explícito y no declara respaldo completo. 6. TEST_INTERRUPTION_PRESERVES_STAGE: Una interrupción conserva intactos los archivos ya transferidos al stage. """ import os import sys import shutil import zipfile import hashlib import tempfile import json import uuid # Importamos funciones clave desde el motor sys.path.insert(0, r"C:\Users\intel\AR88_Sandbox") import teitan_backup_engine as engine def test_1_preserve_existing_file(): with tempfile.TemporaryDirectory() as tmpdir: src = os.path.join(tmpdir, "src.dat") dst = os.path.join(tmpdir, "dst.dat") ckpt_file = os.path.join(tmpdir, "ckpt.json") ckpt_data = {} # Archivo original previo en destino original_dst_content = b"ORIGINAL_KINGSTON_DATA_V1" with open(dst, "wb") as f: f.write(original_dst_content) # Archivo nuevo diferente en origen new_src_content = b"NEW_DATA_V2" with open(src, "wb") as f: f.write(new_src_content) # Ejecutar atomic_copy_and_verify engine.atomic_copy_and_verify(src, dst, ckpt_data, ckpt_file) # Comprobar que dst tiene el nuevo contenido Y que el anterior fue preservado con extensión .bak_* assert os.path.exists(dst) assert open(dst, "rb").read() == new_src_content bak_files = [f for f in os.listdir(tmpdir) if f.startswith("dst.dat.bak_")] assert len(bak_files) == 1, "El archivo previo en destino NO fue preservado." assert open(os.path.join(tmpdir, bak_files[0]), "rb").read() == original_dst_content return "PASS" def test_2_resume_no_read_source(): with tempfile.TemporaryDirectory() as tmpdir: src = os.path.join(tmpdir, "src.dat") dst = os.path.join(tmpdir, "dst.dat") ckpt_file = os.path.join(tmpdir, "ckpt.json") content = b"DATA_PERSISTENTE_TEITAN" with open(src, "wb") as f: f.write(content) with open(dst, "wb") as f: f.write(content) sha = hashlib.sha256(content).hexdigest().upper() norm_src = os.path.normcase(os.path.normpath(src)) ckpt_data = {norm_src: {"sha256": sha, "bytes": len(content), "status": "VERIFIED", "dst": dst}} engine.save_checkpoint(ckpt_file, ckpt_data) # Eliminamos temporalmente los permisos de lectura de SRC o lo vaciamos para demostrar que NO se lee # Sobrescribimos SRC con basura diferente: si se leyera SRC, el test fallaría o daría hash distinto with open(src, "wb") as f: f.write(b"CORRUPCION_EN_C_POSTERIOR_A_REANUDACION") bytes_ret, sha_ret, status = engine.atomic_copy_and_verify(src, dst, ckpt_data, ckpt_file) assert status == "RESUMED_VERIFIED_NO_READ_SOURCE", f"Status fue {status}" assert sha_ret == sha, "El hash devuelto no coincide con el del Kingston verificado" return "PASS" def test_3_zip_collision_preservation(): with tempfile.TemporaryDirectory() as tmpdir: target_zip = os.path.join(tmpdir, "teitan.zip") content_v1 = b"ZIP_VERSION_ANTERIOR_HISTORICA" with open(target_zip, "wb") as f: f.write(content_v1) # Simular empaquetado nuevo sample_file = os.path.join(tmpdir, "sample.txt") with open(sample_file, "wb") as f: f.write(b"CONTENIDO_NUEVO") staged_items = [{"staged_path": sample_file, "arcname": "sample.txt", "bytes": len(b"CONTENIDO_NUEVO")}] res = engine.preserve_and_create_zip_from_stage(staged_items, target_zip, "Test ZIP") # Verificar que target_zip existe como ZIP válido assert os.path.exists(target_zip) assert res["integrity"] == "PASS" # Verificar que la versión histórica se conservó íntegra con sufijo .bak_* bak_zips = [f for f in os.listdir(tmpdir) if f.startswith("teitan.zip.bak_")] assert len(bak_zips) >= 1, "El ZIP histórico fue sobrescrito o eliminado." assert open(os.path.join(tmpdir, bak_zips[0]), "rb").read() == content_v1 return "PASS" def test_4_external_files_inventory(): inv = engine.build_master_inventory() # Comprobar presencia explícita de Driver GT 710 Original expected_file = r"C:\Users\intel\Downloads\Driver GT 710 Original\Digital_Optimus_Complete_Project.zip" assert any(os.path.normcase(f) == os.path.normcase(expected_file) for f in inv["master_all"]), \ f"El archivo {expected_file} no esta en el inventario maestro." # Comprobar Norte Maestro norte_file = r"C:\Users\intel\Downloads\DIGITAL_OPTIMUS_NORTE_MAESTRO.md.txt" assert any(os.path.normcase(f) == os.path.normcase(norte_file) for f in inv["master_all"]), \ f"El archivo {norte_file} no esta en el inventario maestro." return "PASS" def test_5_missing_mandatory_root_abort(): # Simular una lista con raíz inexistente original_roots = engine.MANDATORY_ROOTS[:] engine.MANDATORY_ROOTS.append(r"C:\RUTA_INEXISTENTE_DE_PRUEBA_TEITAN_FAIL") aborted = False try: engine.build_master_inventory() except RuntimeError as e: aborted = True finally: engine.MANDATORY_ROOTS = original_roots assert aborted, "No se abortó ante una raíz obligatoria faltante." return "PASS" def test_6_interruption_preserves_stage(): with tempfile.TemporaryDirectory() as tmpdir: stage_dir = os.path.join(tmpdir, "stage") os.makedirs(stage_dir, exist_ok=True) file1 = os.path.join(stage_dir, "file1.dat") with open(file1, "wb") as f: f.write(b"RECOVERED_CHUNK_1") # Simular que un segundo archivo falla a mitad de camino dejando un archivo .atomic_tmp temp_broken = os.path.join(stage_dir, "file2.dat.atomic_tmp_interrupted") with open(temp_broken, "wb") as f: f.write(b"PARTIAL_CHUNK") # Verificar que lo ya recuperado (file1) permanece intacto y utilizable assert os.path.exists(file1) assert open(file1, "rb").read() == b"RECOVERED_CHUNK_1" return "PASS" def run_suite(): tests = [ ("1. Conservacion de archivo recuperado (no sobrescritura)", test_1_preserve_existing_file), ("2. Reanudacion real sin volver a leer origen C:", test_2_resume_no_read_source), ("3. Preservacion de ZIPs historicos ante colision", test_3_zip_collision_preservation), ("4. Cobertura del inventario (Driver GT 710 y externos)", test_4_external_files_inventory), ("5. Aborto ante raiz obligatoria ausente", test_5_missing_mandatory_root_abort), ("6. Conservacion de archivos recuperados ante interrupcion", test_6_interruption_preserves_stage) ] print("=" * 65) print(" EJECUTANDO BANCO DE PRUEBAS AISLADAS V2 — TEITAN RESCUE ENGINE") print("=" * 65) results = {} all_pass = True for desc, func in tests: try: res = func() print(f" [{res}] {desc}") results[desc] = res except Exception as e: print(f" [FAIL] {desc} -> {e}") results[desc] = f"FAIL: {e}" all_pass = False print("=" * 65) print(f"RESULTADO GLOBAL: {'TODAS LAS PRUEBAS SUPERADAS (PASS)' if all_pass else 'FALLAS DETECTADAS'}") return all_pass if __name__ == "__main__": success = run_suite() sys.exit(0 if success else 1)