Fixed some tests and added schemas
This commit is contained in:
Jonas Linter
218
tests/test_schemas_webhook.py
Normal file
218
tests/test_schemas_webhook.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user