219 lines
7.6 KiB
Python
219 lines
7.6 KiB
Python
"""Tests for webhook-related Pydantic schemas."""
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from alpine_bits_python.const import WebhookStatus
|
|
from alpine_bits_python.schemas import (
|
|
HotelData,
|
|
WebhookEndpointData,
|
|
WebhookRequestData,
|
|
)
|
|
|
|
|
|
class TestHotelData:
|
|
"""Tests for HotelData schema."""
|
|
|
|
def test_valid_hotel_data(self):
|
|
"""Test creating a valid HotelData instance."""
|
|
data = HotelData(
|
|
hotel_id="hotel123",
|
|
hotel_name="Test Hotel",
|
|
username="admin",
|
|
password_hash="hashed_password_123",
|
|
)
|
|
assert data.hotel_id == "hotel123"
|
|
assert data.hotel_name == "Test Hotel"
|
|
assert data.username == "admin"
|
|
assert data.password_hash == "hashed_password_123"
|
|
assert data.is_active is True
|
|
assert isinstance(data.created_at, datetime)
|
|
|
|
def test_whitespace_stripping(self):
|
|
"""Test that whitespace is stripped from string fields."""
|
|
data = HotelData(
|
|
hotel_id=" hotel123 ",
|
|
hotel_name=" Test Hotel ",
|
|
username=" admin ",
|
|
password_hash="hashed_password_123",
|
|
)
|
|
assert data.hotel_id == "hotel123"
|
|
assert data.hotel_name == "Test Hotel"
|
|
assert data.username == "admin"
|
|
|
|
def test_optional_fields(self):
|
|
"""Test that optional fields can be None."""
|
|
data = HotelData(
|
|
hotel_id="hotel123",
|
|
hotel_name="Test Hotel",
|
|
username="admin",
|
|
password_hash="hashed_password_123",
|
|
meta_account_id=None,
|
|
google_account_id=None,
|
|
)
|
|
assert data.meta_account_id is None
|
|
assert data.google_account_id is None
|
|
|
|
|
|
class TestWebhookEndpointData:
|
|
"""Tests for WebhookEndpointData schema."""
|
|
|
|
def test_valid_webhook_endpoint(self):
|
|
"""Test creating a valid WebhookEndpointData instance."""
|
|
data = WebhookEndpointData(
|
|
hotel_id="hotel123",
|
|
webhook_secret="secret_abc123",
|
|
webhook_type="wix_form",
|
|
)
|
|
assert data.hotel_id == "hotel123"
|
|
assert data.webhook_secret == "secret_abc123"
|
|
assert data.webhook_type == "wix_form"
|
|
assert data.is_enabled is True
|
|
assert isinstance(data.created_at, datetime)
|
|
|
|
def test_webhook_endpoint_with_description(self):
|
|
"""Test WebhookEndpointData with optional description."""
|
|
data = WebhookEndpointData(
|
|
hotel_id="hotel123",
|
|
webhook_secret="secret_abc123",
|
|
webhook_type="generic",
|
|
description="Main booking form",
|
|
)
|
|
assert data.description == "Main booking form"
|
|
|
|
def test_whitespace_stripping(self):
|
|
"""Test that whitespace is stripped from string fields."""
|
|
data = WebhookEndpointData(
|
|
hotel_id=" hotel123 ",
|
|
webhook_secret=" secret_abc123 ",
|
|
webhook_type=" wix_form ",
|
|
)
|
|
assert data.hotel_id == "hotel123"
|
|
assert data.webhook_secret == "secret_abc123"
|
|
assert data.webhook_type == "wix_form"
|
|
|
|
|
|
class TestWebhookRequestData:
|
|
"""Tests for WebhookRequestData schema."""
|
|
|
|
def test_auto_calculate_payload_hash(self):
|
|
"""Test that payload_hash is auto-calculated from payload_json."""
|
|
payload = {"name": "John", "email": "john@example.com"}
|
|
data = WebhookRequestData(payload_json=payload)
|
|
|
|
# Verify hash was calculated
|
|
assert data.payload_hash is not None
|
|
assert len(data.payload_hash) == 64 # SHA256 produces 64 hex chars
|
|
|
|
# Verify it matches the expected hash (same algorithm as api.py)
|
|
payload_json_str = json.dumps(payload, sort_keys=True)
|
|
expected_hash = hashlib.sha256(payload_json_str.encode("utf-8")).hexdigest()
|
|
assert data.payload_hash == expected_hash
|
|
|
|
def test_explicit_payload_hash(self):
|
|
"""Test providing payload_hash explicitly (for purged payloads)."""
|
|
explicit_hash = "a" * 64
|
|
data = WebhookRequestData(
|
|
payload_json=None,
|
|
payload_hash=explicit_hash,
|
|
)
|
|
assert data.payload_hash == explicit_hash
|
|
assert data.payload_json is None
|
|
|
|
def test_payload_hash_required(self):
|
|
"""Test that payload_hash is required (either calculated or explicit)."""
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
WebhookRequestData(
|
|
payload_json=None,
|
|
payload_hash=None,
|
|
)
|
|
assert "payload_hash is required" in str(exc_info.value)
|
|
|
|
def test_consistent_hashing(self):
|
|
"""Test that the same payload always produces the same hash."""
|
|
payload = {"b": 2, "a": 1, "c": 3} # Unordered keys
|
|
|
|
data1 = WebhookRequestData(payload_json=payload.copy())
|
|
data2 = WebhookRequestData(payload_json=payload.copy())
|
|
|
|
assert data1.payload_hash == data2.payload_hash
|
|
|
|
def test_default_status(self):
|
|
"""Test that status defaults to PENDING."""
|
|
data = WebhookRequestData(payload_json={"test": "data"})
|
|
assert data.status == WebhookStatus.PENDING
|
|
|
|
def test_status_normalization(self):
|
|
"""Test that status is normalized to WebhookStatus enum."""
|
|
data = WebhookRequestData(
|
|
payload_json={"test": "data"},
|
|
status="completed", # String
|
|
)
|
|
assert data.status == WebhookStatus.COMPLETED
|
|
assert isinstance(data.status, WebhookStatus)
|
|
|
|
def test_retry_count_default(self):
|
|
"""Test that retry_count defaults to 0."""
|
|
data = WebhookRequestData(payload_json={"test": "data"})
|
|
assert data.retry_count == 0
|
|
|
|
def test_optional_foreign_keys(self):
|
|
"""Test optional foreign key fields."""
|
|
data = WebhookRequestData(
|
|
payload_json={"test": "data"},
|
|
webhook_endpoint_id=123,
|
|
hotel_id="hotel456",
|
|
)
|
|
assert data.webhook_endpoint_id == 123
|
|
assert data.hotel_id == "hotel456"
|
|
|
|
def test_result_tracking(self):
|
|
"""Test result tracking fields."""
|
|
data = WebhookRequestData(
|
|
payload_json={"test": "data"},
|
|
created_customer_id=1,
|
|
created_reservation_id=2,
|
|
)
|
|
assert data.created_customer_id == 1
|
|
assert data.created_reservation_id == 2
|
|
|
|
def test_purged_payload(self):
|
|
"""Test representing a purged webhook request (after processing)."""
|
|
explicit_hash = "b" * 64
|
|
data = WebhookRequestData(
|
|
payload_json=None,
|
|
payload_hash=explicit_hash,
|
|
status=WebhookStatus.COMPLETED,
|
|
purged_at=datetime.now(),
|
|
)
|
|
assert data.payload_json is None
|
|
assert data.payload_hash == explicit_hash
|
|
assert data.status == WebhookStatus.COMPLETED
|
|
assert data.purged_at is not None
|
|
|
|
def test_processing_metadata(self):
|
|
"""Test processing tracking fields."""
|
|
now = datetime.now()
|
|
data = WebhookRequestData(
|
|
payload_json={"test": "data"},
|
|
status=WebhookStatus.PROCESSING,
|
|
processing_started_at=now,
|
|
)
|
|
assert data.status == WebhookStatus.PROCESSING
|
|
assert data.processing_started_at == now
|
|
assert data.processing_completed_at is None
|
|
|
|
def test_request_metadata(self):
|
|
"""Test request metadata fields."""
|
|
data = WebhookRequestData(
|
|
payload_json={"test": "data"},
|
|
source_ip="192.168.1.1",
|
|
user_agent="Mozilla/5.0",
|
|
)
|
|
assert data.source_ip == "192.168.1.1"
|
|
assert data.user_agent == "Mozilla/5.0"
|