""" Integration tests for spare-parts search orchestrator and supporting services. Tests the web scraping, spec extraction, and search coordination logic. """ import pytest import asyncio from unittest.mock import AsyncMock, patch, MagicMock from backend.services.spare_parts_search import search_spare_parts, search_multiple_candidates from backend.services.web_scraper import SearchRateLimiter, search_google, search_bing from backend.services.spec_extractor import ExtractedSpecs, extract_specs_from_search, extract_specs_from_multiple_results class TestSearchRateLimiter: """Test rate limiter functionality.""" @pytest.mark.asyncio async def test_rate_limiter_initialization(self): """Rate limiter should initialize with default settings.""" limiter = SearchRateLimiter() assert limiter.refill_rate == 0.2 assert limiter.tokens >= 0 assert limiter.capacity == 1.0 @pytest.mark.asyncio async def test_rate_limiter_custom_rate(self): """Rate limiter should accept custom request rates.""" limiter = SearchRateLimiter(requests_per_second=0.5) assert limiter.refill_rate == 0.5 @pytest.mark.asyncio async def test_rate_limiter_acquire(self): """Rate limiter should acquire tokens.""" limiter = SearchRateLimiter() # Should not raise await limiter.acquire() assert limiter.tokens < 1.0 class TestSpecExtractor: """Test specification extraction from search results.""" def test_extracted_specs_dataclass(self): """ExtractedSpecs should be initialized with defaults.""" specs = ExtractedSpecs() assert specs.manufacturer is None assert specs.model is None assert specs.capacity is None assert specs.confidence == 0.0 def test_extract_memory_specs(self): """Should extract DDR4 RAM specifications.""" title = "Kingston Fury 16GB DDR4-3200 DIMM" snippet = "16GB capacity, DDR4 memory, 3200MHz speed" specs = extract_specs_from_search(title, snippet) assert specs.manufacturer == "Kingston" assert "DDR4" in (specs.memory_type or "") assert "16GB" in (specs.capacity or "") assert specs.confidence > 0.5 def test_extract_storage_specs(self): """Should extract SSD specifications.""" title = "Samsung 970 EVO Plus 1TB NVMe SSD" snippet = "1TB NVMe SSD, 3500MB/s read, M.2 form factor" specs = extract_specs_from_search(title, snippet) assert specs.manufacturer == "Samsung" assert specs.storage_type in ["NVMe", "SSD"] assert "1TB" in (specs.capacity or "") assert specs.confidence > 0.5 def test_extract_processor_specs(self): """Should extract CPU specifications.""" title = "Intel Core i7-12700K Processor" snippet = "12th Gen Intel Core i7, 12 cores, 5.0GHz max" specs = extract_specs_from_search(title, snippet) assert specs.processor_brand == "Intel" assert specs.confidence > 0.3 def test_extract_power_supply_specs(self): """Should extract PSU specifications.""" title = "Corsair RM850x 850W Power Supply" snippet = "850W modular power supply, 80+ Gold certified" specs = extract_specs_from_search(title, snippet) assert specs.manufacturer == "Corsair" assert "850W" in (specs.power_rating or "") assert specs.confidence > 0.3 def test_to_item_fields_memory(self): """Should map memory specs to Item fields.""" specs = ExtractedSpecs( manufacturer="Kingston", memory_type="DDR4", capacity="16GB", speed="3200MHz", description="Kingston DDR4 RAM" ) fields = specs.to_item_fields("Memory") assert fields["type"] == "DDR4" assert "16GB" in fields["notes"] assert "Kingston" in fields["notes"] def test_to_item_fields_storage(self): """Should map storage specs to Item fields.""" specs = ExtractedSpecs( manufacturer="Samsung", model="970 EVO Plus", storage_type="NVMe", capacity="1TB", description="Samsung NVMe SSD" ) fields = specs.to_item_fields("Storage") assert fields["type"] == "NVMe" assert "1TB" in fields["notes"] assert "Samsung" in fields["notes"] def test_extract_specs_from_multiple_results(self): """Should extract specs from multiple results using highest confidence.""" results = [ { "title": "Kingston Fury 16GB DDR4-3200", "snippet": "16GB DDR4 memory module", "url": "https://example.com/1" }, { "title": "Kingston Fury DDR4 RAM", "snippet": "Memory module", "url": "https://example.com/2" } ] fields = extract_specs_from_multiple_results(results, "Memory") assert fields["type"] in ["DDR4", "Generic"] assert "description" in fields assert "notes" in fields def test_extract_specs_empty_results(self): """Should handle empty search results gracefully.""" fields = extract_specs_from_multiple_results([], "Memory") assert fields["type"] == "" assert fields["description"] == "" assert fields["notes"] == "" class TestSearchIntegration: """Test end-to-end search integration.""" @pytest.mark.asyncio async def test_search_spare_parts_non_spare_part(self): """Should return None for consumables (non-spare-parts).""" result = await search_spare_parts(category="6ft SATA Cable") assert result is None @pytest.mark.asyncio async def test_search_spare_parts_missing_query(self): """Should return None if no search query provided.""" result = await search_spare_parts(category="DDR4") # No part_number or item_name assert result is None @pytest.mark.asyncio async def test_search_spare_parts_timeout(self): """Should return None on timeout (graceful degradation).""" with patch('backend.services.spare_parts_search.search_google', side_effect=asyncio.TimeoutError): with patch('backend.services.spare_parts_search.search_bing', side_effect=asyncio.TimeoutError): result = await search_spare_parts( category="Kingston DDR4 16GB", part_number="KF466C40RS-16", timeout=1 ) assert result is None @pytest.mark.asyncio async def test_search_multiple_candidates(self): """Should handle batch search of multiple candidates.""" candidates = [ {"category": "6ft Cable", "part_number": None, "item_name": None}, # Non-spare-part {"category": "DDR4 RAM", "part_number": "KF466C40RS", "item_name": None}, ] # Will fail to search but should handle gracefully result = await search_multiple_candidates(candidates, timeout=2) assert isinstance(result, dict) class TestWebScraper: """Test web scraper functions.""" @pytest.mark.asyncio async def test_search_google_type(self): """search_google should return list of dicts or None.""" # Note: This would require mocking aiohttp to avoid actual network calls # In practice, we'd mock the aiohttp.ClientSession pass @pytest.mark.asyncio async def test_search_bing_type(self): """search_bing should return list of dicts or None.""" # Similar to search_google - would require mocking pass