"""Unit tests: recipient CSV/TXT parsing, validation, duplicates, suppression, caps.""" import pytest from app.services.recipient_service import ( ImportReport, is_valid_email, parse_csv, parse_txt, validate_rows, ) class TestEmailValidation: @pytest.mark.parametrize("good", [ "alice@example.com", "bob.smith@work.co.uk", "c+tag@domain.org", "first_last@sub.domain.io", "x@y.co", ]) def test_valid_addresses(self, good): assert is_valid_email(good) is True @pytest.mark.parametrize("bad", [ "", "not-an-email", "@domain.com", "user@", "user @domain.com", "user@@domain.com", "user@domain", "a" * 400 + "@x.com", ]) def test_invalid_addresses(self, bad): assert is_valid_email(bad) is False class TestTxtParsing: def test_one_email_per_line(self): rows = parse_txt("a@x.com\nb@y.com\n\n c@z.com \n") assert rows == ["a@x.com", "b@y.com", "c@z.com"] def test_comma_separated_with_names(self): rows = parse_txt("a@x.com, Alice, Smith\nb@y.com,Bob,Jones") assert rows == ["a@x.com, Alice, Smith", "b@y.com,Bob,Jones"] def test_comments_and_blank_skipped(self): rows = parse_txt("# header comment\n\na@x.com\n# another\nb@y.com") assert rows == ["a@x.com", "b@y.com"] def test_empty(self): assert parse_txt("") == [] class TestCsvParsing: def test_standard_headers(self): csv_data = ("Email,First Name,Last Name,Company Name,Company Phone,Customer Email\n" "a@x.com,Alice,Smith,Acme,+1555,acust@x.com\n" "b@y.com,Bob,Jones,Zeta,+1556,bcust@y.com") records, headers, errors = parse_csv(csv_data) assert errors == [] assert len(records) == 2 assert records[0]["email"] == "a@x.com" assert records[0]["first_name"] == "Alice" assert records[0]["company_name"] == "Acme" assert records[0]["customer_email"] == "acust@x.com" def test_alias_headers(self): csv_data = "E-Mail,FName,LName\na@x.com,Al,Riv" records, _, _ = parse_csv(csv_data) assert records[0]["email"] == "a@x.com" assert records[0]["first_name"] == "Al" assert records[0]["last_name"] == "Riv" def test_custom_columns_captured(self): csv_data = "Email,OrderRef,Plan Tier\na@x.com,ORD-1,premium" records, _, _ = parse_csv(csv_data) assert records[0]["custom_fields"] == {"OrderRef": "ORD-1", "Plan Tier": "premium"} def test_empty_rows_reported(self): csv_data = "Email\na@x.com\n,,,\nb@y.com" records, _, errors = parse_csv(csv_data) assert len(records) == 2 assert any("empty row" in e for e in errors) def test_empty_file(self): records, headers, errors = parse_csv("") assert records == [] and errors == ["File is empty"] def test_json_content_not_parsed_as_recipients(self): # JSON is never accepted at the router layer (content sniff + extension); # the CSV parser itself yields no usable records for JSON content. records, _, _ = parse_csv('{"email": "a@x.com"}') assert records == [] or all(not r.get("email") for r in records) class TestValidateRows: def test_clean_import(self): records, _, _ = parse_csv("Email,First Name\na@x.com,Al\nb@y.com,Bo") rpt = validate_rows(records, set(), set()) assert rpt.summary()["valid_recipients"] == 2 assert len(rpt.valid) == 2 assert rpt.valid[0]["first_name"] == "Al" def test_duplicates_in_file(self): records, _, _ = parse_csv("Email\nA@X.com\na@x.com\nb@y.com") rpt = validate_rows(records, set(), set()) # duplicates are case-insensitive assert len(rpt.duplicates) == 1 assert len(rpt.valid) == 2 def test_duplicates_against_existing(self): records, _, _ = parse_csv("Email\na@x.com\nb@y.com") rpt = validate_rows(records, set(), set(), existing_emails={"a@x.com"}) assert len(rpt.duplicates) == 1 assert len(rpt.valid) == 1 def test_suppressed_rejected(self): records, _, _ = parse_csv("Email\nsup@x.com\nok@y.com") rpt = validate_rows(records, {"sup@x.com"}, set()) assert len(rpt.suppressed) == 1 assert len(rpt.valid) == 1 def test_bounced_rejected(self): records, _, _ = parse_csv("Email\nbounce@x.com\nok@y.com") rpt = validate_rows(records, set(), {"bounce@x.com"}) assert len(rpt.bounced) == 1 assert len(rpt.valid) == 1 def test_invalid_syntax_rejected(self): records, _, _ = parse_csv("Email\nnot-an-email\nok@y.com") rpt = validate_rows(records, set(), set()) assert len(rpt.invalid) == 1 assert rpt.invalid[0]["reason"] == "Invalid email syntax" def test_missing_email_rejected(self): records, _, _ = parse_csv("Email,First Name\n,Al\nok@y.com,Bo") rpt = validate_rows(records, set(), set()) assert len(rpt.invalid) == 1 assert "Missing email" in rpt.invalid[0]["reason"] def test_max_recipients_cap(self): rows = "\n".join(f"u{i}@x.com" for i in range(2500)) records, _, _ = parse_csv("Email\n" + rows) rpt = validate_rows(records, set(), set(), max_recipients=2000) assert len(rpt.valid) == 2000 assert any("maximum of 2000" in e for e in rpt.errors) def test_case_normalization(self): records, _, _ = parse_csv("Email\nMiXeD@Case.COM") rpt = validate_rows(records, set(), set()) assert rpt.valid[0]["email"] == "mixed@case.com" def test_customer_email_defaults_to_email(self): records, _, _ = parse_csv("Email\na@x.com") rpt = validate_rows(records, set(), set()) assert rpt.valid[0]["customer_email"] == "a@x.com" def test_summary_shape(self): records, _, _ = parse_csv("Email\na@x.com") s = validate_rows(records, set(), set()).summary() for k in ("total_imported", "valid_recipients", "invalid_recipients", "duplicate_recipients", "suppressed_recipients", "bounced_recipients", "errors"): assert k in s class TestUnsubToken: def test_unique_tokens(self): from app.services.recipient_service import make_unsub_token toks = {make_unsub_token() for _ in range(100)} assert len(toks) == 100 assert all(len(t) >= 32 for t in toks)