"""Tests for utility functions.""" from datetime import datetime, timezone import pytest from abacus_models.core.errors import ( ERROR_ENTITY_COMPOSITE_KEY_NOT_SUPPORTED, ERROR_NOT_CREATABLE, ERROR_NOT_SOFT_DELETABLE, ERROR_NOT_UPDATABLE, ) from abacus_models.core.utils import ( current_timestamp, get_class, get_pk, get_pk_columns, is_creatable, is_soft_deletable, is_updatable, try_creatable, try_pk_column, try_soft_deletable, try_updatable, ) from tests.conftest import ( TestModelCompositeKey, TestModelCreateOnly, TestModelFull, TestModelMinimal, TestModelSoftDeleteOnly, TestModelUpdateOnly, ) class TestCurrentTimestamp: """Tests for current_timestamp function.""" def test_current_timestamp_returns_datetime(self): """Test current_timestamp returns a datetime object.""" result = current_timestamp() assert isinstance(result, datetime) def test_current_timestamp_is_naive(self): """Test current_timestamp returns naive datetime (no timezone).""" result = current_timestamp() # The timestamp should be naive (no tzinfo) to match database storage assert result.tzinfo is None def test_current_timestamp_is_recent(self): """Test current_timestamp returns current time.""" before = datetime.now(tz=timezone.utc).replace(tzinfo=None) result = current_timestamp() after = datetime.now(tz=timezone.utc).replace(tzinfo=None) assert before <= result <= after or result == before def test_current_timestamp_uses_system_timezone(self): """Test current_timestamp uses SYSTEM_TIMEZONE constant.""" from abacus_models.core import constants # Get timestamp result = current_timestamp() # Verify it's in the expected timezone (UTC) by comparing to UTC now utc_now = datetime.now(tz=constants.SYSTEM_TIMEZONE).replace(tzinfo=None) local_now = datetime.now() # The result should be closer to UTC time than local time (if different) time_diff_utc = abs((result - utc_now).total_seconds()) time_diff_local = abs((result - local_now).total_seconds()) # Should be within 1 second of UTC assert time_diff_utc < 1.0 def test_current_timestamp_not_local_timezone(self): """Test current_timestamp is NOT using local timezone.""" import time # Get local timezone offset local_offset = time.timezone if not time.daylight else time.altzone local_offset_hours = -local_offset / 3600 # If we're not in UTC (offset != 0) if abs(local_offset_hours) > 0.1: result = current_timestamp() local_now = datetime.now() # Should NOT match local time exactly # Allow 1 second tolerance for execution time time_diff = abs((result - local_now).total_seconds()) # Should be different by roughly the timezone offset # (unless we happen to be in UTC) expected_diff = abs(local_offset_hours * 3600) # Allow 10% tolerance assert ( abs(time_diff - expected_diff) < expected_diff * 0.1 or time_diff < 1.0 ) def test_current_timestamp_consistency(self): """Test current_timestamp returns consistent values in sequence.""" timestamps = [current_timestamp() for _ in range(5)] # All should be within 1 second of each other for i in range(len(timestamps) - 1): diff = (timestamps[i + 1] - timestamps[i]).total_seconds() assert 0 <= diff < 1.0 class TestGetClass: """Tests for get_class function.""" def test_get_class_with_class_returns_class(self): """Test get_class returns the class when passed a class.""" result = get_class(TestModelFull) assert result is TestModelFull assert isinstance(result, type) def test_get_class_with_instance_returns_class(self, session, user_context): """Test get_class returns the class when passed an instance.""" instance = TestModelFull(name='Test') session.add(instance) session.flush() result = get_class(instance) assert result is TestModelFull assert isinstance(result, type) def test_get_class_preserves_type_with_class(self): """Test get_class preserves the correct type when passed a class.""" result = get_class(TestModelMinimal) assert result is TestModelMinimal assert result.__name__ == 'TestModelMinimal' def test_get_class_extracts_type_from_instance(self, session): """Test get_class extracts the correct type from an instance.""" instance = TestModelMinimal(name='Test') result = get_class(instance) assert result is TestModelMinimal assert result.__name__ == 'TestModelMinimal' def test_get_class_with_composite_key_model(self, session): """Test get_class works with composite key models.""" # Test with class result_class = get_class(TestModelCompositeKey) assert result_class is TestModelCompositeKey # Test with instance instance = TestModelCompositeKey(id1=1, id2=2, name='Test') result_instance = get_class(instance) assert result_instance is TestModelCompositeKey def test_get_class_return_type_consistency(self, session, user_context): """Test get_class returns the same type for class and instance.""" cls_result = get_class(TestModelFull) instance = TestModelFull(name='Test') session.add(instance) session.flush() instance_result = get_class(instance) assert cls_result is instance_result assert cls_result is TestModelFull class TestGetPk: """Tests for get_pk function.""" def test_get_pk_single_key(self, session): """Test get_pk returns primary key dict for single PK.""" model = TestModelFull(name='Test') session.add(model) session.commit() pk = get_pk(model) assert isinstance(pk, dict) assert 'id' in pk assert pk['id'] == model.id def test_get_pk_composite_key(self, session): """Test get_pk returns all primary keys for composite PK.""" model = TestModelCompositeKey(id1=1, id2=2, name='Test') session.add(model) session.commit() pk = get_pk(model) assert isinstance(pk, dict) assert 'id1' in pk assert 'id2' in pk assert pk['id1'] == 1 assert pk['id2'] == 2 def test_get_pk_none_value(self, session): """Test get_pk handles None PK values.""" model = TestModelFull(name='Test') # Before commit, id is None pk = get_pk(model) assert isinstance(pk, dict) assert 'id' in pk assert pk['id'] is None class TestGetPkColumns: """Tests for get_pk_columns function.""" def test_get_pk_columns_single_key_with_class(self): """Test get_pk_columns returns list with one column when passed a class.""" cols = get_pk_columns(TestModelFull) assert len(cols) == 1 assert cols[0].name == 'id' def test_get_pk_columns_single_key_with_instance(self, session, user_context): """Test get_pk_columns returns list with one column when passed an instance.""" instance = TestModelFull(name='Test') session.add(instance) session.flush() cols = get_pk_columns(instance) assert len(cols) == 1 assert cols[0].name == 'id' def test_get_pk_columns_composite_key_with_class(self): """Test get_pk_columns returns list with multiple columns when passed a class.""" cols = get_pk_columns(TestModelCompositeKey) assert len(cols) == 2 col_names = {col.name for col in cols} assert col_names == {'id1', 'id2'} def test_get_pk_columns_composite_key_with_instance(self, session): """Test get_pk_columns returns list with multiple columns when passed an instance.""" instance = TestModelCompositeKey(id1=1, id2=2, name='Test') session.add(instance) session.flush() cols = get_pk_columns(instance) assert len(cols) == 2 col_names = {col.name for col in cols} assert col_names == {'id1', 'id2'} class TestIsCreatable: """Tests for is_creatable function.""" def test_is_creatable_with_both_fields(self): """Test is_creatable returns True when both created_at and created_by exist.""" assert is_creatable(TestModelFull) is True def test_is_creatable_with_create_mixin(self): """Test is_creatable returns True for model with CreateMixin.""" assert is_creatable(TestModelCreateOnly) is True def test_is_creatable_without_fields(self): """Test is_creatable returns False when no creation fields exist.""" assert is_creatable(TestModelMinimal) is False def test_is_creatable_checks_instance(self, sample_full_model): """Test is_creatable works on instances.""" assert is_creatable(sample_full_model) is True class TestIsUpdatable: """Tests for is_updatable function.""" def test_is_updatable_with_both_fields(self): """Test is_updatable returns True when both last_modified fields exist.""" assert is_updatable(TestModelFull) is True def test_is_updatable_with_update_mixin(self): """Test is_updatable returns True for model with UpdateMixin.""" assert is_updatable(TestModelUpdateOnly) is True def test_is_updatable_without_fields(self): """Test is_updatable returns False when no update fields exist.""" assert is_updatable(TestModelMinimal) is False def test_is_updatable_checks_instance(self, sample_full_model): """Test is_updatable works on instances.""" assert is_updatable(sample_full_model) is True class TestIsSoftDeletable: """Tests for is_soft_deletable function.""" def test_is_soft_deletable_with_both_fields(self): """Test is_soft_deletable returns True when both deletion fields exist.""" assert is_soft_deletable(TestModelFull) is True def test_is_soft_deletable_with_soft_delete_mixin(self): """Test is_soft_deletable returns True for model with SoftDeleteMixin.""" assert is_soft_deletable(TestModelSoftDeleteOnly) is True def test_is_soft_deletable_without_fields(self): """Test is_soft_deletable returns False when no deletion fields exist.""" assert is_soft_deletable(TestModelMinimal) is False def test_is_soft_deletable_checks_instance(self, sample_full_model): """Test is_soft_deletable works on instances.""" assert is_soft_deletable(sample_full_model) is True class TestTryCreatable: """Tests for try_creatable function.""" def test_try_creatable_succeeds_with_fields(self): """Test try_creatable passes when model has creation fields.""" try_creatable(TestModelFull) # Should not raise def test_try_creatable_raises_without_fields(self): """Test try_creatable raises TypeError when model lacks creation fields.""" with pytest.raises(TypeError, match=ERROR_NOT_CREATABLE): try_creatable(TestModelMinimal) def test_try_creatable_checks_instance(self, sample_full_model): """Test try_creatable works on instances.""" try_creatable(sample_full_model) # Should not raise class TestTryUpdatable: """Tests for try_updatable function.""" def test_try_updatable_succeeds_with_fields(self): """Test try_updatable passes when model has update fields.""" try_updatable(TestModelFull) # Should not raise def test_try_updatable_raises_without_fields(self): """Test try_updatable raises TypeError when model lacks update fields.""" with pytest.raises(TypeError, match=ERROR_NOT_UPDATABLE): try_updatable(TestModelMinimal) def test_try_updatable_checks_instance(self, sample_full_model): """Test try_updatable works on instances.""" try_updatable(sample_full_model) # Should not raise class TestTrySoftDeletable: """Tests for try_soft_deletable function.""" def test_try_soft_deletable_succeeds_with_fields(self): """Test try_soft_deletable passes when model has deletion fields.""" try_soft_deletable(TestModelFull) # Should not raise def test_try_soft_deletable_raises_without_fields(self): """Test try_soft_deletable raises TypeError when model lacks deletion fields.""" with pytest.raises(TypeError, match=ERROR_NOT_SOFT_DELETABLE): try_soft_deletable(TestModelMinimal) def test_try_soft_deletable_checks_instance(self, sample_full_model): """Test try_soft_deletable works on instances.""" try_soft_deletable(sample_full_model) # Should not raise class TestTryPkColumn: """Tests for try_pk_column function.""" def test_try_pk_column_single_key(self, session): """Test try_pk_column returns column for single PK.""" model = TestModelFull(name='Test') session.add(model) session.commit() col = try_pk_column(model) assert col.name == 'id' def test_try_pk_column_composite_key_raises(self, session): """Test try_pk_column raises ValueError for composite PK.""" model = TestModelCompositeKey(id1=1, id2=2, name='Test') session.add(model) session.commit() with pytest.raises(ValueError) as exc_info: try_pk_column(model) error_msg = str(exc_info.value) assert 'TestModelCompositeKey' in error_msg assert '2' in error_msg # Should mention 2 columns def test_try_pk_column_error_message_format(self, session): """Test try_pk_column error message contains expected information.""" model = TestModelCompositeKey(id1=1, id2=2, name='Test') with pytest.raises(ValueError) as exc_info: try_pk_column(model) # Verify error message follows expected format error_msg = str(exc_info.value) assert ( ERROR_ENTITY_COMPOSITE_KEY_NOT_SUPPORTED.format( object_type='TestModelCompositeKey', num_cols=2 ) == error_msg )