443 lines
17 KiB
Python
443 lines
17 KiB
Python
"""
|
|
Unit Tests for JsonManualLoader
|
|
|
|
Tests the database loading functionality including:
|
|
- Clear/Append mode operations
|
|
- Referential integrity maintenance
|
|
- Duplicate handling and conflict resolution
|
|
- Load statistics and reporting
|
|
- Error handling and rollback scenarios
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from typing import List
|
|
|
|
# Import the classes we're testing
|
|
from ..loaders.json_manual_loader import (
|
|
JsonManualLoader, LoadMode, LoadResult, LoadStatistics
|
|
)
|
|
from ..extractors.json_extractor import MakeData, ModelData
|
|
from ..utils.engine_spec_parser import EngineSpec
|
|
|
|
|
|
class TestJsonManualLoader(unittest.TestCase):
|
|
"""Test cases for JsonManualLoader functionality"""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment before each test"""
|
|
self.mock_postgres_loader = Mock()
|
|
self.loader = JsonManualLoader(self.mock_postgres_loader)
|
|
|
|
def create_test_engine_spec(self, displacement: float, config: str, cylinders: int, fuel_type: str = "Gasoline") -> EngineSpec:
|
|
"""Create a test engine specification"""
|
|
return EngineSpec(
|
|
displacement_l=displacement,
|
|
configuration=config,
|
|
cylinders=cylinders,
|
|
fuel_type=fuel_type,
|
|
aspiration="Natural",
|
|
raw_string=f"{displacement}L {config}{cylinders}"
|
|
)
|
|
|
|
def create_test_model_data(self, name: str, years: List[int], engine_count: int = 1, trim_count: int = 2, is_electric: bool = False) -> ModelData:
|
|
"""Create test model data"""
|
|
engines = []
|
|
if is_electric:
|
|
engines = [EngineSpec(None, "Electric", None, "Electric", None, "Electric Motor")]
|
|
else:
|
|
for i in range(engine_count):
|
|
engines.append(self.create_test_engine_spec(2.0 + i, "I", 4))
|
|
|
|
trims = [f"Trim_{i}" for i in range(trim_count)]
|
|
|
|
return ModelData(
|
|
name=name,
|
|
years=years,
|
|
engines=engines,
|
|
trims=trims,
|
|
is_electric=is_electric
|
|
)
|
|
|
|
def create_test_make_data(self, name: str, model_count: int = 2) -> MakeData:
|
|
"""Create test make data"""
|
|
models = []
|
|
for i in range(model_count):
|
|
models.append(self.create_test_model_data(f"Model_{i}", [2024], 1, 2))
|
|
|
|
return MakeData(
|
|
name=name,
|
|
filename=f"{name.lower()}.json",
|
|
models=models,
|
|
processing_errors=[],
|
|
processing_warnings=[]
|
|
)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_load_make_append_mode_new_make(self, mock_conn_context):
|
|
"""Test loading a new make in append mode"""
|
|
# Mock database connection and cursor
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn_context.return_value.__enter__.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# Mock make doesn't exist (returns None)
|
|
mock_cursor.fetchone.return_value = None
|
|
# Mock make insert returns ID
|
|
mock_cursor.fetchone.side_effect = [10] # make ID
|
|
|
|
# Create test data
|
|
make_data = self.create_test_make_data("Toyota", 1)
|
|
stats = LoadStatistics()
|
|
|
|
# Test loading
|
|
make_id = self.loader.load_make(make_data, LoadMode.APPEND, stats)
|
|
|
|
self.assertEqual(make_id, 10)
|
|
self.assertEqual(stats.makes_processed, 1)
|
|
|
|
# Verify make insert was called
|
|
mock_cursor.execute.assert_any_call(
|
|
"INSERT INTO vehicles.make (name) VALUES (%s) RETURNING id",
|
|
("Toyota",)
|
|
)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_load_make_append_mode_existing_make(self, mock_conn_context):
|
|
"""Test loading an existing make in append mode"""
|
|
# Mock database connection and cursor
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn_context.return_value.__enter__.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# Mock make exists (returns ID)
|
|
mock_cursor.fetchone.return_value = (5,) # existing make ID
|
|
|
|
# Create test data
|
|
make_data = self.create_test_make_data("Honda", 1)
|
|
stats = LoadStatistics()
|
|
|
|
# Test loading
|
|
make_id = self.loader.load_make(make_data, LoadMode.APPEND, stats)
|
|
|
|
self.assertEqual(make_id, 5)
|
|
self.assertEqual(stats.duplicate_makes, 1)
|
|
|
|
# Verify make select was called, but not insert
|
|
mock_cursor.execute.assert_any_call(
|
|
"SELECT id FROM vehicles.make WHERE name = %s",
|
|
("Honda",)
|
|
)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_load_engine_with_full_spec(self, mock_conn_context):
|
|
"""Test loading engine with complete specification"""
|
|
# Mock database cursor
|
|
mock_cursor = Mock()
|
|
|
|
# Mock engine doesn't exist
|
|
mock_cursor.fetchone.side_effect = [None, 100] # check, then insert
|
|
|
|
# Create test engine
|
|
engine_spec = self.create_test_engine_spec(2.5, "V", 6, "Gasoline")
|
|
stats = LoadStatistics()
|
|
|
|
# Test loading
|
|
engine_id = self.loader.load_engine(mock_cursor, engine_spec, LoadMode.APPEND, stats)
|
|
|
|
self.assertEqual(engine_id, 100)
|
|
self.assertEqual(stats.engines_inserted, 1)
|
|
|
|
# Verify engine insert with correct parameters
|
|
expected_call = mock_cursor.execute.call_args_list[-1]
|
|
args = expected_call[0]
|
|
|
|
self.assertIn("INSERT INTO vehicles.engine", args[0])
|
|
self.assertEqual(args[1][0], "2.5L V6") # engine name
|
|
self.assertEqual(args[1][1], "2.5lv6") # engine code
|
|
self.assertEqual(args[1][2], 2.5) # displacement
|
|
self.assertEqual(args[1][3], 6) # cylinders
|
|
self.assertEqual(args[1][4], "Gasoline") # fuel type
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_load_engine_electric(self, mock_conn_context):
|
|
"""Test loading electric motor"""
|
|
# Mock database cursor
|
|
mock_cursor = Mock()
|
|
|
|
# Mock engine doesn't exist
|
|
mock_cursor.fetchone.side_effect = [None, 200] # check, then insert
|
|
|
|
# Create electric engine
|
|
engine_spec = EngineSpec(None, "Electric", None, "Electric", None, "Electric Motor")
|
|
stats = LoadStatistics()
|
|
|
|
# Test loading
|
|
engine_id = self.loader.load_engine(mock_cursor, engine_spec, LoadMode.APPEND, stats)
|
|
|
|
self.assertEqual(engine_id, 200)
|
|
self.assertEqual(stats.engines_inserted, 1)
|
|
|
|
# Verify electric motor was inserted correctly
|
|
expected_call = mock_cursor.execute.call_args_list[-1]
|
|
args = expected_call[0]
|
|
|
|
self.assertIn("INSERT INTO vehicles.engine", args[0])
|
|
self.assertEqual(args[1][0], "Electric Motor") # engine name
|
|
self.assertEqual(args[1][4], "Electric") # fuel type
|
|
|
|
def test_load_mode_enum(self):
|
|
"""Test LoadMode enum values"""
|
|
self.assertEqual(LoadMode.CLEAR.value, "clear")
|
|
self.assertEqual(LoadMode.APPEND.value, "append")
|
|
|
|
def test_load_result_properties(self):
|
|
"""Test LoadResult calculated properties"""
|
|
result = LoadResult(
|
|
total_makes=10,
|
|
total_models=50,
|
|
total_model_years=100,
|
|
total_trims=200,
|
|
total_engines=75,
|
|
total_trim_engine_mappings=400,
|
|
failed_makes=["BMW", "Audi"],
|
|
warnings=["Warning 1"],
|
|
load_mode=LoadMode.APPEND
|
|
)
|
|
|
|
self.assertEqual(result.success_count, 8) # 10 - 2 failed
|
|
self.assertEqual(result.success_rate, 0.8) # 8/10
|
|
|
|
def test_load_statistics_initialization(self):
|
|
"""Test LoadStatistics initialization"""
|
|
stats = LoadStatistics()
|
|
|
|
self.assertEqual(stats.makes_processed, 0)
|
|
self.assertEqual(stats.makes_skipped, 0)
|
|
self.assertEqual(len(stats.errors), 0)
|
|
self.assertEqual(len(stats.warnings), 0)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_clear_all_tables(self, mock_conn_context):
|
|
"""Test clearing all tables in correct order"""
|
|
# Mock database connection and cursor
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn_context.return_value.__enter__.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# Test clearing
|
|
self.loader.clear_all_tables()
|
|
|
|
# Verify truncate calls were made
|
|
truncate_calls = [call for call in mock_cursor.execute.call_args_list
|
|
if 'TRUNCATE' in str(call)]
|
|
|
|
self.assertGreater(len(truncate_calls), 0)
|
|
|
|
# Verify commit was called
|
|
mock_conn.commit.assert_called()
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_get_database_statistics(self, mock_conn_context):
|
|
"""Test getting database statistics"""
|
|
# Mock database connection and cursor
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn_context.return_value.__enter__.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# Mock count queries return values
|
|
mock_cursor.fetchone.side_effect = [(10,), (50,), (100,), (200,), (75,), (400,)]
|
|
|
|
# Test getting statistics
|
|
stats = self.loader.get_database_statistics()
|
|
|
|
expected_stats = {
|
|
'make': 10,
|
|
'model': 50,
|
|
'model_year': 100,
|
|
'trim': 200,
|
|
'engine': 75,
|
|
'trim_engine': 400
|
|
}
|
|
|
|
self.assertEqual(stats, expected_stats)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_validate_referential_integrity_clean(self, mock_conn_context):
|
|
"""Test referential integrity validation with clean data"""
|
|
# Mock database connection and cursor
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn_context.return_value.__enter__.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# Mock all integrity checks return 0 (no issues)
|
|
mock_cursor.fetchone.side_effect = [(0,), (0,), (0,), (0,)]
|
|
|
|
# Test validation
|
|
issues = self.loader.validate_referential_integrity()
|
|
|
|
self.assertEqual(len(issues), 0)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_validate_referential_integrity_issues(self, mock_conn_context):
|
|
"""Test referential integrity validation with issues found"""
|
|
# Mock database connection and cursor
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn_context.return_value.__enter__.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# Mock integrity checks return issues
|
|
mock_cursor.fetchone.side_effect = [(2,), (0,), (1,), (0,)] # 2 orphaned models, 1 orphaned trim
|
|
|
|
# Test validation
|
|
issues = self.loader.validate_referential_integrity()
|
|
|
|
self.assertEqual(len(issues), 2)
|
|
self.assertIn("orphaned models", issues[0])
|
|
self.assertIn("orphaned trims", issues[1])
|
|
|
|
def test_load_all_makes_with_errors(self):
|
|
"""Test loading makes where some have processing errors"""
|
|
# Create test data with errors
|
|
good_make = self.create_test_make_data("Toyota", 1)
|
|
bad_make = MakeData(
|
|
name="ErrorMake",
|
|
filename="error.json",
|
|
models=[],
|
|
processing_errors=["JSON parse error"],
|
|
processing_warnings=[]
|
|
)
|
|
|
|
makes_data = [good_make, bad_make]
|
|
|
|
# Mock load_make to work for good make
|
|
with patch.object(self.loader, 'load_make') as mock_load_make:
|
|
mock_load_make.return_value = 1
|
|
|
|
# Test loading
|
|
result = self.loader.load_all_makes(makes_data, LoadMode.APPEND)
|
|
|
|
self.assertEqual(result.total_makes, 2)
|
|
self.assertEqual(result.success_count, 1) # Only good make succeeded
|
|
self.assertIn("ErrorMake", result.failed_makes)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_load_model_new_model(self, mock_conn_context):
|
|
"""Test loading a new model"""
|
|
# Mock database cursor
|
|
mock_cursor = Mock()
|
|
|
|
# Mock model doesn't exist, then return new ID
|
|
mock_cursor.fetchone.side_effect = [None, 25] # check, then insert
|
|
|
|
# Create test data
|
|
model_data = self.create_test_model_data("Camry", [2024], 1, 2)
|
|
stats = LoadStatistics()
|
|
|
|
# Mock load_model_year
|
|
with patch.object(self.loader, 'load_model_year') as mock_load_model_year:
|
|
mock_load_model_year.return_value = 100
|
|
|
|
# Test loading
|
|
model_id = self.loader.load_model(mock_cursor, 1, model_data, LoadMode.APPEND, stats)
|
|
|
|
self.assertEqual(model_id, 25)
|
|
self.assertEqual(stats.models_inserted, 1)
|
|
|
|
# Verify model_year was loaded for each year
|
|
mock_load_model_year.assert_called_once_with(mock_cursor, 25, 2024, model_data, LoadMode.APPEND, stats)
|
|
|
|
@patch('loaders.json_manual_loader.db_connections.postgres_connection')
|
|
def test_load_trim_with_engines(self, mock_conn_context):
|
|
"""Test loading trim and connecting to engines"""
|
|
# Mock database cursor
|
|
mock_cursor = Mock()
|
|
|
|
# Mock trim doesn't exist, then return new ID
|
|
mock_cursor.fetchone.side_effect = [None, 50, None] # trim check, insert, mapping check
|
|
|
|
# Test data
|
|
stats = LoadStatistics()
|
|
engine_ids = [100, 101]
|
|
|
|
# Test loading
|
|
trim_id = self.loader.load_trim(mock_cursor, 1, "Sport", engine_ids, LoadMode.APPEND, stats)
|
|
|
|
self.assertEqual(trim_id, 50)
|
|
self.assertEqual(stats.trims_inserted, 1)
|
|
self.assertEqual(stats.trim_engine_mappings_inserted, 2) # One for each engine
|
|
|
|
|
|
class TestLoadModeAndResults(unittest.TestCase):
|
|
"""Test LoadMode enum and result classes"""
|
|
|
|
def test_load_mode_values(self):
|
|
"""Test LoadMode enum has correct values"""
|
|
self.assertEqual(LoadMode.CLEAR.value, "clear")
|
|
self.assertEqual(LoadMode.APPEND.value, "append")
|
|
|
|
def test_load_result_with_no_failures(self):
|
|
"""Test LoadResult with perfect success"""
|
|
result = LoadResult(
|
|
total_makes=5,
|
|
total_models=25,
|
|
total_model_years=50,
|
|
total_trims=100,
|
|
total_engines=40,
|
|
total_trim_engine_mappings=200,
|
|
failed_makes=[],
|
|
warnings=[],
|
|
load_mode=LoadMode.APPEND
|
|
)
|
|
|
|
self.assertEqual(result.success_count, 5)
|
|
self.assertEqual(result.success_rate, 1.0)
|
|
|
|
def test_load_result_with_failures(self):
|
|
"""Test LoadResult with some failures"""
|
|
result = LoadResult(
|
|
total_makes=10,
|
|
total_models=40,
|
|
total_model_years=80,
|
|
total_trims=160,
|
|
total_engines=60,
|
|
total_trim_engine_mappings=320,
|
|
failed_makes=["BMW", "Audi", "Mercedes"],
|
|
warnings=["Warning 1", "Warning 2"],
|
|
load_mode=LoadMode.CLEAR
|
|
)
|
|
|
|
self.assertEqual(result.success_count, 7) # 10 - 3 failed
|
|
self.assertEqual(result.success_rate, 0.7)
|
|
|
|
def test_load_statistics_defaults(self):
|
|
"""Test LoadStatistics default values"""
|
|
stats = LoadStatistics()
|
|
|
|
# Check all counters start at 0
|
|
self.assertEqual(stats.makes_processed, 0)
|
|
self.assertEqual(stats.makes_skipped, 0)
|
|
self.assertEqual(stats.models_inserted, 0)
|
|
self.assertEqual(stats.model_years_inserted, 0)
|
|
self.assertEqual(stats.trims_inserted, 0)
|
|
self.assertEqual(stats.engines_inserted, 0)
|
|
self.assertEqual(stats.trim_engine_mappings_inserted, 0)
|
|
self.assertEqual(stats.duplicate_makes, 0)
|
|
self.assertEqual(stats.duplicate_models, 0)
|
|
self.assertEqual(stats.duplicate_engines, 0)
|
|
|
|
# Check lists are initialized
|
|
self.assertIsInstance(stats.errors, list)
|
|
self.assertIsInstance(stats.warnings, list)
|
|
self.assertEqual(len(stats.errors), 0)
|
|
self.assertEqual(len(stats.warnings), 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2) |