"""Tests for data_processing module.""" import pandas as pd from io import StringIO from src.data_processing import ( validate_pub_song_id, validate_and_transform_timestamp, validate_and_transform_row, process_dataframe, ValidationError, ProcessedData, ) class TestValidatePubSongId: """Tests for Pub Song ID validation.""" def test_valid_integers(self): """Test valid integer values.""" valid, result = validate_pub_song_id(134224) assert valid is True assert result == 134224 valid, result = validate_pub_song_id("134224") assert valid is True assert result == 134224 def test_valid_float_integers(self): """Test float values that are actually integers.""" valid, result = validate_pub_song_id(134224.0) assert valid is True assert result == 134224 def test_invalid_values(self): """Test invalid values.""" test_cases = [None, "", "abc", -1, 0, "123.45"] for value in test_cases: valid, result = validate_pub_song_id(value) assert valid is False, f"Expected False for value {value}, got {valid}" assert result is None def test_non_integer_floats(self): """Test non-integer float values.""" # Float that becomes positive integer when converted should be valid valid, result = validate_pub_song_id(1.5) # int(1.5) = 1 assert valid is True assert result == 1 def test_nan_values(self): """Test NaN values.""" valid, result = validate_pub_song_id(pd.NA) assert valid is False assert result is None class TestValidateAndTransformTimestamp: """Tests for timestamp validation and transformation.""" def test_est_timezone(self): """Test EST timezone conversion.""" valid, result = validate_and_transform_timestamp("8/19/2025 3:20 PM EST") assert valid is True assert result == "2025-08-19 20:20" def test_edt_timezone(self): """Test EDT timezone conversion.""" valid, result = validate_and_transform_timestamp("8/19/2025 3:20 PM EDT") assert valid is True assert result == "2025-08-19 19:20" def test_other_timezones(self): """Test various timezone conversions.""" test_cases = [ ("8/19/2025 3:20 PM PST", "2025-08-19 23:20"), ("8/19/2025 3:20 PM PDT", "2025-08-19 22:20"), ("8/19/2025 3:20 PM CST", "2025-08-19 21:20"), ("8/19/2025 3:20 PM CDT", "2025-08-19 20:20"), ] for input_time, expected in test_cases: valid, result = validate_and_transform_timestamp(input_time) assert valid is True, f"Failed for {input_time}" assert result == expected, ( f"Expected {expected}, got {result} for {input_time}" ) def test_24_hour_format(self): """Test 24-hour format timestamps.""" valid, result = validate_and_transform_timestamp("08/19/2025 20:20") assert valid is True # Should default to Eastern timezone assert result in ["2025-08-20 00:20", "2025-08-20 01:20"] # EST or EDT def test_invalid_timestamps(self): """Test invalid timestamp values.""" invalid_cases = [ None, "", "invalid date", "13/45/2025 25:99 PM EST", pd.NA, ] for invalid_time in invalid_cases: valid, result = validate_and_transform_timestamp(invalid_time) assert valid is False assert result is None class TestValidateAndTransformRow: """Tests for row validation and transformation.""" def test_valid_row(self): """Test valid row processing.""" row = pd.Series({"Pub Song ID": 134224, "timestamp": "8/19/2025 3:20 PM EST"}) transformed_row, errors = validate_and_transform_row(row, 0) assert transformed_row is not None assert len(errors) == 0 assert transformed_row["pubSongId"] == 134224 assert transformed_row["deliveryDate"] == "2025-08-19 20:20" def test_invalid_pub_song_id(self): """Test row with invalid Pub Song ID.""" row = pd.Series( {"Pub Song ID": "invalid", "timestamp": "8/19/2025 3:20 PM EST"} ) transformed_row, errors = validate_and_transform_row(row, 0) assert transformed_row is None assert len(errors) == 1 assert errors[0].column == "Pub Song ID" def test_invalid_timestamp(self): """Test row with invalid timestamp.""" row = pd.Series({"Pub Song ID": 134224, "timestamp": "invalid timestamp"}) transformed_row, errors = validate_and_transform_row(row, 0) assert transformed_row is None assert len(errors) == 1 assert errors[0].column == "timestamp" def test_both_invalid(self): """Test row with both fields invalid.""" row = pd.Series({"Pub Song ID": "invalid", "timestamp": "invalid timestamp"}) transformed_row, errors = validate_and_transform_row(row, 0) assert transformed_row is None assert len(errors) == 2 class TestProcessDataframe: """Tests for DataFrame processing.""" def test_valid_dataframe(self): """Test processing valid DataFrame.""" csv_content = """Pub Song ID;timestamp 134224;8/19/2025 3:20 PM EST 134246;8/19/2025 3:20 PM EST 134229;8/19/2025 3:20 PM EST""" df = pd.read_csv(StringIO(csv_content), sep=";") result = process_dataframe(df) assert isinstance(result, ProcessedData) assert result.total_rows == 3 assert len(result.valid_rows) == 3 assert len(result.invalid_rows) == 0 assert not result.has_errors # Check transformed data assert all(result.valid_rows["deliveryDate"] == "2025-08-19 20:20") def test_mixed_valid_invalid(self): """Test processing DataFrame with mixed valid/invalid rows.""" data = { "Pub Song ID": [134224, "invalid", 134229], "timestamp": ["8/19/2025 3:20 PM EST", "8/19/2025 3:20 PM EST", "invalid"], } df = pd.DataFrame(data) result = process_dataframe(df) assert result.total_rows == 3 assert len(result.valid_rows) == 1 # Only first row is fully valid assert len(result.invalid_rows) == 2 assert result.has_errors assert len(result.errors) == 2 def test_empty_dataframe(self): """Test processing empty DataFrame.""" df = pd.DataFrame(columns=["Pub Song ID", "timestamp"]) result = process_dataframe(df) assert result.total_rows == 0 assert len(result.valid_rows) == 0 assert len(result.invalid_rows) == 0 assert not result.has_errors class TestValidationError: """Tests for ValidationError class.""" def test_error_string_representation(self): """Test string representation of ValidationError.""" error = ValidationError( row_index=0, column="Pub Song ID", value="invalid", error_message="Must be a positive integer", ) error_str = str(error) assert "Row 1" in error_str # Row index is 0-based, display is 1-based assert "Pub Song ID" in error_str assert "Must be a positive integer" in error_str assert "invalid" in error_str class TestProcessedData: """Tests for ProcessedData class.""" def test_properties(self): """Test ProcessedData properties.""" valid_df = pd.DataFrame( { "pubSongId": [1, 2], "deliveryDate": ["2025-01-01 12:00", "2025-01-01 13:00"], } ) invalid_df = pd.DataFrame({"Pub Song ID": [3], "timestamp": ["invalid"]}) errors = [ValidationError(0, "test", "value", "message")] result = ProcessedData(valid_df, invalid_df, errors) assert result.total_rows == 3 assert result.has_errors is True assert len(result.valid_rows) == 2 assert len(result.invalid_rows) == 1 assert len(result.errors) == 1