from datetime import date import faker from dirty_equals import IsDate, IsDatetime, IsJson, IsList from preference_center.profile.enums import Gender from preference_center.profile.events import ( ProfileDeletedV1, ProfileSubscriptionsUpdatedV1, ProfileUpdatedV1, ) from preference_center.profile.models import Profile, Subscription from preference_center.profile.types import ProfileValues from tests.unit.types import BuildModel class TestProfileUpdatedV1: def test_json_schema(self) -> None: assert ProfileUpdatedV1.model_json_schema(by_alias=True) == { "title": "ProfileUpdatedV1", "type": "object", "properties": { "crmId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "CRM ID", }, "id": { "description": "Unique profile identifier in UUID format", "title": "ID", "type": "string", }, "updatedAt": { "format": "date-time", "title": "Updated at", "type": "string", }, "updatedFields": { "items": {"$ref": "#/$defs/ProfileUpdateField"}, "title": "Updated fields", "type": "array", "uniqueItems": True, }, "updatedBy": { "anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Updated by", }, }, "required": ["id", "crmId", "updatedAt", "updatedFields"], "$defs": { "ProfileUpdateField": { "properties": { "fieldName": { "description": "The target field or column name", "enum": [ "firstName", "lastName", "email", "phoneNumber", "dateOfBirth", "birthday", "gender", "countryCode", "city", "address", "zipCode", ], "title": "Field name", "type": "string", }, "newValue": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "New value", }, "oldValue": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "Old value", }, }, "required": ["fieldName", "oldValue", "newValue"], "title": "ProfileUpdateField", "type": "object", } }, } def test_from_profile(self, build_model: BuildModel) -> None: # Old values old_values = ProfileValues( email="old@mail.com", first_name="Peter", last_name="Parker", phone_number="+37255555555", date_of_birth=date(1990, 6, 16), birthday="06/16", gender=Gender.MALE, country_code="EE", city="Tallinn", address="Street 1", zip_code="12345", ) profile = build_model(Profile, updated_by=None, **old_values) # Updated values new_values = ProfileValues( email="new@mail.com", first_name="Mary", last_name="Jane", phone_number="+37266666666", date_of_birth=date(1990, 6, 30), birthday="06/30", gender=Gender.FEMALE, country_code="US", city="New York", address="Street 2", zip_code="54321", ) profile.set_values(new_values) event = ProfileUpdatedV1.from_profile(profile) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "updatedAt": IsDatetime( approx=profile.updated_at, delta=0, iso_string=True ), "updatedBy": None, "updatedFields": IsList( { "fieldName": "email", "oldValue": old_values["email"], "newValue": new_values["email"], }, { "fieldName": "firstName", "oldValue": old_values["first_name"], "newValue": new_values["first_name"], }, { "fieldName": "lastName", "oldValue": old_values["last_name"], "newValue": new_values["last_name"], }, { "fieldName": "phoneNumber", "oldValue": old_values["phone_number"], "newValue": new_values["phone_number"], }, { "fieldName": "dateOfBirth", "oldValue": IsDate( approx=old_values["date_of_birth"], iso_string=True ), "newValue": IsDate( approx=new_values["date_of_birth"], iso_string=True ), }, { "fieldName": "birthday", "oldValue": old_values["birthday"], "newValue": new_values["birthday"], }, { "fieldName": "gender", "oldValue": old_values["gender"], "newValue": new_values["gender"], }, { "fieldName": "countryCode", "oldValue": old_values["country_code"], "newValue": new_values["country_code"], }, { "fieldName": "city", "oldValue": old_values["city"], "newValue": new_values["city"], }, { "fieldName": "address", "oldValue": old_values["address"], "newValue": new_values["address"], }, { "fieldName": "zipCode", "oldValue": old_values["zip_code"], "newValue": new_values["zip_code"], }, check_order=False, ), } ) def test_from_profile_with_updated_by( self, build_model: BuildModel, faker: faker.Faker ) -> None: updated_by = faker.pystr() profile = build_model(Profile, updated_by=updated_by, first_name=None) profile.set_values(ProfileValues(first_name=faker.first_name())) event = ProfileUpdatedV1.from_profile(profile) assert event.updated_by == updated_by class TestProfileDeletedV1: def test_json_schema(self) -> None: assert ProfileDeletedV1.model_json_schema(by_alias=True) == { "title": "ProfileDeletedV1", "type": "object", "properties": { "id": { "title": "ID", "type": "string", "description": "Unique profile identifier in UUID format", }, "crmId": { "title": "CRM ID", "anyOf": [ {"type": "string"}, {"type": "null"}, ], }, "emailCampaignId": { "title": "Email campaign ID", "anyOf": [ {"type": "string"}, {"type": "null"}, ], "default": None, "deprecated": True, }, "originId": { "title": "Origin ID", "anyOf": [ {"type": "string"}, {"type": "null"}, ], }, "originType": { "title": "Origin type (EMAIL_CAMPAIGN|AUTOMATED_EMAIL)", "anyOf": [ {"type": "string"}, {"type": "null"}, ], }, "automatedEmailTriggerId": { "anyOf": [ {"type": "string"}, {"type": "null"}, ], "default": None, "title": "Automatedemailtriggerid", }, "deletedAt": { "title": "Deleted at", "type": "string", "format": "date-time", }, "updatedBy": { "anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Updated by", }, }, "required": [ "id", "crmId", "originId", "originType", "deletedAt", ], } def test_from_profile(self, build_model: BuildModel) -> None: profile = build_model(Profile, updated_by=None) event = ProfileDeletedV1.from_profile(profile, origin_id=None, origin_type=None) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "emailCampaignId": None, "originId": None, "originType": None, "automatedEmailTriggerId": None, "deletedAt": IsDatetime(iso_string=True), "updatedBy": None, } ) def test_from_profile_with_updated_by( self, build_model: BuildModel, faker: faker.Faker ) -> None: updated_by = faker.pystr() profile = build_model(Profile, updated_by=updated_by) event = ProfileDeletedV1.from_profile(profile, origin_id=None, origin_type=None) assert event.updated_by == updated_by def test_from_profile_with_email_campaign_id( self, build_model: BuildModel, faker: faker.Faker ) -> None: origin_id = faker.pystr() origin_type = "EMAIL_CAMPAIGN" profile = build_model(Profile, updated_by=None) event = ProfileDeletedV1.from_profile( profile, origin_id=origin_id, origin_type=origin_type, ) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "originId": origin_id, "originType": origin_type, "automatedEmailTriggerId": None, "deletedAt": IsDatetime(iso_string=True), "emailCampaignId": origin_id, "updatedBy": None, } ) def test_from_profile_with_automated_email( self, build_model: BuildModel, faker: faker.Faker ) -> None: origin_id = faker.pystr() origin_type = "AUTOMATED_EMAIL" profile = build_model(Profile, updated_by=None) event = ProfileDeletedV1.from_profile( profile, origin_id=origin_id, origin_type=origin_type, ) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "originId": origin_id, "originType": origin_type, "automatedEmailTriggerId": None, "deletedAt": IsDatetime(iso_string=True), "emailCampaignId": None, "updatedBy": None, } ) class TestProfileSubscriptionsUpdatedV1: def test_json_schema(self) -> None: assert ProfileSubscriptionsUpdatedV1.model_json_schema(by_alias=True) == { "title": "ProfileSubscriptionsUpdatedV1", "type": "object", "properties": { "id": { "description": "Unique profile identifier in UUID format", "title": "ID", "type": "string", }, "crmId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "CRM ID", }, "updatedAt": { "format": "date-time", "title": "Updated at", "type": "string", }, "updatedSubscriptions": { "items": {"$ref": "#/$defs/SubscriptionStatusUpdate"}, "title": "Updated subscriptions", "type": "array", }, "updatedBy": { "anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Updated by", }, }, "required": ["id", "crmId", "updatedSubscriptions", "updatedAt"], "$defs": { "SubscriptionStatusUpdate": { "properties": { "crmId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "CRM ID", }, "mailingListId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "Mailing list ID", }, "mailingListCrmId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "Mailing list CRM ID", }, "emailCampaignId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "Email campaign ID", "default": None, "deprecated": True, }, "originId": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "Origin ID", }, "originType": { "anyOf": [{"type": "string"}, {"type": "null"}], "title": "Origin type (EMAIL_CAMPAIGN|AUTOMATED_EMAIL)", }, "automatedEmailTriggerId": { "anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Automatedemailtriggerid", }, "id": { "description": ( "Unique subscription identifier in UUID format" ), "title": "ID", "type": "string", }, "newValue": {"title": "New value", "type": "boolean"}, "oldValue": {"title": "Old value", "type": "boolean"}, }, "required": [ "id", "crmId", "mailingListId", "mailingListCrmId", "originId", "originType", "oldValue", "newValue", ], "title": "SubscriptionStatusUpdate", "type": "object", } }, } def test_from_profile_subscriptions(self, build_model: BuildModel) -> None: profile = build_model(Profile, updated_by=None) subscription = build_model(Subscription, profile_id=profile.id) event = ProfileSubscriptionsUpdatedV1.from_profile_subscriptions( profile, subscriptions=[subscription], origin_id=None, origin_type=None, ) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "updatedSubscriptions": [ { "id": subscription.id, "crmId": None, "mailingListId": subscription.mailing_list_id, "mailingListCrmId": None, "originId": None, "originType": None, "emailCampaignId": None, "automatedEmailTriggerId": None, "oldValue": not subscription.is_active, "newValue": subscription.is_active, } ], "updatedAt": IsDatetime(iso_string=True), "updatedBy": None, } ) def test_from_profile_subscriptions_with_email_campaign_id( self, build_model: BuildModel, faker: faker.Faker ) -> None: email_campaign_id = faker.pystr() profile = build_model(Profile, updated_by=None) subscription = build_model(Subscription, profile_id=profile.id) event = ProfileSubscriptionsUpdatedV1.from_profile_subscriptions( profile, subscriptions=[subscription], origin_id=email_campaign_id, origin_type="EMAIL_CAMPAIGN", ) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "updatedSubscriptions": [ { "id": subscription.id, "crmId": None, "mailingListId": subscription.mailing_list_id, "mailingListCrmId": None, "originId": email_campaign_id, "originType": "EMAIL_CAMPAIGN", "emailCampaignId": email_campaign_id, "automatedEmailTriggerId": None, "oldValue": not subscription.is_active, "newValue": subscription.is_active, } ], "updatedAt": IsDatetime(iso_string=True), "updatedBy": None, } ) def test_from_profile_subscriptions_with_automated_email( self, build_model: BuildModel, faker: faker.Faker ) -> None: automated_email_id = faker.pystr() profile = build_model(Profile, updated_by=None) subscription = build_model(Subscription, profile_id=profile.id) event = ProfileSubscriptionsUpdatedV1.from_profile_subscriptions( profile, subscriptions=[subscription], origin_id=automated_email_id, origin_type="AUTOMATED_EMAIL", ) assert event.model_dump_json(by_alias=True) == IsJson( { "id": profile.id, "crmId": profile.crm_id, "updatedSubscriptions": [ { "id": subscription.id, "crmId": None, "mailingListId": subscription.mailing_list_id, "mailingListCrmId": None, "originId": automated_email_id, "originType": "AUTOMATED_EMAIL", "emailCampaignId": None, "automatedEmailTriggerId": None, "oldValue": not subscription.is_active, "newValue": subscription.is_active, } ], "updatedAt": IsDatetime(iso_string=True), "updatedBy": None, } ) def test_from_profile_subscriptions_with_updated_by( self, build_model: BuildModel, faker: faker.Faker ) -> None: updated_by = faker.pystr() profile = build_model(Profile, updated_by=updated_by) subscription = build_model(Subscription, profile_id=profile.id) event = ProfileSubscriptionsUpdatedV1.from_profile_subscriptions( profile, subscriptions=[subscription], origin_id=None, origin_type=None, ) assert event.updated_by == updated_by