Beyond Simple Requests: 7 Python Snippets to Upgrade Script‑Kiddie Tests

Environment preparation

Install core dependencies:

pip install requests pytest httpx sqlalchemy

1. Basic GET request with timeout and error handling

import requests
url = "https://httpbin.org/get"
try:
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    print(f"✅ 请求成功！状态码: {response.status_code}")
    print(f"📦 响应耗时: {response.elapsed.total_seconds()}秒")
    print(f"📄 返回内容: {response.json()}")
except requests.exceptions.RequestException as e:
    print(f"❌ 请求发生异常: {e}")

2. POST request with authentication headers

url = "https://httpbin.org/post"
payload = {"username": "admin", "password": "secure_pwd"}
headers = {
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
    "Content-Type": "application/json"
}
response = requests.post(url, json=payload, headers=headers, timeout=10)
if response.status_code == 200:
    res_json = response.json()
    if res_json.get("json", {}).get("username") == "admin":
        print("✅ 业务逻辑校验通过：用户登录信息正确")

3. Wrapper HTTP client using Session and logging

import logging, requests

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class APIClient:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = requests.Session()

    def request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}{endpoint}"
        logger.info(f"🚀 发起请求: {method} {url}")
        try:
            res = self.session.request(method, url, timeout=10, **kwargs)
            logger.info(f"📥 响应状态: {res.status_code}")
            res.raise_for_status()
            return res
        except Exception as e:
            logger.error(f"❌ 请求失败: {e}")
            raise

client = APIClient("https://httpbin.org")
resp = client.request("GET", "/get", params={"id": 1})

4. Data‑driven tests with pytest parametrize

import pytest

test_data = [
    {"user": "admin", "pwd": "123", "expected": 200},
    {"user": "", "pwd": "123", "expected": 400},
    {"user": "admin", "pwd": "wrong", "expected": 401}
]

class TestLogin:
    @pytest.mark.parametrize("case", test_data)
    def test_login_flow(self, case):
        print(f"正在测试用户: {case['user']}, 预期状态码: {case['expected']}")
        assert True  # placeholder

5. Resource management with a context manager

class UserAPITest:
    def __init__(self, base_url):
        self.client = requests.Session()
        self.base_url = base_url
        self.created_ids = []

    def create_user(self, name):
        print(f"创建用户: {name}")
        user_id = 1001
        self.created_ids.append(user_id)
        return user_id

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("🧹 开始执行 Teardown（环境清理）...")
        for uid in self.created_ids:
            print(f"删除脏数据用户 ID: {uid}")
        self.client.close()

with UserAPITest("https://api.example.com") as api:
    api.create_user("张三")

6. Database verification with SQLAlchemy ORM

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    status = Column(String)

def verify_db_order(order_id, expected_status):
    engine = create_engine('sqlite:///test.db')
    Session = sessionmaker(bind=engine)
    session = Session()
    order = session.query(Order).filter_by(id=order_id).first()
    assert order is not None, f"❌ 数据库中未找到订单 {order_id}"
    assert order.status == expected_status, f"❌ 状态不一致，期望 {expected_status}，实际 {order.status}"
    print(f"✅ 数据库校验通过：订单状态确认为 {order.status}")
    session.close()

7. Asynchronous concurrent testing with httpx

import httpx, asyncio

async def fetch_status(client, url):
    response = await client.get(url)
    return f"{url} -> {response.status_code}"

async def main():
    async with httpx.AsyncClient() as client:
        urls = ["https://httpbin.org/get"] * 5
        tasks = [fetch_status(client, url) for url in urls]
        print("⚡ 开始异步并发请求...")
        results = await asyncio.gather(*tasks)
        for res in results:
            print(res)

if __name__ == "__main__":
    asyncio.run(main())