semantica/tests/semantic_extract/test_performance.py

import time
import unittest
print("Starting tests module...")
from unittest.mock import MagicMock, patch
from semantica.semantic_extract.providers import create_provider, ProviderPool, _provider_pool
from semantica.semantic_extract.ner_extractor import NERExtractor
from semantica.semantic_extract.relation_extractor import RelationExtractor
from semantica.semantic_extract.triplet_extractor import TripletExtractor, Triplet
from semantica.semantic_extract.methods import _result_cache, extract_entities_llm, extract_relations_llm, extract_triplets_llm, match_entity
from semantica.semantic_extract.ner_extractor import Entity

class TestSemanticExtractImprovements(unittest.TestCase):
    def setUp(self):
        _provider_pool.clear()
        # Clear cache before each test
        if _result_cache:
            _result_cache._caches["entities"].clear()
            _result_cache._caches["relations"].clear()
            _result_cache._caches["triplets"].clear()

    def test_entity_matching(self):
        print("\nTesting Entity Matching...")
        entities = [
            Entity(text="Apple Inc.", label="ORG", start_char=0, end_char=10, confidence=1.0),
            Entity(text="Steve Jobs", label="PERSON", start_char=0, end_char=10, confidence=1.0)
        ]
        
        # Exact match
        m1 = match_entity("Apple Inc.", entities)
        self.assertIsNotNone(m1)
        self.assertEqual(m1.text, "Apple Inc.")
        
        # Case insensitive
        m2 = match_entity("apple inc.", entities)
        self.assertIsNotNone(m2)
        self.assertEqual(m2.text, "Apple Inc.")
        
        # Substring/Partial match (should work via calculate_similarity)
        # "Apple" is contained in "Apple Inc."
        # calculate_similarity gives a boost for containment
        m3 = match_entity("Apple", entities)
        if m3:
            self.assertEqual(m3.text, "Apple Inc.")
            print("  Partial match 'Apple' -> 'Apple Inc.' successful.")
        else:
            print("  Partial match 'Apple' -> 'Apple Inc.' failed (score too low).")
            
        # No match
        m4 = match_entity("Microsoft", entities)
        self.assertIsNone(m4)
        print("  No match verified.")

        # Synonym match
        # We need entities that match the synonym keys in methods.py (e.g. "acquired" -> "bought")
        # Let's create an entity "bought"
        rel_entities = [Entity(text="bought", label="RELATION", start_char=0, end_char=6, confidence=1.0)]
        m5 = match_entity("acquired", rel_entities)
        self.assertIsNotNone(m5)
        self.assertEqual(m5.text, "bought")
        print("  Synonym match 'acquired' -> 'bought' verified.")
        
        # Empty input
        m6 = match_entity("", entities)
        self.assertIsNone(m6)
        print("  Empty input handled.")

    def test_caching(self):
        print("\nTesting Caching...")
        
        text = "Apple Inc. was founded in 1976."
        
        # Mock provider
        mock_provider = MagicMock()
        mock_provider.is_available.return_value = True
        # Setup mock response for entities
        mock_entities_response = MagicMock()
        mock_entities_response.entities = [
            MagicMock(text="Apple Inc.", label="ORG", confidence=0.9),
            MagicMock(text="1976", label="DATE", confidence=0.9)
        ]
        mock_provider.generate_typed.return_value = mock_entities_response
        
        with patch('semantica.semantic_extract.methods.create_provider', return_value=mock_provider) as mock_create:
            # First call - should hit provider
            print("  First call (cache miss)...")
            results1 = extract_entities_llm(text, provider="openai", model="gpt-4", api_key="test")
            self.assertEqual(len(results1), 2)
            self.assertEqual(mock_provider.generate_typed.call_count, 1)
            
            # Check cache state
            print(f"  Cache size: {len(_result_cache._caches['entities'])}")
            
            # Second call - should hit cache
            print("  Second call (cache hit)...")
            results2 = extract_entities_llm(text, provider="openai", model="gpt-4", api_key="test")
            self.assertEqual(len(results2), 2)
            # Provider should NOT be called again
            self.assertEqual(mock_provider.generate_typed.call_count, 1)
            
            print("  Cache hit verified for entities.")

    def test_secure_caching(self):
        """Test that sensitive parameters are excluded from cache keys."""
        print("\nTesting Secure Caching...")
        
        text = "Security test."
        
        # Mock provider
        mock_provider = MagicMock()
        mock_provider.is_available.return_value = True
        mock_entities_response = MagicMock()
        mock_entities_response.entities = [MagicMock(text="Test", label="TEST", confidence=1.0)]
        mock_provider.generate_typed.return_value = mock_entities_response
        
        with patch('semantica.semantic_extract.methods.create_provider', return_value=mock_provider):
            # First call with one API key
            extract_entities_llm(text, provider="openai", model="gpt-4", api_key="secret_key_1")
            
            # Second call with DIFFERENT API key
            # If secure caching is working, this should be a CACHE HIT because api_key is ignored
            extract_entities_llm(text, provider="openai", model="gpt-4", api_key="secret_key_2")
            
            # Provider should have been called ONLY ONCE
            self.assertEqual(mock_provider.generate_typed.call_count, 1)
            print("  Secure caching verified: Changing API key did not trigger new extraction.")
            
            # Verify cache content
            self.assertIn("entities", _result_cache._caches)
            self.assertTrue(len(_result_cache._caches["entities"]) > 0)

    def test_provider_pool(self):
        print("\nTesting Provider Pool...")
        # Create provider twice with same args
        # We need to mock the actual provider init to avoid API keys requirement if not present
        with patch('semantica.semantic_extract.providers.OpenAIProvider') as MockProvider:
            MockProvider.side_effect = lambda *args, **kwargs: MagicMock()
            
            p1 = create_provider("openai", api_key="test", model_name="gpt-4")
            p2 = create_provider("openai", api_key="test", model_name="gpt-4")
            
            # Should be same instance
            self.assertIs(p1, p2)
            print("  Provider reuse verified.")
            
            # Different args
            p3 = create_provider("openai", api_key="test", model_name="gpt-3.5")
            self.assertIsNot(p1, p3)
            print("  Different args create new instance verified.")
            
            # Explicitly not using pool
            p4 = create_provider("openai", use_pool=False, api_key="test", model_name="gpt-4")
            self.assertIsNot(p1, p4)
            print("  Opt-out of pool verified.")

    def test_ner_parallel_processing(self):
        print("\nTesting NER Parallel Processing...")
        extractor = NERExtractor(method="pattern") # Use pattern which is fast/local
        
        # Mock extract_entities to simulate work and track thread execution
        original_extract = extractor.extract_entities
        
        def mock_extract(text, **kwargs):
            time.sleep(0.1) # Simulate delay
            return original_extract(text, **kwargs)
            
        extractor.extract_entities = mock_extract
        
        texts = ["Text 1", "Text 2", "Text 3", "Text 4"]
        
        start_time = time.time()
        results = extractor.extract(texts)
        end_time = time.time()
        
        duration = end_time - start_time
        print(f"  Parallel NER (default workers) took {duration:.4f}s")
        
        self.assertEqual(len(results), 4)
        
        # Verify sequential fallback
        start_time_seq = time.time()
        extractor.extract(texts, max_workers=1)
        end_time_seq = time.time()
        duration_seq = end_time_seq - start_time_seq
        print(f"  Sequential NER took {duration_seq:.4f}s")

        # Check if parallel was indeed parallel (faster)
        # With 0.1s sleep * 4 items:
        # Sequential ~ 0.4s
        # Parallel (2 workers) ~ 0.2s + overhead
        self.assertLess(duration, duration_seq * 0.8) 
        print("  Parallel execution speedup verified.")

    def test_relation_parallel_processing(self):
        print("\nTesting Relation Parallel Processing...")
        extractor = RelationExtractor(method="pattern")
        
        # Mock extract_relations
        original_extract = extractor.extract_relations
        def mock_extract(text, entities, **kwargs):
            time.sleep(0.1)
            return original_extract(text, entities, **kwargs)
        extractor.extract_relations = mock_extract
        
        texts = ["Text 1", "Text 2", "Text 3", "Text 4"]
        entities = [[], [], [], []]
        
        start_time = time.time()
        results = extractor.extract(texts, entities)
        end_time = time.time()
        
        duration = end_time - start_time
        print(f"  Parallel RE (default workers) took {duration:.4f}s")
        
        self.assertEqual(len(results), 4)
        
        # Sequential
        start_time_seq = time.time()
        extractor.extract(texts, entities, max_workers=1)
        end_time_seq = time.time()
        duration_seq = end_time_seq - start_time_seq
        print(f"  Sequential RE took {duration_seq:.4f}s")
        
        self.assertLess(duration, duration_seq * 0.8)
        print("  Parallel execution speedup verified.")

    def test_relation_extraction_fuzzy_matching(self):
        print("\nTesting Relation Extraction Fuzzy Matching...")
        extractor = RelationExtractor(method="pattern")
        
        # Entities have formal names
        entities = [
            Entity(text="Apple Inc.", label="ORG", start_char=0, end_char=10, confidence=1.0),
            Entity(text="Steve Jobs", label="PERSON", start_char=21, end_char=31, confidence=1.0)
        ]
        
        # Text uses informal name "Apple"
        text = "Apple was founded by Steve Jobs."
        
        relations = extractor.extract(text, entities)
        
        found = False
        for rel in relations:
            # Check if subject matches "Apple Inc." even though text said "Apple"
            if rel.subject.text == "Apple Inc." and rel.object.text == "Steve Jobs" and rel.predicate == "founded_by":
                found = True
                print("  Successfully matched 'Apple' -> 'Apple Inc.' in relation extraction.")
                break
                
        self.assertTrue(found, "Failed to extract relation with fuzzy entity matching")

    def test_triplet_parallel_processing(self):
        print("\nTesting Triplet Parallel Processing...")
        extractor = TripletExtractor(method="pattern")
        
        # Mock extract_triplets
        original_extract = extractor.extract_triplets
        def mock_extract(text, **kwargs):
            time.sleep(0.1)
            # Return dummy triplets to avoid actual extraction overhead
            return [Triplet(subject="s", predicate="p", object="o")]
        extractor.extract_triplets = mock_extract
        
        texts = ["Text 1", "Text 2", "Text 3", "Text 4"]
        
        start_time = time.time()
        results = extractor.extract(texts)
        end_time = time.time()
        
        duration = end_time - start_time
        print(f"  Parallel TE (default workers) took {duration:.4f}s")
        
        self.assertEqual(len(results), 4)
        
        # Sequential
        start_time_seq = time.time()
        extractor.extract(texts, max_workers=1)
        end_time_seq = time.time()
        duration_seq = end_time_seq - start_time_seq
        print(f"  Sequential TE took {duration_seq:.4f}s")
        
        self.assertLess(duration, duration_seq * 0.8)
        print("  Parallel execution speedup verified.")

if __name__ == "__main__":
    suite = unittest.TestLoader().loadTestsFromTestCase(TestSemanticExtractImprovements)
    unittest.TextTestRunner(verbosity=2).run(suite)