🚀 Building a Robust API Test Automation Framework with Python

class APIPlayer:
    """Your API testing commander-in-chief 🎮"""
    
    def __init__(self, base_url: str, results: Results):
        self.results = results
        self.logger = get_logger("APIPlayer")
        
        # Create our endpoint specialists
        self.cars_api = CarsAPIEndpoints(base_url)
        self.auth_api = AuthAPIEndpoints(base_url)
        self.users_api = UsersAPIEndpoints(base_url)
        
        # Init session state
        self.session_token = None

class Results:
    """Your friendly neighborhood test tracker 📊"""
    
    def __init__(self, log_file_path: Optional[str] = None):
        self.total = 0
        self.passed = 0
        self.failed = 0
        self.logger = get_logger("Results", log_file_path)
        self.start_time = time.time()

class CarsAPIEndpoints(BaseAPI):
    """Your gateway to the world of cars 🚗"""
    def __init__(self, base_url: str):
        super().__init__(base_url)
        self.logger = get_logger("CarsAPI")

    def cars_url(self, suffix: str = '') -> str:
        """Build cars endpoint URL"""
        return f"{self.base_url}/api/v1/cars{suffix}"
        
    def get_cars(self, headers: dict = None) -> Response:
        """Get all cars from the kingdom"""
        self.logger.info("Fetching all cars from the royal garage...")
        return self.get(self.cars_url(), headers=headers)

class BaseAPI:
    """The foundation of our API kingdom 🏰"""
    
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.logger = get_logger("BaseAPI")
        self.session = requests.Session()
        
        # Setup connection pooling for better performance
        adapter = HTTPAdapter(max_retries=3, 
                             pool_connections=100, 
                             pool_maxsize=100)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
    
    def __del__(self):
        """Close session when object is destroyed"""
        if hasattr(self, 'session'):
            self.session.close()

# config/environments.py
environments = {
    "dev": {
        "base_url": "https://dev-api.example.com",
        "username": "testuser",
        "password": "devpassword123"
    },
    "staging": {
        "base_url": "https://stg-api.example.com",
        "username": "stguser",
        "password": "stgpassword456"
    },
    "prod": {
        "base_url": "https://api.example.com",
        "username": "produser",
        "password": "prod!password789"
    }
}

pytest --env=staging

self.logger.info("🚀 Starting our epic car creation journey...")
self.logger.info("📡 Sending our request to the Cars Kingdom")
self.logger.info(f"🎉 Victory! Got response with status: {response.status_code}")
self.logger.error("🚨 Oops! Something went wrong with our request")

def success(self, message: str) -> None:
    """Celebrate another victory! 🎉"""
    self.logger.info(f"🟢 VICTORY: {message}")
    self.total += 1     # Another battle fought
    self.passed += 1    # Another battle won
    
def failure(self, message: str, error=None) -> None:
    """Record the challenges we faced 📝"""
    self.total += 1
    self.failed += 1
    
    if error:
        self.logger.error(f"🔴 DEFEAT: {message} - {str(error)}")
    else:
        self.logger.error(f"🔴 DEFEAT: {message}")

@allure.epic("Cars API")
@allure.feature("Car Management")
@allure.story("Creating Cars")
@allure.title("Add new car and verify it exists in the list")
@allure.description("Test to verify that the car count increases correctly after adding a new car. This ensures the system's counting mechanism works properly.")
@allure.tag("API", "Cars", "Validation")
@allure.severity(allure.severity_level.NORMAL)
def test_verify_car_count(api_player_fixture, auth_details, initial_car_count):
    # Our epic test continues here...

pytest --alluredir=./allure-results
allure serve allure-results  # Open the magical portal to view reports

@pytest.fixture
def api_player_fixture():
    """Create our main character for this adventure"""
    results = Results(log_file_path="./logs/test_log.txt")
    player = APIPlayer(conf.base_url, results)
    return player

def test_add_car(api_player_fixture, auth_details):
    """Add a car and verify it was added successfully"""
    # GIVEN our API Player is ready for action
    player = api_player_fixture
    
    # AND we have a new car design blueprint
    car_details = conf.car_details
    
    # WHEN we add the car to the system
    response = player.add_car(car_details, auth_details)
    
    # THEN the API should confirm success
    assert response.status_code == 201
    assert response.json().get("success") == True
    
    # AND when we get all cars
    all_cars = player.get_cars(auth_details)
    
    # THEN our new car should be in the list
    cars_list = all_cars.json().get("cars", [])
    new_car = next((car for car in cars_list if car["model"] == car_details["model"]), None)
    assert new_car is not None
    
    # Celebrate our victory! 🎉
    print("🎉 Woohoo! Tesla has joined our awesome car collection!")

def _execute_api_call(self, operation_name: str, api_call: Callable) -> Response:
    """Execute an API call with full error handling and logging"""
    start_time = time.time()
    
    try:
        self.logger.info(f"🚀 Starting {operation_name}")
        response = api_call()
        
        # Log performance metrics
        execution_time = round((time.time() - start_time) * 1000, 2)
        self.logger.info(f"⏱️ {operation_name} completed in {execution_time}ms")
        
        # Validate response
        if response.status_code < 400:
            self.results.success(f"{operation_name} returned {response.status_code}")
        else:
            error_msg = f"{operation_name} failed with status {response.status_code}"
            self.results.failure(error_msg)
            self.logger.error(f"📝 Response body: {response.text}")
            
        return response
        
    except Exception as e:
        # Handle unexpected errors
        execution_time = round((time.time() - start_time) * 1000, 2)
        error_msg = f"{operation_name} failed with exception: {str(e)}"
        self.results.failure(error_msg, e)
        self.logger.error(f"⏱️ Failed after {execution_time}ms")
        raise APIPlayerException(error_msg) from e

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, min=2, max=10),
       retry=retry_if_exception_type(ConnectionError))
def get(self, url: str, params: dict = None, headers: dict = None) -> Response:
    """Performs a GET request with retry logic for transient errors"""
    self.logger.info(f"🌐 GET: {url}")
    return self.session.get(url, params=params, headers=headers, timeout=30)

return self._execute_api_call(
    "add_car", 
    lambda: self.cars_api.add_car(data=car_details, headers=headers)
)