"""Quick integration test for Stories 4.1-4.3 (without Azure OpenAI). Tests the PostgreSQL knowledge indexing workflow: - Story 4.1: Store records in PostgreSQL - Story 4.2: Verify source metadata - Story 4.3: Test exact lookup indexes Usage: PYTHONPATH=. python examples/test_stories_4_1_to_4_3_quick.py """ import asyncio from uuid import uuid4 from sqlalchemy import select from src.db.models import Document, ExtractedRecord as DBExtractedRecord from src.db.session import async_session_maker from src.extraction.models import ExtractedRecord from src.knowledge.indexer import Indexer from src.knowledge.metadata_service import MetadataService from src.knowledge.structured_store import StructuredStore def create_mock_records(document_id: str, filename: str): """Create 5 mock records for testing.""" return [ ExtractedRecord( content={ "Item Code": "2024-4_0019", "Item Name": "Distribution Module A", "Report": "not applicable", "Screen": "Applicable", }, headers=["Item Code", "Item Name", "Report", "Screen"], _source={ "document_id": document_id, "filename": filename, "sheet": "Item Mapping", "table_id": 1, "row": 10, "col_range": "A:D", }, ), ExtractedRecord( content={ "Item Code": "2024-4_0020", "Item Name": "Distribution Module B", "Report": "applicable", "Screen": "○", }, headers=["Item Code", "Item Name", "Report", "Screen"], _source={ "document_id": document_id, "filename": filename, "sheet": "Item Mapping", "table_id": 1, "row": 11, "col_range": "A:D", }, ), ExtractedRecord( content={ "Item Code": "HW-2024-001", "Item Name": "Server Hardware", "Report": "applicable", "Screen": "Applicable", }, headers=["Item Code", "Item Name", "Report", "Screen"], _source={ "document_id": document_id, "filename": filename, "sheet": "Hardware Items", "table_id": 2, "row": 5, "col_range": "A:D", }, ), ExtractedRecord( content={ "Item Code": "SW-2024-001", "Item Name": "Database System", "Report": "not applicable", "Screen": "×", }, headers=["Item Code", "Item Name", "Report", "Screen"], _source={ "document_id": document_id, "filename": filename, "sheet": "Software Items", "table_id": 3, "row": 8, "col_range": "A:D", }, ), ExtractedRecord( content={ "Item Code": "NET-2024-001", "Item Name": "Network Equipment", "Report": "applicable", "Screen": "Applicable", }, headers=["Item Code", "Item Name", "Report", "Screen"], _source={ "document_id": document_id, "filename": filename, "sheet": "Network Items", "table_id": 4, "row": 3, "col_range": "A:D", }, ), ] async def main(): """Run quick integration test.""" print("\n" + "=" * 80) print("QUICK INTEGRATION TEST: Stories 4.1-4.3") print("=" * 80) document_id = uuid4() filename = "test_distribution_spec.xlsx" records = create_mock_records(str(document_id), filename) print(f"\n✅ Created {len(records)} mock records") async with async_session_maker() as session: # Create document document = Document( id=document_id, filename=filename, file_path=f"/uploads/{document_id}.xlsx", file_size_bytes=1024 * 100, status="completed", ) session.add(document) await session.commit() print(f"✅ Created document: {filename}") # Story 4.1: Store records print("\n" + "=" * 80) print("Story 4.1: Structured Store") print("=" * 80) structured_store = StructuredStore(session) count = await structured_store.store_records(records, document_id) print(f"✅ Stored {count} records in PostgreSQL") # Get record IDs result = await session.execute( select(DBExtractedRecord).where( DBExtractedRecord.document_id == document_id ) ) db_records = result.scalars().all() print(f"✅ Verified {len(db_records)} records in database") print(f" Sample: {db_records[0].content}") # Story 4.2: Metadata print("\n" + "=" * 80) print("Story 4.2: Source Metadata Preservation") print("=" * 80) metadata_service = MetadataService(session) # Get source citation for first record first_record_id = db_records[0].id citation = await metadata_service.get_source_citation(first_record_id) if citation: print(f"✅ Retrieved source citation:") print(f" {citation.filename}, Sheet: {citation.sheet_name}, Row: {citation.row_number}") # Get records by location records_by_location = await metadata_service.get_records_by_location( document_id, "Item Mapping" ) print(f"✅ Found {len(records_by_location)} records in 'Item Mapping' sheet") # Validate source completeness validation = await metadata_service.validate_source_completeness(document_id) print(f"✅ Metadata validation:") print(f" Total records: {validation.total_records}") print(f" Complete: {validation.complete_records}") print(f" Valid: {validation.is_valid}") print(f" Completeness: {validation.completeness_percentage:.1f}%") # Story 4.3: Indexes print("\n" + "=" * 80) print("Story 4.3: Exact Lookup Indexing") print("=" * 80) indexer = Indexer(session) index_report = await indexer.verify_indexes() if index_report.all_indexes_exist: print(f"✅ All {len(index_report.expected_indexes)} indexes exist") else: print(f"⚠️ Missing {len(index_report.missing_indexes)} indexes:") for idx in index_report.missing_indexes: print(f" - {idx}") # Test query performance perf_report = await indexer.benchmark_query_performance(document_id) print(f"\n✅ Query Performance:") print(f" Exact lookup: {perf_report.exact_lookup_ms:.2f}ms (target: <50ms)") print(f" Symbol lookup: {perf_report.symbol_lookup_ms:.2f}ms (target: <20ms)") print(f" JSONB query: {perf_report.jsonb_containment_ms:.2f}ms (target: <100ms)") print("\n" + "=" * 80) print("✅ ALL TESTS PASSED") print("=" * 80) print("\nStories tested:") print(" ✅ Story 4.1: Structured Store (PostgreSQL)") print(" ✅ Story 4.2: Source Metadata Preservation") print(" ✅ Story 4.3: Exact Lookup Indexing") print("\n📝 Note: Story 4.4 requires Azure OpenAI credentials") print(" Run: PYTHONPATH=. python examples/test_stories_4_1_to_4_4.py") print("=" * 80 + "\n") if __name__ == "__main__": asyncio.run(main())