from base64 import b32encode, b64encode import faker import pytest from fansifter_common.email import ( EmailType, decode_local_part_to_email, decode_reply_to_local_part, encode_email_to_local_part, encode_reply_to_local_part, ) # Tests for encode_email_to_local_part def test_encode_email_to_local_part_v1(faker: faker.Faker) -> None: email_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v1", ) expected_b64 = b64encode(email_id.encode()).decode() assert encoded == f"{expected_b64}.v1" def test_encode_email_to_local_part_v2_campaign(faker: faker.Faker) -> None: email_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v2", ) expected_b64 = b64encode(f"camp.{email_id}".encode()).decode() assert encoded == f"{expected_b64}.v2" def test_encode_email_to_local_part_v2_automated_without_trigger( faker: faker.Faker, ) -> None: email_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=None, version="v2", ) expected_b64 = b64encode(f"auto.{email_id}".encode()).decode() assert encoded == f"{expected_b64}.v2" def test_encode_email_to_local_part_v2_automated_with_trigger( faker: faker.Faker, ) -> None: email_id = faker.pystr() automated_email_trigger_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=automated_email_trigger_id, version="v2", ) expected_b64 = b64encode( f"auto.{email_id}.{automated_email_trigger_id}".encode() ).decode() assert encoded == f"{expected_b64}.v2" def test_encode_email_to_local_part_v2_invalid_type(faker: faker.Faker) -> None: with pytest.raises(ValueError, match="Invalid email type"): encode_email_to_local_part( email_id=faker.pystr(), email_type="INVALID", # type: ignore automated_email_trigger_id=None, version="v2", ) # Tests for decode_local_part_to_email def test_decode_local_part_v1(faker: faker.Faker) -> None: email_id = faker.pystr() encoded_b64 = b64encode(email_id.encode()).decode() local_part = f"{encoded_b64}.v1" result = decode_local_part_to_email(local_part) assert result.email_id == email_id assert result.email_type == "CAMPAIGN" assert result.automated_email_trigger_id is None assert result.version == "v1" def test_decode_local_part_v2_campaign(faker: faker.Faker) -> None: email_id = faker.pystr() encoded_b64 = b64encode(f"camp.{email_id}".encode()).decode() local_part = f"{encoded_b64}.v2" result = decode_local_part_to_email(local_part) assert result.email_id == email_id assert result.email_type == "CAMPAIGN" assert result.automated_email_trigger_id is None assert result.version == "v2" def test_decode_local_part_v2_automated_without_trigger(faker: faker.Faker) -> None: email_id = faker.pystr() encoded_b64 = b64encode(f"auto.{email_id}".encode()).decode() local_part = f"{encoded_b64}.v2" result = decode_local_part_to_email(local_part) assert result.email_id == email_id assert result.email_type == "AUTOMATED" assert result.automated_email_trigger_id is None assert result.version == "v2" def test_decode_local_part_v2_automated_with_trigger(faker: faker.Faker) -> None: email_id = faker.pystr() automated_email_trigger_id = faker.pystr() encoded_b64 = b64encode( f"auto.{email_id}.{automated_email_trigger_id}".encode() ).decode() local_part = f"{encoded_b64}.v2" result = decode_local_part_to_email(local_part) assert result.email_id == email_id assert result.email_type == "AUTOMATED" assert result.automated_email_trigger_id == automated_email_trigger_id assert result.version == "v2" def test_decode_local_part_invalid_format(faker: faker.Faker) -> None: with pytest.raises(ValueError, match="Invalid local part format"): decode_local_part_to_email(faker.pystr()) def test_decode_local_part_invalid_version(faker: faker.Faker) -> None: local_part = f"{faker.pystr()}.v4" with pytest.raises(ValueError, match="Invalid local part format"): decode_local_part_to_email(local_part) def test_decode_local_part_invalid_v2_format() -> None: # Only one part instead of at least 2 encoded_b64 = b64encode(b"onlyonepart").decode() local_part = f"{encoded_b64}.v2" with pytest.raises(ValueError, match="Invalid v2 format"): decode_local_part_to_email(local_part) def test_decode_local_part_invalid_email_type_prefix(faker: faker.Faker) -> None: email_id = faker.pystr() encoded_b64 = b64encode(f"invalid.{email_id}".encode()).decode() local_part = f"{encoded_b64}.v2" with pytest.raises(ValueError, match="Invalid email type prefix"): decode_local_part_to_email(local_part) # Roundtrip tests def test_encode_decode_roundtrip_v1(faker: faker.Faker) -> None: email_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v1", ) decoded = decode_local_part_to_email(encoded) assert decoded.email_id == email_id assert decoded.email_type == "CAMPAIGN" assert decoded.automated_email_trigger_id is None assert decoded.version == "v1" def test_encode_decode_roundtrip_v2_campaign(faker: faker.Faker) -> None: email_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v2", ) decoded = decode_local_part_to_email(encoded) assert decoded.email_id == email_id assert decoded.email_type == "CAMPAIGN" assert decoded.automated_email_trigger_id is None assert decoded.version == "v2" def test_encode_decode_roundtrip_v2_automated_with_trigger(faker: faker.Faker) -> None: email_id = faker.pystr() automated_email_trigger_id = faker.pystr() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=automated_email_trigger_id, version="v2", ) decoded = decode_local_part_to_email(encoded) assert decoded.email_id == email_id assert decoded.email_type == "AUTOMATED" assert decoded.automated_email_trigger_id == automated_email_trigger_id assert decoded.version == "v2" def test_encode_email_to_local_part_v3_campaign(faker: faker.Faker) -> None: email_id = faker.uuid4() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, version="v3", ) expected = b32encode(f"camp.{email_id}".encode()).decode().lower() assert encoded == f"{expected}.v3" def test_encode_email_to_local_part_v3_automated_with_trigger( faker: faker.Faker, ) -> None: email_id = faker.uuid4() trigger_id = faker.md5() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=trigger_id, version="v3", ) expected = b32encode(f"auto.{email_id}.{trigger_id}".encode()).decode().lower() assert encoded == f"{expected}.v3" def test_decode_local_part_v3_campaign(faker: faker.Faker) -> None: email_id = faker.uuid4() encoded = b32encode(f"camp.{email_id}".encode()).decode().lower() local_part = f"{encoded}.v3" result = decode_local_part_to_email(local_part) assert result.email_id == email_id assert result.email_type == EmailType.CAMPAIGN assert result.automated_email_trigger_id is None assert result.version == "v3" def test_decode_local_part_v3_survives_lowercasing(faker: faker.Faker) -> None: """Email providers may lowercase the local part; v3 must survive this.""" email_id = faker.uuid4() local_part = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, version="v3", ) result = decode_local_part_to_email(local_part.lower()) assert result.email_id == email_id assert result.email_type == EmailType.CAMPAIGN assert result.version == "v3" def test_encode_decode_roundtrip_v3_automated_with_trigger(faker: faker.Faker) -> None: email_id = faker.uuid4() trigger_id = faker.md5() encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=trigger_id, version="v3", ) result = decode_local_part_to_email(encoded) assert result.email_id == email_id assert result.email_type == EmailType.AUTOMATED assert result.automated_email_trigger_id == trigger_id assert result.version == "v3" def test_encode_email_to_local_part_without_trigger_id(faker: faker.Faker) -> None: """Test that automated_email_trigger_id is optional (NotRequired).""" email_id = faker.pystr() # Should work without automated_email_trigger_id key encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, version="v2", ) expected_b64 = b64encode(f"camp.{email_id}".encode()).decode() assert encoded == f"{expected_b64}.v2" def test_encode_with_email_type_enum(faker: faker.Faker) -> None: """Test that encode_email_to_local_part works with EmailType enum.""" email_id = faker.pystr() # Use EmailType enum directly encoded = encode_email_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, # Use enum instead of string version="v2", ) expected_b64 = b64encode(f"camp.{email_id}".encode()).decode() assert encoded == f"{expected_b64}.v2" def test_decode_returns_email_type_enum(faker: faker.Faker) -> None: """Test that decode_local_part_to_email returns EmailType enum.""" email_id = faker.pystr() encoded_b64 = b64encode(f"camp.{email_id}".encode()).decode() local_part = f"{encoded_b64}.v2" result = decode_local_part_to_email(local_part) assert result.email_type == EmailType.CAMPAIGN class TestReplyToEncoding: def test_encode_campaign(self, faker: faker.Faker) -> None: email_id = faker.uuid4() key = faker.uuid4() result = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key=key, ) assert result def test_encode_automated_without_trigger(self, faker: faker.Faker) -> None: email_id = faker.uuid4() result = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, key=faker.uuid4(), ) assert result def test_encode_automated_with_trigger(self, faker: faker.Faker) -> None: email_id = faker.uuid4() trigger_id = faker.md5() result = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, key=faker.uuid4(), automated_email_trigger_id=trigger_id, ) assert result def test_encode_invalid_type_raises(self, faker: faker.Faker) -> None: with pytest.raises(ValueError, match="Invalid email type"): encode_reply_to_local_part( email_id=faker.uuid4(), email_type="INVALID", # type: ignore key=faker.uuid4(), ) def test_encode_different_keys_produce_different_results( self, faker: faker.Faker ) -> None: email_id = faker.uuid4() result1 = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key="key1", ) result2 = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key="key2", ) assert result1 != result2 def test_encode_same_for_all_fans(self, faker: faker.Faker) -> None: """Signature does not depend on fan email - same result for all fans.""" email_id = faker.uuid4() key = faker.uuid4() result1 = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key=key, ) result2 = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key=key, ) assert result1 == result2 def test_decode_campaign(self, faker: faker.Faker) -> None: email_id = faker.uuid4() key = faker.uuid4() local_part = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key=key, ) result = decode_reply_to_local_part(local_part, key=key) assert result.email_id == email_id assert result.email_type == EmailType.CAMPAIGN assert result.automated_email_trigger_id is None assert result.version == "v2" def test_decode_automated_without_trigger(self, faker: faker.Faker) -> None: email_id = faker.uuid4() key = faker.uuid4() local_part = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, key=key, ) result = decode_reply_to_local_part(local_part, key=key) assert result.email_id == email_id assert result.email_type == EmailType.AUTOMATED assert result.automated_email_trigger_id is None assert result.version == "v2" def test_decode_automated_with_trigger(self, faker: faker.Faker) -> None: email_id = faker.uuid4() trigger_id = faker.md5() key = faker.uuid4() local_part = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, key=key, automated_email_trigger_id=trigger_id, ) result = decode_reply_to_local_part(local_part, key=key) assert result.email_id == email_id assert result.email_type == EmailType.AUTOMATED assert result.automated_email_trigger_id == trigger_id assert result.version == "v2" def test_decode_wrong_key_raises(self, faker: faker.Faker) -> None: local_part = encode_reply_to_local_part( email_id=faker.uuid4(), email_type=EmailType.CAMPAIGN, key="correct-key", ) with pytest.raises(ValueError, match="Invalid signature"): decode_reply_to_local_part(local_part, key="wrong-key") def test_decode_invalid_email_type_prefix_raises(self, faker: faker.Faker) -> None: with pytest.raises(ValueError, match="Invalid email type prefix"): decode_reply_to_local_part( "invalid.email_id.somesignature", key=faker.uuid4() ) def test_roundtrip_campaign(self, faker: faker.Faker) -> None: email_id = faker.uuid4() key = faker.uuid4() local_part = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.CAMPAIGN, key=key, ) result = decode_reply_to_local_part(local_part, key=key) assert result.email_id == email_id assert result.email_type == EmailType.CAMPAIGN assert result.automated_email_trigger_id is None def test_roundtrip_automated_with_trigger(self, faker: faker.Faker) -> None: email_id = faker.uuid4() trigger_id = faker.md5() key = faker.uuid4() local_part = encode_reply_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, key=key, automated_email_trigger_id=trigger_id, ) result = decode_reply_to_local_part(local_part, key=key) assert result.email_id == email_id assert result.email_type == EmailType.AUTOMATED assert result.automated_email_trigger_id == trigger_id