avaair-e2e-tests/conftest.py

# conftest.py
import os
import datetime as dt

import pytest
import requests

# =====================================================================================
# Global test configuration
# =====================================================================================

#: Base URL for the AvaAir public API under test.
#: Can be overridden via `API_BASE_URL` to point to staging/production.
BASE_URL = os.getenv("API_BASE_URL", "http://192.168.19.19:8200/api/v1")

#: Default mobile configuration for the test user.
MOBILE_COUNTRY_CODE = int(os.getenv("TEST_MOBILE_COUNTRY_CODE", "98"))
MOBILE_NUMBER = os.getenv("TEST_MOBILE_NUMBER", "9014332990")

#: Contact information used when creating flight bookings.
TEST_EMAIL = os.getenv("TEST_BOOK_EMAIL", "iamrezazoom@gmail.com")
TEST_PHONE_NUMBER = os.getenv("TEST_BOOK_PHONE", MOBILE_NUMBER)

# Passenger profile used across E2E booking / refund scenarios.
# These values should be overridden by environment variables in real environments
# (e.g. CI, staging, production-like smoke tests).
TRAVELER_FIRST_NAME = os.getenv("TEST_TRAVELER_FIRST_NAME", "REZA")
TRAVELER_LAST_NAME = os.getenv("TEST_TRAVELER_LAST_NAME", "ESMAEILI")
TRAVELER_GENDER = os.getenv("TEST_TRAVELER_GENDER", "male")
TRAVELER_NATIONAL_CODE = os.getenv("TEST_TRAVELER_NATIONAL_CODE", "0890500363")
TRAVELER_DATE_OF_BIRTH = os.getenv("TEST_TRAVELER_DATE_OF_BIRTH", "1998-05-20")
TRAVELER_PLACE_OF_BIRTH = os.getenv("TEST_TRAVELER_PLACE_OF_BIRTH", "IRN")


def _today_plus_days(days: int) -> str:
    """
    Return an ISO-formatted date (YYYY-MM-DD) relative to "today".

    This helper is intentionally minimal: it ensures that test dates are always
    slightly in the future (e.g. +7 days) so that we consistently hit
    "upcoming flights" in search results instead of past/expired inventory.
    """
    return (dt.date.today() + dt.timedelta(days=days)).isoformat()


@pytest.fixture
def api_wait():
    """
    Lightweight timing utility used to orchestrate API call pacing.

    Some flows (search → book → issue → refund) rely on asynchronous processing
    on the backend (e.g. ticket issuance, wallet updates, sync with suppliers).
    Introducing small, explicit delays between steps makes the E2E flow much
    more deterministic on real environments (staging/production).

    Usage:
        def test_something(api_wait):
            ... call API A ...
            api_wait(1.0)  # let the system converge
            ... call API B ...
    """
    import time

    def _do(seconds: float = 1.0) -> None:
        """
        Block the current test for the given number of seconds.

        The default delay (1.0s) is conservative and can be tuned per call site
        based on how heavy the underlying operation is (e.g. issue/refund may
        require a slightly longer buffer than a simple GET).
        """
        time.sleep(seconds)

    return _do


@pytest.fixture(scope="session")
def customer_auth():
    """
    Establish an authenticated customer session using the OTP (One-Time Password) flow.

    Behavior:
      • Always triggers an OTP request for the configured test mobile number.
      • If the `TEST_OTP_CODE` environment variable is set, it is used directly
        (ideal for CI and fully automated test runs).
      • If `TEST_OTP_CODE` is NOT set, the fixture falls back to interactive
        input and prompts the user to type the OTP received on their device.

    The fixture returns a small auth context dictionary containing:
      { "auth_token": str, "refresh_token": str, "user": dict }

    This structure is intentionally simple and reused by downstream fixtures
    (e.g. `auth_header`) and E2E tests.
    """

    # -------------------------------------------------------------------------
    # Step 1: Request OTP for the configured mobile number
    # -------------------------------------------------------------------------
    print(f"\n🔐 Initiating OTP login flow against {BASE_URL} ...")
    res = requests.post(
        f"{BASE_URL}/users/otp/request",
        json={
            "mobile_country_code": MOBILE_COUNTRY_CODE,
            "mobile_number": MOBILE_NUMBER,
        },
        headers={
            "accept": "application/json",
            "Content-Type": "application/json",
        },
        timeout=10,
    )
    res.raise_for_status()
    print("📨 OTP request accepted by API. Please check the device associated with the test number.\n")

    # -------------------------------------------------------------------------
    # Step 2: Resolve the OTP code
    #   - Prefer non-interactive mode via TEST_OTP_CODE for CI.
    #   - Fallback to interactive prompt in local/dev environments.
    # -------------------------------------------------------------------------
    otp = os.getenv("TEST_OTP_CODE")
    if otp:
        print("⚙️  Using OTP from TEST_OTP_CODE environment variable.")
    else:
        otp = input("➡️  Enter OTP received on the test device: ").strip()

    # -------------------------------------------------------------------------
    # Step 3: Verify OTP and obtain tokens + user payload
    # -------------------------------------------------------------------------
    verify_payload = {
        "description": "E2E FLIGHT FLOW",
        "mobile_country_code": MOBILE_COUNTRY_CODE,
        "mobile_number": MOBILE_NUMBER,
        "otp_code": otp,
    }

    res2 = requests.post(
        f"{BASE_URL}/users/otp/check",
        json=verify_payload,
        headers={
            "accept": "application/json",
            "Content-Type": "application/json",
        },
        timeout=10,
    )
    res2.raise_for_status()

    data = res2.json()

    # Defensive validation to fail fast with clear messaging if the auth
    # contract ever changes on the backend.
    assert "auth_token" in data and data["auth_token"], "OTP response missing auth_token"
    assert "refresh_token" in data and data["refresh_token"], "OTP response missing refresh_token"
    assert "user" in data and data["user"], "OTP response missing user object"

    print("✅ OTP verified successfully. Customer session is now authenticated.\n")

    return {
        "auth_token": data["auth_token"],
        "refresh_token": data["refresh_token"],
        "user": data["user"],
    }


@pytest.fixture(scope="session")
def auth_header(customer_auth):
    """
    Canonical Authorization header for all customer-facing API calls.

    This fixture is intentionally kept small and reusable. Any test that talks
    to the authenticated customer API surface should depend on this fixture
    instead of constructing its own headers.
    """
    return {
        "Authorization": f"Bearer {customer_auth['auth_token']}",
        "accept": "application/json",
    }


@pytest.fixture(scope="session")
def thr_to_mhd_search_params():
    """
    Shared search configuration for the canonical THR → MHD domestic flight scenario.

    This single-source-of-truth is consumed by the E2E tests to:
      • Always search from Tehran (THR) to Mashhad (MHD).
      • Always target a date 7 days in the future (relative to test execution).
      • Use a deterministic pagination and sorting strategy (lowest-price first).

    Adjustments to the default search behavior (e.g. multi-passenger, date
    offsets, or sorting strategies) should be funneled through this fixture
    to keep the test suite consistent and maintainable.
    """
    return {
        "origin_iata": "THR",
        "destination_iata": "MHD",
        "flight_date": _today_plus_days(7),
        "page": 1,
        "page_size": 20,
        "sort_option": "lowest-price",
    }