""" Phase 3 Test: Database CRUD Operations Validation Tests CREATE, READ, UPDATE, DELETE operations on the ClaudeTools database with real database connections and verifies foreign key relationships. """ import sys from datetime import datetime, timezone from uuid import uuid4 import random from sqlalchemy import text from sqlalchemy.exc import IntegrityError, SQLAlchemyError # Add api directory to path sys.path.insert(0, 'D:\\ClaudeTools') from api.database import SessionLocal, check_db_connection from api.models import Client, Machine, Session, Tag, SessionTag class CRUDTester: """Test harness for CRUD operations.""" def __init__(self): self.db = None self.test_ids = { 'client': None, 'machine': None, 'session': None, 'tag': None } self.passed = 0 self.failed = 0 self.errors = [] def connect(self): """Test database connection.""" print("=" * 80) print("PHASE 3: DATABASE CRUD OPERATIONS TEST") print("=" * 80) print("\n1. CONNECTION TEST") print("-" * 80) try: if not check_db_connection(): self.fail("Connection", "check_db_connection() returned False") return False self.db = SessionLocal() # Test basic query result = self.db.execute(text("SELECT DATABASE()")).scalar() self.success("Connection", f"Connected to database: {result}") return True except Exception as e: self.fail("Connection", str(e)) return False def test_create(self): """Test INSERT operations.""" print("\n2. CREATE TEST (INSERT)") print("-" * 80) try: # Create a client (type is required field) with unique name test_suffix = random.randint(1000, 9999) client = Client( name=f"Test Client Corp {test_suffix}", type="msp_client", primary_contact="test@client.com", is_active=True ) self.db.add(client) self.db.commit() self.db.refresh(client) self.test_ids['client'] = client.id self.success("Create Client", f"Created client with ID: {client.id}") # Create a machine (no client_id FK, simplified fields) machine = Machine( hostname=f"test-machine-{test_suffix}", machine_fingerprint=f"test-fingerprint-{test_suffix}", friendly_name="Test Machine", machine_type="laptop", platform="win32", username="testuser" ) self.db.add(machine) self.db.commit() self.db.refresh(machine) self.test_ids['machine'] = machine.id self.success("Create Machine", f"Created machine with ID: {machine.id}") # Create a session with required fields session = Session( client_id=client.id, machine_id=machine.id, session_date=datetime.now(timezone.utc).date(), start_time=datetime.now(timezone.utc), status="completed", session_title="Test CRUD Session" ) self.db.add(session) self.db.commit() self.db.refresh(session) self.test_ids['session'] = session.id self.success("Create Session", f"Created session with ID: {session.id}") # Create a tag tag = Tag( name=f"test-tag-{test_suffix}", category="testing" ) self.db.add(tag) self.db.commit() self.db.refresh(tag) self.test_ids['tag'] = tag.id self.success("Create Tag", f"Created tag with ID: {tag.id}") return True except Exception as e: self.fail("Create", str(e)) return False def test_read(self): """Test SELECT operations.""" print("\n3. READ TEST (SELECT)") print("-" * 80) try: # Query client client = self.db.query(Client).filter( Client.id == self.test_ids['client'] ).first() if not client: self.fail("Read Client", "Client not found") return False if not client.name.startswith("Test Client Corp"): self.fail("Read Client", f"Wrong name: {client.name}") return False self.success("Read Client", f"Retrieved client: {client.name}") # Query machine machine = self.db.query(Machine).filter( Machine.id == self.test_ids['machine'] ).first() if not machine: self.fail("Read Machine", "Machine not found") return False if not machine.hostname.startswith("test-machine"): self.fail("Read Machine", f"Wrong hostname: {machine.hostname}") return False self.success("Read Machine", f"Retrieved machine: {machine.hostname}") # Query session session = self.db.query(Session).filter( Session.id == self.test_ids['session'] ).first() if not session: self.fail("Read Session", "Session not found") return False if session.status != "completed": self.fail("Read Session", f"Wrong status: {session.status}") return False self.success("Read Session", f"Retrieved session with status: {session.status}") # Query tag tag = self.db.query(Tag).filter( Tag.id == self.test_ids['tag'] ).first() if not tag: self.fail("Read Tag", "Tag not found") return False self.success("Read Tag", f"Retrieved tag: {tag.name}") return True except Exception as e: self.fail("Read", str(e)) return False def test_relationships(self): """Test foreign key relationships.""" print("\n4. RELATIONSHIP TEST (Foreign Keys)") print("-" * 80) try: # Test valid relationship: Create session_tag session_tag = SessionTag( session_id=self.test_ids['session'], tag_id=self.test_ids['tag'] ) self.db.add(session_tag) self.db.commit() self.db.refresh(session_tag) self.success("Valid FK", "Created session_tag with valid foreign keys") # Test invalid relationship: Try to create session with non-existent machine try: invalid_session = Session( machine_id="non-existent-machine-id", client_id=self.test_ids['client'], session_date=datetime.now(timezone.utc).date(), start_time=datetime.now(timezone.utc), status="running", session_title="Invalid Session" ) self.db.add(invalid_session) self.db.commit() # If we get here, FK constraint didn't work self.db.rollback() self.fail("Invalid FK", "Foreign key constraint not enforced!") return False except IntegrityError: self.db.rollback() self.success("Invalid FK", "Foreign key constraint properly rejected invalid reference") # Test relationship traversal session = self.db.query(Session).filter( Session.id == self.test_ids['session'] ).first() if not session: self.fail("Relationship Traversal", "Session not found") return False # Access related machine through relationship if hasattr(session, 'machine') and session.machine: machine_hostname = session.machine.hostname self.success("Relationship Traversal", f"Accessed machine through session: {machine_hostname}") else: # Fallback: query machine directly machine = self.db.query(Machine).filter( Machine.machine_id == session.machine_id ).first() if machine: self.success("Relationship Traversal", f"Verified machine exists: {machine.hostname}") else: self.fail("Relationship Traversal", "Could not find related machine") return False return True except Exception as e: self.db.rollback() self.fail("Relationships", str(e)) return False def test_update(self): """Test UPDATE operations.""" print("\n5. UPDATE TEST") print("-" * 80) try: # Update client client = self.db.query(Client).filter( Client.id == self.test_ids['client'] ).first() old_name = client.name new_name = "Updated Test Client Corp" client.name = new_name self.db.commit() self.db.refresh(client) if client.name != new_name: self.fail("Update Client", f"Name not updated: {client.name}") return False self.success("Update Client", f"Updated name: {old_name} -> {new_name}") # Update machine machine = self.db.query(Machine).filter( Machine.id == self.test_ids['machine'] ).first() old_name = machine.friendly_name new_name = "Updated Test Machine" machine.friendly_name = new_name self.db.commit() self.db.refresh(machine) if machine.friendly_name != new_name: self.fail("Update Machine", f"Name not updated: {machine.friendly_name}") return False self.success("Update Machine", f"Updated name: {old_name} -> {new_name}") # Update session status session = self.db.query(Session).filter( Session.id == self.test_ids['session'] ).first() old_status = session.status new_status = "in_progress" session.status = new_status self.db.commit() self.db.refresh(session) if session.status != new_status: self.fail("Update Session", f"Status not updated: {session.status}") return False self.success("Update Session", f"Updated status: {old_status} -> {new_status}") return True except Exception as e: self.fail("Update", str(e)) return False def test_delete(self): """Test DELETE operations and cleanup.""" print("\n6. DELETE TEST (Cleanup)") print("-" * 80) try: # Delete in correct order (respecting FK constraints) # Delete session_tag session_tag = self.db.query(SessionTag).filter( SessionTag.session_id == self.test_ids['session'], SessionTag.tag_id == self.test_ids['tag'] ).first() if session_tag: self.db.delete(session_tag) self.db.commit() self.success("Delete SessionTag", "Deleted session_tag") # Delete tag tag = self.db.query(Tag).filter( Tag.id == self.test_ids['tag'] ).first() if tag: tag_name = tag.name self.db.delete(tag) self.db.commit() self.success("Delete Tag", f"Deleted tag: {tag_name}") # Delete session session = self.db.query(Session).filter( Session.id == self.test_ids['session'] ).first() if session: session_id = session.id self.db.delete(session) self.db.commit() self.success("Delete Session", f"Deleted session: {session_id}") # Delete machine machine = self.db.query(Machine).filter( Machine.id == self.test_ids['machine'] ).first() if machine: hostname = machine.hostname self.db.delete(machine) self.db.commit() self.success("Delete Machine", f"Deleted machine: {hostname}") # Delete client client = self.db.query(Client).filter( Client.id == self.test_ids['client'] ).first() if client: name = client.name self.db.delete(client) self.db.commit() self.success("Delete Client", f"Deleted client: {name}") # Verify all deleted remaining_client = self.db.query(Client).filter( Client.id == self.test_ids['client'] ).first() if remaining_client: self.fail("Delete Verification", "Client still exists after deletion") return False self.success("Delete Verification", "All test records successfully deleted") return True except Exception as e: self.fail("Delete", str(e)) return False def success(self, operation, message): """Record a successful test.""" self.passed += 1 print(f"[PASS] {operation} - {message}") def fail(self, operation, error): """Record a failed test.""" self.failed += 1 self.errors.append(f"{operation}: {error}") print(f"[FAIL] {operation} - {error}") def print_summary(self): """Print test summary.""" print("\n" + "=" * 80) print("TEST SUMMARY") print("=" * 80) print(f"Total Passed: {self.passed}") print(f"Total Failed: {self.failed}") print(f"Success Rate: {(self.passed / (self.passed + self.failed) * 100):.1f}%") if self.errors: print("\nERRORS:") for error in self.errors: print(f" - {error}") print("\nCONCLUSION:") if self.failed == 0: print("[SUCCESS] All CRUD operations working correctly!") print(" - Database connectivity verified") print(" - INSERT operations successful") print(" - SELECT operations successful") print(" - UPDATE operations successful") print(" - DELETE operations successful") print(" - Foreign key constraints enforced") print(" - Relationship traversal working") else: print(f"[FAILURE] {self.failed} test(s) failed - review errors above") print("=" * 80) def cleanup(self): """Clean up database connection.""" if self.db: self.db.close() def main(): """Run all CRUD tests.""" tester = CRUDTester() try: # Test 1: Connection if not tester.connect(): print("\n[ERROR] Cannot proceed without database connection") return # Test 2: Create if not tester.test_create(): print("\n[ERROR] Cannot proceed without successful CREATE operations") tester.cleanup() return # Test 3: Read tester.test_read() # Test 4: Relationships tester.test_relationships() # Test 5: Update tester.test_update() # Test 6: Delete tester.test_delete() except KeyboardInterrupt: print("\n\n[WARNING] Test interrupted by user") except Exception as e: print(f"\n\n[ERROR] Unexpected error: {e}") finally: tester.print_summary() tester.cleanup() if __name__ == "__main__": main()