import io from unittest.mock import MagicMock import pandas as pd from backend.actions.file_upload import ( INPUT_FILE_COLUMNS, build_campaign_id, check_data_quality, filter_options_df, normalize_columns, prepare_upload_dataframe, submit_upload, validate_fan_data, validate_upload_columns, ) from backend.dtos import FileUploadFilterOption VALID_CSV_HEADER = ",".join(INPUT_FILE_COLUMNS) def _valid_csv_row(email: str = "fan@example.com", optin: str = "yes") -> str: values = {col: "" for col in INPUT_FILE_COLUMNS} values["EMAIL_REQUIRED"] = email values["OPTIN_REQUIRED"] = optin values["GENDER"] = "female" values["COUNTRY_REGION"] = "US" return ",".join(values[col] for col in INPUT_FILE_COLUMNS) def _valid_csv() -> io.BytesIO: content = f"{VALID_CSV_HEADER}\n{_valid_csv_row()}" return io.BytesIO(content.encode("utf-8")) # --- normalize_columns --- def test_normalize_columns_strips_and_uppercases(): df = pd.DataFrame(columns=[" email required ", "Optin-Required"]) normalized = normalize_columns(df) assert "EMAIL_REQUIRED" in normalized.columns assert "OPTINREQUIRED" in normalized.columns # --- validate_upload_columns --- def test_validate_upload_columns_accepts_exact_match(): df = pd.DataFrame(columns=INPUT_FILE_COLUMNS) assert validate_upload_columns(df) == [] def test_validate_upload_columns_rejects_missing_column(): columns = [c for c in INPUT_FILE_COLUMNS if c != "GENDER"] df = pd.DataFrame(columns=columns) errors = validate_upload_columns(df) assert len(errors) == 1 assert "wrong column names" in errors[0] def test_validate_upload_columns_rejects_missing_optin(): df = pd.DataFrame(columns=INPUT_FILE_COLUMNS) df.loc[0] = [None] * len(INPUT_FILE_COLUMNS) errors = validate_upload_columns(df) assert len(errors) == 1 assert "opt-in" in errors[0] # --- build_campaign_id --- def test_build_campaign_id_deterministic(): id1 = build_campaign_id("Label", "Artist", "Description") id2 = build_campaign_id("Label", "Artist", "Description") assert id1 == id2 def test_build_campaign_id_differs_per_input(): id1 = build_campaign_id("Label", "Artist", "Description") id2 = build_campaign_id("Other Label", "Artist", "Description") assert id1 != id2 # --- check_data_quality --- def test_check_data_quality_no_warnings_when_clean(): df = pd.DataFrame({"EMAIL_REQUIRED": ["a@example.com", "b@example.com"]}) assert check_data_quality(df, "EMAIL_REQUIRED") == [] def test_check_data_quality_flags_duplicates(): df = pd.DataFrame({"EMAIL_REQUIRED": ["a@example.com", "a@example.com"]}) warnings = check_data_quality(df, "EMAIL_REQUIRED") assert any("duplicate" in w.lower() for w in warnings) def test_check_data_quality_flags_missing_values(): df = pd.DataFrame({"EMAIL_REQUIRED": ["a@example.com", None]}) warnings = check_data_quality(df, "EMAIL_REQUIRED") assert any("missing" in w.lower() for w in warnings) def test_check_data_quality_flags_all_missing_as_error(): df = pd.DataFrame({"EMAIL_REQUIRED": [None, None]}) warnings = check_data_quality(df, "EMAIL_REQUIRED") assert any(w.startswith("Error!") for w in warnings) # --- filter_options_df --- def test_filter_options_df_converts_to_dataframe(): options = [ FileUploadFilterOption( vendor_id="v1", label_name="Label A", global_participant_id="g1", virtual_participant_id=None, artist_name="Artist A", mailing_list_id="m1", mailing_list_name="List A", type="artist", ) ] df = filter_options_df(options) assert list(df["label_name"]) == ["Label A"] assert list(df["artist_name"]) == ["Artist A"] assert list(df["mailing_list_name"]) == ["List A"] # --- prepare_upload_dataframe --- def test_prepare_upload_dataframe_valid_csv(): dataframe, errors = prepare_upload_dataframe( _valid_csv(), "upload.csv", "Label A", "Artist A", "List A", "Some description", "Spotify", "No", "Website", False, "TEST_USER", "v1", "g1", None, ) assert errors == [] assert dataframe["FILE_NAME"].iloc[0] == "upload.csv" assert dataframe["LABEL"].iloc[0] == "Label A" assert dataframe["ARTIST"].iloc[0] == "Artist A" assert dataframe["MAILING_LIST"].iloc[0] == "List A" assert dataframe["TLA_ID"].iloc[0] is None assert dataframe["VENDOR_ID"].iloc[0] == "v1" assert dataframe["GLOBAL_PARTICIPANT_ID"].iloc[0] == "g1" assert dataframe["VIRTUAL_PARTICIPANT_ID"].iloc[0] is None assert "LABEL_REQUIRED" not in dataframe.columns assert "TERRITORY_2_DIGIT_ISO_REQUIRED" not in dataframe.columns assert dataframe["CAMPAIGN_ID"].iloc[0] == build_campaign_id( "Label A", "Artist A", "Some description" ) def test_prepare_upload_dataframe_defaults_ids_to_none(): dataframe, errors = prepare_upload_dataframe( _valid_csv(), "upload.csv", "Label A", "Artist A", "List A", "Some description", "Spotify", "No", "Website", False, "TEST_USER", ) assert errors == [] assert dataframe["VENDOR_ID"].iloc[0] is None assert dataframe["GLOBAL_PARTICIPANT_ID"].iloc[0] is None assert dataframe["VIRTUAL_PARTICIPANT_ID"].iloc[0] is None def test_prepare_upload_dataframe_invalid_columns_returns_errors(): csv_content = io.BytesIO(b"WRONG_COLUMN\nvalue") dataframe, errors = prepare_upload_dataframe( csv_content, "upload.csv", "Label A", "Artist A", "List A", "Some description", "Spotify", "No", "Website", False, "TEST_USER", ) assert len(errors) == 1 def test_prepare_upload_dataframe_invalid_gender_becomes_missing(): content = f"{VALID_CSV_HEADER}\n{_valid_csv_row()}".replace( "female", "not-a-gender" ) dataframe, errors = prepare_upload_dataframe( io.BytesIO(content.encode("utf-8")), "upload.csv", "Label A", "Artist A", "List A", "Some description", "Spotify", "No", "Website", False, "TEST_USER", ) assert errors == [] assert pd.isna(dataframe["GENDER"].iloc[0]) # --- submit_upload --- def _mock_session_for_submit(existing_filenames=None, existing_campaign_ids=None): session = MagicMock() filenames_df = MagicMock() filenames_df.values.flatten.return_value.tolist.return_value = ( existing_filenames or [] ) campaign_ids_df = MagicMock() campaign_ids_df.values.flatten.return_value.tolist.return_value = ( existing_campaign_ids or [] ) session.table.return_value.select.return_value.distinct.return_value.to_pandas.side_effect = [ filenames_df, campaign_ids_df, ] return session def test_submit_upload_rejects_duplicate_filename(): session = _mock_session_for_submit(existing_filenames=["upload.csv"]) dataframe = pd.DataFrame({"CAMPAIGN_ID": ["abc123"]}) result = submit_upload(session, "QA", dataframe, "upload.csv") assert result.success is False assert "already been uploaded" in result.errors[0] session.create_dataframe.return_value.write.save_as_table.assert_not_called() def test_submit_upload_rejects_duplicate_campaign_id(): session = _mock_session_for_submit( existing_filenames=[], existing_campaign_ids=["abc123"] ) dataframe = pd.DataFrame({"CAMPAIGN_ID": ["abc123"]}) result = submit_upload(session, "QA", dataframe, "upload.csv") assert result.success is False assert "already exists" in result.errors[0] def test_submit_upload_inserts_when_valid(): session = _mock_session_for_submit(existing_filenames=[], existing_campaign_ids=[]) dataframe = pd.DataFrame({"CAMPAIGN_ID": ["abc123"]}) result = submit_upload(session, "QA", dataframe, "upload.csv") assert result.success is True assert result.campaign_id == "abc123" assert result.rows_uploaded == 1 def test_submit_upload_rejects_empty_dataframe(): session = MagicMock() result = submit_upload(session, "QA", pd.DataFrame(), "upload.csv") assert result.success is False assert "No valid rows" in result.errors[0] session.table.assert_not_called() # --- validate_fan_data --- def _valid_row(**overrides: object) -> dict: row = { "EMAIL_REQUIRED": "fan@example.com", "MOBILE_PHONE": None, "FIRST_NAME": "Jane", "LAST_NAME": "Doe", "BIRTHDATE_MM_DD_YYYY": "01/01/2000", "BIRTHDAY_MM_DD_US_ONLY": "01/01", "GENDER": "female", "ADDRESS_1": "123 Main St", "CITY": "Anytown", "STATE": "CA", "COUNTRY_REGION": "US", "POSTAL_CODE": "12345", "PREFERRED_LANGUAGE": "English", "FACEBOOK_PAGE": "facebook.com/janedoe", "TWITTER_HANDLE": "@janedoe", "DATE_CREATED": "2024-01-01T00:00:00+00:00", "CRM_GENERATED": "No", "DSP": "Spotify", "FILE_SOURCE_DESCRIPTION": "Some source", "ARTIST": "Artist A", "TERRITORY": "US", "CAMPAIGN_ID": "campaign-1", "CAMPAIGN_NAME": "Campaign", "OVERWRITE_FAN_DATA": False, } row.update(overrides) return row def test_validate_fan_data_accepts_valid_rows(): dataframe = pd.DataFrame([_valid_row()]) valid_dataframe, failures = validate_fan_data(dataframe) assert len(valid_dataframe) == 1 assert failures == [] def test_validate_fan_data_rejects_invalid_email(): dataframe = pd.DataFrame([_valid_row(EMAIL_REQUIRED="not-an-email")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "Invalid Email" and f.count == 1 for f in failures) def test_validate_fan_data_requires_email_or_mobile_phone(): dataframe = pd.DataFrame([_valid_row(EMAIL_REQUIRED=None, MOBILE_PHONE=None)]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any( f.reason == "Either email OR mobile_phone should be present" for f in failures ) def test_validate_fan_data_accepts_valid_mobile_phone_without_email(): dataframe = pd.DataFrame( [_valid_row(EMAIL_REQUIRED=None, MOBILE_PHONE="+14155552671")] ) valid_dataframe, failures = validate_fan_data(dataframe) assert len(valid_dataframe) == 1 assert failures == [] def test_validate_fan_data_rejects_invalid_mobile_phone(): dataframe = pd.DataFrame([_valid_row(MOBILE_PHONE="123")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "mobile_phone is invalid" for f in failures) def test_validate_fan_data_rejects_invalid_birthdate(): dataframe = pd.DataFrame([_valid_row(BIRTHDATE_MM_DD_YYYY="not-a-date")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "birthdate_mm_dd_yyyy is invalid" for f in failures) def test_validate_fan_data_rejects_invalid_birthday_mm_dd(): dataframe = pd.DataFrame([_valid_row(BIRTHDAY_MM_DD_US_ONLY="02/30")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "birthday_mm_dd_us_only is invalid" for f in failures) def test_validate_fan_data_rejects_first_name_too_long(): dataframe = pd.DataFrame([_valid_row(FIRST_NAME="x" * 111)]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "first_name is too long" for f in failures) def test_validate_fan_data_rejects_last_name_too_long(): dataframe = pd.DataFrame([_valid_row(LAST_NAME="x" * 111)]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "last_name is too long" for f in failures) def test_validate_fan_data_rejects_invalid_gender(): dataframe = pd.DataFrame([_valid_row(GENDER="not-a-gender")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "Invalid gender" for f in failures) def test_validate_fan_data_rejects_fields_that_are_too_long(): too_long_values = { "ADDRESS_1": ("x" * 256, "address is too long"), "CITY": ("x" * 76, "city is too long"), "STATE": ("x" * 256, "state is too long"), "COUNTRY_REGION": ("x" * 57, "country_region is too long"), "POSTAL_CODE": ("x" * 21, "postal_code is too long"), "PREFERRED_LANGUAGE": ("x" * 256, "preferred_language is too long"), "FACEBOOK_PAGE": ("x" * 256, "facebook_page is too long"), "TWITTER_HANDLE": ("x" * 101, "twitter_handle is too long"), } for column, (value, expected_reason) in too_long_values.items(): dataframe = pd.DataFrame([_valid_row(**{column: value})]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty, f"{column} should have failed validation" assert any(f.reason == expected_reason for f in failures), column def test_validate_fan_data_rejects_empty_date_created(): dataframe = pd.DataFrame([_valid_row(DATE_CREATED=None)]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "Empty date_created" for f in failures) def test_validate_fan_data_rejects_invalid_date_created(): dataframe = pd.DataFrame([_valid_row(DATE_CREATED="not-a-timestamp")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "date_created is invalid" for f in failures) def test_validate_fan_data_rejects_invalid_crm_generated(): dataframe = pd.DataFrame([_valid_row(CRM_GENERATED="Maybe")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "crm_generated is invalid" for f in failures) def test_validate_fan_data_rejects_dsp_too_long(): dataframe = pd.DataFrame([_valid_row(DSP="x" * 256)]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "dsp is invalid" for f in failures) def test_validate_fan_data_rejects_empty_required_fields(): required_field_messages = { "FILE_SOURCE_DESCRIPTION": "Empty file_source_description", "ARTIST": "Empty artist", "TERRITORY": "Empty territory", "CAMPAIGN_ID": "Empty campaign_id", "CAMPAIGN_NAME": "Empty campaign_name", } for column, expected_reason in required_field_messages.items(): dataframe = pd.DataFrame([_valid_row(**{column: None})]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty, f"{column} should have failed validation" assert any(f.reason == expected_reason for f in failures), column def test_validate_fan_data_rejects_invalid_overwrite_fan_data(): dataframe = pd.DataFrame([_valid_row(OVERWRITE_FAN_DATA="yes")]) valid_dataframe, failures = validate_fan_data(dataframe) assert valid_dataframe.empty assert any(f.reason == "overwrite_fan_data is invalid" for f in failures) def test_validate_fan_data_excludes_only_failing_rows_and_counts_reasons(): dataframe = pd.DataFrame( [ _valid_row(), _valid_row(EMAIL_REQUIRED="not-an-email"), _valid_row(EMAIL_REQUIRED="also-not-an-email"), ] ) valid_dataframe, failures = validate_fan_data(dataframe) assert len(valid_dataframe) == 1 assert len(failures) == 1 assert failures[0].reason == "Invalid Email" assert failures[0].count == 2 def test_validate_fan_data_empty_dataframe_returns_no_failures(): valid_dataframe, failures = validate_fan_data(pd.DataFrame()) assert valid_dataframe.empty assert failures == []