diff --git a/src/alpine_bits_python/alpinebits_server.py b/src/alpine_bits_python/alpinebits_server.py index 80c07e3..5fe494a 100644 --- a/src/alpine_bits_python/alpinebits_server.py +++ b/src/alpine_bits_python/alpinebits_server.py @@ -11,7 +11,7 @@ import re from abc import ABC from dataclasses import dataclass from datetime import datetime -from enum import Enum, IntEnum +from enum import Enum from typing import Any, Optional, override from xsdata.formats.dataclass.serializers.config import SerializerConfig @@ -23,6 +23,7 @@ from alpine_bits_python.alpine_bits_helpers import ( ) from alpine_bits_python.logging_config import get_logger +from .const import HttpStatusCode from .db import Customer, Reservation from .generated.alpinebits import ( OtaNotifReportRq, @@ -38,15 +39,6 @@ from .reservation_service import ReservationService _LOGGER = get_logger(__name__) -class HttpStatusCode(IntEnum): - """Allowed HTTP status codes for AlpineBits responses.""" - - OK = 200 - BAD_REQUEST = 400 - UNAUTHORIZED = 401 - INTERNAL_SERVER_ERROR = 500 - - def dump_json_for_xml(json_content: Any) -> str: """Dump JSON content as a pretty-printed string for embedding in XML. diff --git a/src/alpine_bits_python/api.py b/src/alpine_bits_python/api.py index 16b82b3..1548358 100644 --- a/src/alpine_bits_python/api.py +++ b/src/alpine_bits_python/api.py @@ -36,6 +36,7 @@ from .alpinebits_server import ( ) from .auth import generate_unique_id, validate_api_key from .config_loader import load_config +from .const import HttpStatusCode from .customer_service import CustomerService from .db import Base, get_database_url from .db import Customer as DBCustomer @@ -147,7 +148,7 @@ async def push_listener(customer: DBCustomer, reservation: DBReservation, hotel) version=Version.V2024_10, ) - if request.status_code != 200: + if request.status_code != HttpStatusCode.OK: _LOGGER.error( "Failed to generate push request for hotel %s, reservation %s: %s", hotel_id, diff --git a/src/alpine_bits_python/const.py b/src/alpine_bits_python/const.py index 7e4d2ce..cf939d6 100644 --- a/src/alpine_bits_python/const.py +++ b/src/alpine_bits_python/const.py @@ -1,5 +1,16 @@ +from enum import IntEnum from typing import Final + +class HttpStatusCode(IntEnum): + """Allowed HTTP status codes for AlpineBits responses.""" + + OK = 200 + BAD_REQUEST = 400 + UNAUTHORIZED = 401 + INTERNAL_SERVER_ERROR = 500 + + RESERVATION_ID_TYPE: str = ( "13" # Default reservation ID type for Reservation. 14 would be cancellation ) diff --git a/tests/test_alpine_bits_server_read.py b/tests/test_alpine_bits_server_read.py index 4f193ed..6b36243 100644 --- a/tests/test_alpine_bits_server_read.py +++ b/tests/test_alpine_bits_server_read.py @@ -16,14 +16,12 @@ from xsdata_pydantic.bindings import XmlParser, XmlSerializer from alpine_bits_python.alpine_bits_helpers import create_res_retrieve_response from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo, AlpineBitsServer +from alpine_bits_python.const import HttpStatusCode from alpine_bits_python.db import AckedRequest, Base, Customer, Reservation from alpine_bits_python.generated import OtaReadRq from alpine_bits_python.generated.alpinebits import OtaResRetrieveRs from alpine_bits_python.schemas import ReservationData -# HTTP status code constants -HTTP_OK = 200 - @pytest_asyncio.fixture async def test_db_engine(): @@ -558,7 +556,7 @@ class TestAcknowledgments: ) assert response is not None - assert response.status_code == HTTP_OK + assert response.status_code == HttpStatusCode.OK assert response.xml_content is not None # Verify response contains reservation data @@ -609,7 +607,7 @@ class TestAcknowledgments: ) assert ack_response is not None - assert ack_response.status_code == HTTP_OK + assert ack_response.status_code == HttpStatusCode.OK assert "OTA_NotifReportRS" in ack_response.xml_content @pytest.mark.asyncio @@ -920,7 +918,7 @@ class TestAcknowledgments: ) assert response is not None - assert response.status_code == HTTP_OK + assert response.status_code == HttpStatusCode.OK # Parse response to verify both reservations are returned parser = XmlParser() diff --git a/tests/test_alpinebits_server_ping.py b/tests/test_alpinebits_server_ping.py index 3ad8009..6d339aa 100644 --- a/tests/test_alpinebits_server_ping.py +++ b/tests/test_alpinebits_server_ping.py @@ -4,6 +4,7 @@ import pytest from xsdata_pydantic.bindings import XmlParser from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo, AlpineBitsServer +from alpine_bits_python.const import HttpStatusCode from alpine_bits_python.generated.alpinebits import OtaPingRs @@ -60,7 +61,7 @@ async def test_ping_action_response_success(): client_info=client_info, version="2024-10", ) - assert response.status_code == 200 + assert response.status_code == HttpStatusCode.OK assert "", ) - assert response.status_code == 401 + assert response.status_code == HttpStatusCode.UNAUTHORIZED def test_xml_upload_invalid_path(self, client, basic_auth_headers): """Test XML upload with path traversal attempt. @@ -805,7 +803,7 @@ class TestAuthentication: ) # Should not be 401 - assert response.status_code != 401 + assert response.status_code != HttpStatusCode.UNAUTHORIZED def test_basic_auth_missing_credentials(self, client): """Test basic auth with missing credentials.""" @@ -814,7 +812,7 @@ class TestAuthentication: data={"action": "OTA_Ping:Handshaking"}, ) - assert response.status_code == 401 + assert response.status_code == HttpStatusCode.UNAUTHORIZED def test_basic_auth_malformed_header(self, client): """Test basic auth with malformed Authorization header.""" @@ -839,7 +837,7 @@ class TestEventDispatcher: # The async task runs in background and doesn't affect response response = client.post("/api/webhook/wix-form", json=sample_wix_form_data) - assert response.status_code == 200 + assert response.status_code == HttpStatusCode.OK # Event dispatcher is tested separately in its own test suite @@ -902,7 +900,7 @@ class TestCORS: # TestClient returns 400 for OPTIONS requests # In production, CORS middleware handles preflight correctly - assert response.status_code in [200, 400, 405] + assert response.status_code in [HttpStatusCode.OK, 400, 405] class TestRateLimiting: @@ -917,7 +915,7 @@ class TestRateLimiting: responses.append(response.status_code) # All should succeed if under limit - assert all(status == 200 for status in responses) + assert all(status == HttpStatusCode.OK for status in responses) if __name__ == "__main__":