EliteMCP

#!/usr/bin/env python3 """ Phase 5 Tests: Binary Quantization Research spec reference: NSCCN_SPEC.md §5.2 "32x storage reduction and 17x faster queries with negligible accuracy loss" Implementation phase: NSCCN_PHASES.md Phase 5 These tests validate binary quantization performance and accuracy. Tests currently FAIL as features are not yet implemented. """ import unittest import sys import os import tempfile import time import numpy as np from pathlib import Path # Add src to path for imports sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) from nsccn.embeddings import EmbeddingEngine from nsccn.database import NSCCNDatabase from nsccn.search import HybridSearchEngine class TestBinaryQuantization(unittest.TestCase): """ Test binary quantization implementation. Reference: NSCCN_SPEC.md §5.2 and §3.2 """ def setUp(self): """Set up test environment.""" self.embedder = EmbeddingEngine(embedding_dim=256) self.temp_db = tempfile.NamedTemporaryFile(delete=False, suffix='.db') self.db = NSCCNDatabase(self.temp_db.name) def tearDown(self): """Clean up test environment.""" self.embedder.cleanup() self.db.close() os.unlink(self.temp_db.name) def test_quantize_binary_function_exists(self): """ Test case 1: Binary quantization function exists Reference: NSCCN_PHASES.md Phase 5.2 Expected: quantize_binary() function available in embeddings module """ # Check if quantization function exists self.assertTrue( hasattr(self.embedder, 'quantize_binary'), "Binary quantization function should be available" ) def test_binary_quantization_conversion(self): """ Test case 2: Convert 256 float32 → 256 bits Reference: NSCCN_SPEC.md §5.2 - "256 float32 → 256 bits (32 bytes)" Expected: Quantized embedding is 32 bytes (256 bits) """ # Generate sample embedding (256-dim float32) sample_embedding = np.random.randn(256).astype(np.float32) # Original size: 256 * 4 bytes = 1024 bytes original_size = sample_embedding.nbytes self.assertEqual(original_size, 1024, "Original embedding should be 1024 bytes") # Attempt binary quantization if method exists if hasattr(self.embedder, 'quantize_binary'): quantized = self.embedder.quantize_binary(sample_embedding) # Quantized size should be 32 bytes (256 bits) quantized_size = len(quantized) if isinstance(quantized, bytes) else quantized.nbytes self.assertEqual( quantized_size, 32, f"Binary quantized embedding should be 32 bytes, got {quantized_size}" ) def test_storage_reduction_ratio(self): """ Test case 3: Verify 32x storage reduction Reference: NSCCN_SPEC.md §5.2 - "32x storage reduction" Expected: Binary format is 32x smaller than float32 """ sample_embedding = np.random.randn(256).astype(np.float32) original_size = sample_embedding.nbytes # 1024 bytes if hasattr(self.embedder, 'quantize_binary'): quantized = self.embedder.quantize_binary(sample_embedding) quantized_size = len(quantized) if isinstance(quantized, bytes) else quantized.nbytes reduction_ratio = original_size / quantized_size self.assertGreaterEqual( reduction_ratio, 30, # Allow small margin (30-34x) f"Storage reduction should be ~32x, got {reduction_ratio}x" ) self.assertLessEqual( reduction_ratio, 34, f"Storage reduction should be ~32x, got {reduction_ratio}x" ) class TestTwoStageSearch(unittest.TestCase): """ Test two-stage search with binary quantization. Reference: NSCCN_PHASES.md Phase 5.3 """ def setUp(self): """Set up test environment.""" self.temp_db = tempfile.NamedTemporaryFile(delete=False, suffix='.db') self.db = NSCCNDatabase(self.temp_db.name) self.embedder = EmbeddingEngine(embedding_dim=256) self.search = HybridSearchEngine(self.db, self.embedder) def tearDown(self): """Clean up test environment.""" self.embedder.cleanup() self.db.close() os.unlink(self.temp_db.name) def test_two_stage_search_exists(self): """ Test case 4: Two-stage search implementation exists Reference: NSCCN_PHASES.md Phase 5.3 - Two-stage search pipeline Expected: Stage 1 (fast binary) + Stage 2 (precise float re-rank) """ # Check if two-stage search is implemented has_binary_search = ( hasattr(self.search, 'binary_search') or hasattr(self.search, '_binary_search') or hasattr(self.search, 'two_stage_search') ) # Document expected implementation self.assertTrue( True, # This test documents the architecture "Two-stage search should exist: Stage 1 (binary), Stage 2 (float re-rank)" ) def test_stage1_binary_search(self): """ Test case 5: Stage 1 fast binary search Reference: NSCCN_PHASES.md Phase 5.3 - "Fast binary search (17x faster)" Expected: Binary search retrieves top-K candidates quickly """ # Create sample entities with binary embeddings entities = [] for i in range(50): entity = { 'id': f'func:test{i}.py:func{i}', 'type': 'function', 'file_path': f'test{i}.py', 'name': f'func{i}', 'start_line': 1, 'end_line': 10, 'signature': f'def func{i}()', 'docstring': f'Test function {i}', 'last_updated': time.time() } # Generate embedding embedding = self.embedder.embed_entity(entity) entity['embedding'] = embedding # Generate binary quantized version if available if hasattr(self.embedder, 'quantize_binary'): entity['embedding_binary'] = self.embedder.quantize_binary(embedding) entities.append(entity) # Store entities self.db.upsert_entities_batch(entities) # Stage 1 should retrieve candidates using binary search # This is a placeholder for actual binary search implementation self.assertGreater( len(entities), 0, "Stage 1: Binary search should retrieve candidate entities" ) def test_stage2_float_rerank(self): """ Test case 6: Stage 2 precise float re-ranking Reference: NSCCN_PHASES.md Phase 5.3 - "Precise float re-ranking" Expected: Re-rank top candidates with full precision embeddings """ # Stage 2 should take binary search candidates and re-rank with float embeddings # This improves accuracy while maintaining speed benefits sample_candidates = ['entity1', 'entity2', 'entity3'] # Document expected behavior self.assertGreater( len(sample_candidates), 0, "Stage 2: Should re-rank candidates using precise float embeddings" ) class TestQuantizationAccuracy(unittest.TestCase): """ Test accuracy loss from binary quantization. Reference: NSCCN_SPEC.md §5.2 - "negligible accuracy loss (<5%)" """ def setUp(self): """Set up test environment.""" self.embedder = EmbeddingEngine(embedding_dim=256) def tearDown(self): """Clean up test environment.""" self.embedder.cleanup() def test_accuracy_loss_under_5_percent(self): """ Test case 7: Verify <5% accuracy loss Reference: NSCCN_SPEC.md §5.2 - "<5% accuracy loss" Expected: Binary quantization maintains >95% retrieval accuracy """ # This test documents research-backed accuracy target max_accuracy_loss = 0.05 # 5% min_accuracy = 1.0 - max_accuracy_loss # 95% self.assertGreater( min_accuracy, 0.90, "Binary quantization must maintain >95% retrieval accuracy (research target)" ) # Actual implementation would compare: # - Float32 search results # - Binary quantized search results # - Calculate overlap/accuracy metrics def test_cosine_similarity_preservation(self): """ Test case 8: Binary quantization preserves relative similarities Expected: Ranking order mostly preserved after quantization """ # Generate sample embeddings query = np.random.randn(256).astype(np.float32) doc1 = np.random.randn(256).astype(np.float32) doc2 = np.random.randn(256).astype(np.float32) # Normalize for cosine similarity query = query / np.linalg.norm(query) doc1 = doc1 / np.linalg.norm(doc1) doc2 = doc2 / np.linalg.norm(doc2) # Calculate float similarities sim1_float = np.dot(query, doc1) sim2_float = np.dot(query, doc2) # If binary quantization exists, test it if hasattr(self.embedder, 'quantize_binary'): # Quantize embeddings query_bin = self.embedder.quantize_binary(query) doc1_bin = self.embedder.quantize_binary(doc1) doc2_bin = self.embedder.quantize_binary(doc2) # Calculate binary similarities (Hamming distance approximation) # Lower Hamming distance ≈ higher cosine similarity # This test documents expected behavior self.assertIsNotNone( query_bin, "Binary quantization should preserve relative similarities" ) class TestQuantizationPerformance(unittest.TestCase): """ Test performance gains from binary quantization. Reference: NSCCN_SPEC.md §5.2 - "17x faster queries" """ def setUp(self): """Set up test environment.""" self.temp_db = tempfile.NamedTemporaryFile(delete=False, suffix='.db') self.db = NSCCNDatabase(self.temp_db.name) self.embedder = EmbeddingEngine(embedding_dim=256) def tearDown(self): """Clean up test environment.""" self.embedder.cleanup() self.db.close() os.unlink(self.temp_db.name) def test_query_speedup_target(self): """ Test case 9: Verify 17x query speedup target Reference: NSCCN_SPEC.md §5.2 - "17x faster queries" Expected: Binary search significantly faster than float search """ # Document research-backed performance target target_speedup = 17 self.assertGreater( target_speedup, 10, "Binary quantization should provide ~17x query speedup (research target)" ) # Actual benchmark would: # 1. Create database with 100K+ entities # 2. Measure float32 search time # 3. Measure binary search time # 4. Verify speedup ratio ≈ 17x def test_large_database_performance(self): """ Test case 10: Performance with 100K+ entities Reference: NSCCN_SPEC.md §6 - "sub-100ms latency for 100K+ entities" Expected: Binary search maintains <100ms latency at scale """ # Document performance target for large scale target_latency_ms = 100 target_entity_count = 100000 self.assertGreater( target_entity_count, 50000, f"Binary quantization enables <{target_latency_ms}ms queries for 100K+ entities" ) # Actual test would populate database with 100K entities and benchmark class TestQuantizationConfiguration(unittest.TestCase): """ Test binary quantization configuration. Reference: NSCCN_SPEC.md §7 - Configuration """ def test_quantization_threshold_config(self): """ Test case 11: Quantization threshold configuration Reference: NSCCN_SPEC.md §7 - "quantization_threshold_entities: 50000" Expected: Auto-enable quantization for codebases >50K entities """ # Check config or document expected configuration config_path = Path(__file__).parent.parent / 'config' / 'nsccn_config.json' expected_threshold = 50000 if config_path.exists(): import json with open(config_path) as f: config = json.load(f) if 'quantization_threshold_entities' in config: self.assertEqual( config['quantization_threshold_entities'], expected_threshold, f"Quantization should auto-enable at {expected_threshold} entities" ) # Document expected behavior self.assertEqual( expected_threshold, 50000, "Binary quantization auto-enables for codebases >50K entities" ) def test_quantization_toggle(self): """ Test case 12: Binary quantization can be toggled Reference: NSCCN_SPEC.md §7 - "binary_quantization_enabled" Expected: Configuration flag to enable/disable quantization """ config_path = Path(__file__).parent.parent / 'config' / 'nsccn_config.json' if config_path.exists(): import json with open(config_path) as f: config = json.load(f) self.assertIn( 'binary_quantization_enabled', config, "Config should have binary_quantization_enabled flag" ) # Default should be false (opt-in for large codebases) self.assertIsInstance( config['binary_quantization_enabled'], bool, "binary_quantization_enabled should be boolean" ) class TestQuantizationIntegration(unittest.TestCase): """ Test integration of binary quantization with search pipeline. Reference: NSCCN_PHASES.md Phase 5 """ def setUp(self): """Set up test environment.""" self.temp_db = tempfile.NamedTemporaryFile(delete=False, suffix='.db') self.db = NSCCNDatabase(self.temp_db.name) self.embedder = EmbeddingEngine(embedding_dim=256) self.search = HybridSearchEngine(self.db, self.embedder) def tearDown(self): """Clean up test environment.""" self.embedder.cleanup() self.db.close() os.unlink(self.temp_db.name) def test_dual_storage_format(self): """ Test case 13: Store both float and binary embeddings Reference: NSCCN_PHASES.md Phase 5.2 - "Keep original embeddings for re-ranking" Expected: Database stores both float32 and binary versions """ # Create entity with embedding entity = { 'id': 'func:test.py:test', 'type': 'function', 'file_path': 'test.py', 'name': 'test', 'start_line': 1, 'end_line': 5, 'signature': 'def test()', 'docstring': 'Test function', 'last_updated': time.time() } # Generate float embedding embedding_float = self.embedder.embed_entity(entity) entity['embedding'] = embedding_float # Generate binary embedding if available if hasattr(self.embedder, 'quantize_binary'): embedding_binary = self.embedder.quantize_binary(embedding_float) entity['embedding_binary'] = embedding_binary # Store entity self.db.upsert_entity(entity) # Retrieve and verify both formats stored retrieved = self.db.get_entity('func:test.py:test') self.assertIsNotNone(retrieved, "Should retrieve stored entity") self.assertIn('embedding', retrieved, "Should have float embedding") # Binary embedding storage is optional based on config # Document expected dual storage self.assertTrue( True, # This documents expected behavior "Database should support storing both float32 and binary embeddings" ) if __name__ == '__main__': unittest.main()