326 lines
11 KiB
Python
326 lines
11 KiB
Python
"""Tests for CustomerService functionality."""
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from alpine_bits_python.customer_service import CustomerService
|
|
from alpine_bits_python.db import Base, Customer
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_session():
|
|
"""Create an async session for testing."""
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with async_session_maker() as session:
|
|
yield session
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_customer_creates_hashed_version(async_session: AsyncSession):
|
|
"""Test that creating a customer automatically creates hashed version."""
|
|
service = CustomerService(async_session)
|
|
|
|
customer_data = {
|
|
"given_name": "John",
|
|
"surname": "Doe",
|
|
"contact_id": "test123",
|
|
"email_address": "john@example.com",
|
|
"phone": "+1234567890",
|
|
}
|
|
|
|
customer = await service.create_customer(customer_data)
|
|
|
|
assert customer.id is not None
|
|
assert customer.given_name == "John"
|
|
|
|
# Check that hashed version was created
|
|
hashed = await service.get_customer(customer.id)
|
|
assert hashed is not None
|
|
assert hashed.id == customer.id
|
|
assert hashed.hashed_email is not None
|
|
assert hashed.hashed_phone is not None
|
|
assert hashed.hashed_given_name is not None
|
|
assert hashed.hashed_surname is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_customer_updates_hashed_version(async_session: AsyncSession):
|
|
"""Test that updating a customer updates the hashed version."""
|
|
service = CustomerService(async_session)
|
|
|
|
# Create initial customer
|
|
customer_data = {
|
|
"given_name": "John",
|
|
"surname": "Doe",
|
|
"contact_id": "test123",
|
|
"email_address": "john@example.com",
|
|
}
|
|
customer = await service.create_customer(customer_data)
|
|
|
|
# Get initial hashed email
|
|
hashed = await service.get_customer(customer.id)
|
|
original_hashed_email = hashed.hashed_email
|
|
|
|
# Update customer email
|
|
update_data = {"email_address": "newemail@example.com"}
|
|
updated_customer = await service.update_customer(customer, update_data)
|
|
|
|
# Check that hashed version was updated
|
|
updated_hashed = await service.get_customer(updated_customer.id)
|
|
assert updated_hashed.hashed_email != original_hashed_email
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_customer_creates_new(async_session: AsyncSession):
|
|
"""Test get_or_create creates new customer when not found."""
|
|
service = CustomerService(async_session)
|
|
|
|
customer_data = {
|
|
"given_name": "Jane",
|
|
"surname": "Smith",
|
|
"contact_id": "new123",
|
|
"email_address": "jane@example.com",
|
|
}
|
|
|
|
customer = await service.get_or_create_customer(customer_data)
|
|
assert customer.id is not None
|
|
assert customer.contact_id == "new123"
|
|
|
|
# Verify hashed version exists
|
|
hashed = await service.get_customer(customer.id)
|
|
assert hashed is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_customer_updates_existing(async_session: AsyncSession):
|
|
"""Test get_or_create updates existing customer when found."""
|
|
service = CustomerService(async_session)
|
|
|
|
# Create initial customer
|
|
customer_data = {
|
|
"given_name": "Jane",
|
|
"surname": "Smith",
|
|
"contact_id": "existing123",
|
|
"email_address": "jane@example.com",
|
|
}
|
|
original_customer = await service.create_customer(customer_data)
|
|
|
|
# Try to create again with same contact_id but different data
|
|
updated_data = {
|
|
"given_name": "Janet",
|
|
"surname": "Smith",
|
|
"contact_id": "existing123",
|
|
"email_address": "janet@example.com",
|
|
}
|
|
customer = await service.get_or_create_customer(updated_data)
|
|
|
|
# Should be same customer ID but updated data
|
|
assert customer.id == original_customer.id
|
|
assert customer.given_name == "Janet"
|
|
assert customer.email_address == "janet@example.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hash_existing_customers_backfills(async_session: AsyncSession):
|
|
"""Test that hash_existing_customers backfills missing hashed data."""
|
|
# Create a customer directly in DB without using service
|
|
customer = Customer(
|
|
given_name="Bob",
|
|
surname="Builder",
|
|
contact_id="bob123",
|
|
email_address="bob@example.com",
|
|
phone="+9876543210",
|
|
)
|
|
async_session.add(customer)
|
|
await async_session.commit()
|
|
await async_session.refresh(customer)
|
|
|
|
# Verify no hashed version exists
|
|
result = await async_session.execute(
|
|
select(Customer).where(Customer.id == customer.id)
|
|
)
|
|
hashed = result.scalar_one_or_none()
|
|
assert hashed, "Customer should exist."
|
|
|
|
assert hashed.hashed_given_name is None, "Hashed given name should be None."
|
|
assert hashed.hashed_email is None, "Hashed email should be None."
|
|
|
|
# Run backfill
|
|
service = CustomerService(async_session)
|
|
count = await service.hash_existing_customers()
|
|
|
|
assert count == 1
|
|
|
|
# Verify hashed version now exists
|
|
result = await async_session.execute(
|
|
select(Customer).where(Customer.id == customer.id)
|
|
)
|
|
hashed = result.scalar_one_or_none()
|
|
assert hashed is not None, "Customer should still exist after backfill."
|
|
assert hashed.hashed_email is not None, "Hashed email should be populated."
|
|
assert hashed.hashed_given_name is not None, "Hashed given name should be populated."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hash_existing_customers_skips_already_hashed(
|
|
async_session: AsyncSession,
|
|
):
|
|
"""Test that hash_existing_customers skips customers already hashed."""
|
|
service = CustomerService(async_session)
|
|
|
|
# Create customer with service (creates hashed version)
|
|
customer_data = {
|
|
"given_name": "Alice",
|
|
"surname": "Wonder",
|
|
"contact_id": "alice123",
|
|
"email_address": "alice@example.com",
|
|
}
|
|
await service.create_customer(customer_data)
|
|
|
|
# Run backfill - should return 0 since customer is already hashed
|
|
count = await service.hash_existing_customers()
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hashing_normalization(async_session: AsyncSession):
|
|
"""Test that hashing properly normalizes data."""
|
|
service = CustomerService(async_session)
|
|
|
|
# Create customer with mixed case and spaces
|
|
customer_data = {
|
|
"given_name": " John ",
|
|
"surname": "DOE",
|
|
"contact_id": "test123",
|
|
"email_address": " John.Doe@Example.COM ",
|
|
"phone": "+1 (234) 567-8900",
|
|
}
|
|
customer = await service.create_customer(customer_data)
|
|
|
|
hashed = await service.get_customer(customer.id)
|
|
|
|
# Verify hashes exist (normalization should have occurred)
|
|
assert hashed.hashed_email is not None
|
|
assert hashed.hashed_phone is not None
|
|
|
|
# Hash should be consistent for same normalized value
|
|
from alpine_bits_python.db import Customer as CustomerModel
|
|
|
|
normalized_email_hash = CustomerModel._normalize_and_hash(
|
|
"john.doe@example.com"
|
|
)
|
|
assert hashed.hashed_email == normalized_email_hash
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hash_existing_customers_normalizes_country_code(
|
|
async_session: AsyncSession,
|
|
):
|
|
"""Test that hash_existing_customers normalizes invalid country codes.
|
|
|
|
Specifically tests the case where a customer was saved with a full country
|
|
name (e.g., "Italy") instead of the ISO 3166-1 alpha-2 code (e.g., "IT").
|
|
The validation should convert "Italy" to "IT", which is then hashed.
|
|
"""
|
|
# Create a customer directly in DB with invalid country code "Italy"
|
|
# This simulates a customer that was saved before validation was implemented
|
|
customer = Customer(
|
|
given_name="Marco",
|
|
surname="Rossi",
|
|
contact_id="marco123",
|
|
email_address="marco@example.com",
|
|
phone="+39123456789",
|
|
country_code="Italy", # Invalid - should be "IT"
|
|
city_name="Rome",
|
|
postal_code="00100",
|
|
)
|
|
async_session.add(customer)
|
|
await async_session.commit()
|
|
await async_session.refresh(customer)
|
|
|
|
# Verify no hashed version exists yet
|
|
result = await async_session.execute(
|
|
select(Customer).where(Customer.id == customer.id)
|
|
)
|
|
hashed = result.scalar_one_or_none()
|
|
assert hashed is not None, "Customer should exist."
|
|
|
|
assert hashed.hashed_given_name is None, "Hashed given name should be None."
|
|
assert hashed.hashed_email is None, "Hashed email should be None."
|
|
assert hashed.hashed_country_code is None, "Hashed country code should be None."
|
|
|
|
# Verify the customer has the invalid country code stored in the DB
|
|
assert hashed.country_code == "Italy"
|
|
|
|
# Run hash_existing_customers - this should normalize "Italy" to "IT"
|
|
# during validation and successfully create a hashed customer
|
|
service = CustomerService(async_session)
|
|
count = await service.hash_existing_customers()
|
|
|
|
# Should successfully hash this customer (country code normalized during validation)
|
|
assert count == 1
|
|
|
|
# Verify hashed version was created
|
|
await async_session.refresh(customer)
|
|
result = await async_session.execute(
|
|
select(Customer).where(Customer.id == customer.id)
|
|
)
|
|
hashed = result.scalar_one_or_none()
|
|
assert hashed is not None
|
|
|
|
# The hashed customer should have the hashed version of normalized country code "IT"
|
|
# "IT" -> lowercase "it" -> sha256 hash
|
|
expected_hash = "2ad8a7049d7c5511ac254f5f51fe70a046ebd884729056f0fe57f5160d467153"
|
|
assert hashed.hashed_country_code == expected_hash
|
|
# Note: The original customer's country_code in DB remains "Italy" unchanged
|
|
|
|
# Now let's test the case where we have a valid 2-letter code
|
|
# but in the wrong case (should be normalized to uppercase)
|
|
customer2 = Customer(
|
|
given_name="Maria",
|
|
surname="Bianchi",
|
|
contact_id="maria123",
|
|
email_address="maria@example.com",
|
|
phone="+39987654321",
|
|
country_code="it", # Valid but lowercase - should be normalized to "IT"
|
|
city_name="Milan",
|
|
postal_code="20100",
|
|
)
|
|
async_session.add(customer2)
|
|
await async_session.commit()
|
|
await async_session.refresh(customer2)
|
|
|
|
# Run hash_existing_customers again
|
|
count = await service.hash_existing_customers()
|
|
|
|
# Should hash only the second customer (first one already hashed)
|
|
assert count == 1
|
|
|
|
# Verify the customer's country_code was normalized to uppercase
|
|
await async_session.refresh(customer2)
|
|
assert customer2.country_code == "IT"
|
|
|
|
# Verify hashed version was created with correct hash
|
|
result = await async_session.execute(
|
|
select(Customer).where(Customer.id == customer2.id)
|
|
)
|
|
hashed = result.scalar_one_or_none()
|
|
assert hashed is not None
|
|
|
|
# Verify the hashed country code matches the expected hash
|
|
# "IT" -> lowercase "it" -> sha256 hash
|
|
expected_hash = "2ad8a7049d7c5511ac254f5f51fe70a046ebd884729056f0fe57f5160d467153"
|
|
assert hashed.hashed_country_code == expected_hash
|
|
|
|
# For comparison, verify that "italy" would produce a different hash
|
|
italy_hash = "93ff074ca77b5f7e61e63320b615081149b818863599189b0e356ec3889d51f7"
|
|
assert hashed.hashed_country_code != italy_hash
|