|
| 1 | +import base64 |
| 2 | +import json |
| 3 | +import logging |
| 4 | +import os |
| 5 | +from collections.abc import Callable |
| 6 | + |
| 7 | +from mailjet_rest import Client |
| 8 | + |
| 9 | +# Configure logging for the smoke test |
| 10 | +logging.getLogger("urllib3").setLevel(logging.WARNING) |
| 11 | +logging.getLogger("mailjet_rest.client").setLevel(logging.DEBUG) |
| 12 | +logging.basicConfig(format="%(levelname)s - %(message)s") |
| 13 | + |
| 14 | +# Fetch credentials from environment variables |
| 15 | +API_KEY = os.environ.get("MJ_APIKEY_PUBLIC", "") |
| 16 | +API_SECRET = os.environ.get("MJ_APIKEY_PRIVATE", "") |
| 17 | +BEARER_TOKEN = os.environ.get("MJ_CONTENT_TOKEN", "") |
| 18 | + |
| 19 | +# Initialize clients for different API versions |
| 20 | +mailjet_v3 = Client(auth=(API_KEY, API_SECRET)) |
| 21 | +mailjet_v3_1 = Client(auth=(API_KEY, API_SECRET), version="v3.1") |
| 22 | +mailjet_v1 = Client(auth=BEARER_TOKEN or (API_KEY, API_SECRET), version="v1") |
| 23 | + |
| 24 | + |
| 25 | +def run_test(test_name: str, func: Callable, expected_status: tuple[int, ...] = (200,)) -> None: |
| 26 | + """Wrapper that checks if the status code matches the expected one.""" |
| 27 | + print(f"\n{'=' * 60}\n🚀 RUNNING: {test_name}\n{'=' * 60}") |
| 28 | + try: |
| 29 | + result = func() |
| 30 | + if getattr(result, "status_code", None) in expected_status: |
| 31 | + print(f"✅ SUCCESS (Status Code: {result.status_code})") |
| 32 | + else: |
| 33 | + print(f"❌ FAILED (Expected {expected_status}, got {getattr(result, 'status_code', None)})") |
| 34 | + |
| 35 | + try: |
| 36 | + print(json.dumps(result.json(), indent=2)) |
| 37 | + except ValueError: |
| 38 | + print(f"Response Text: '{getattr(result, 'text', '')}'") |
| 39 | + except Exception as e: |
| 40 | + print(f"❌ Failed Exception: {type(e).__name__}: {e}") |
| 41 | + |
| 42 | + |
| 43 | +def test_send_sandbox(): |
| 44 | + """Test 1: Send API v3.1 (Sandbox)""" |
| 45 | + data = { |
| 46 | + "Messages": [ |
| 47 | + { |
| 48 | + "From": {"Email": "pilot@mailjet.com", "Name": "Pilot"}, |
| 49 | + "To": [{"Email": "passenger@mailjet.com"}], |
| 50 | + "Subject": "Smoke Test", |
| 51 | + "TextPart": "This is a live routing test.", |
| 52 | + } |
| 53 | + ], |
| 54 | + "SandboxMode": True, |
| 55 | + } |
| 56 | + return mailjet_v3_1.send.create(data=data) |
| 57 | + |
| 58 | + |
| 59 | +def test_get_contacts(): |
| 60 | + """Test 2: Email API v3 (Contacts)""" |
| 61 | + return mailjet_v3.contact.get(filters={"limit": 2}) |
| 62 | + |
| 63 | + |
| 64 | +def test_get_statistics(): |
| 65 | + """Test 3: Email API v3 (Statistics)""" |
| 66 | + filters = { |
| 67 | + "CounterSource": "APIKey", |
| 68 | + "CounterTiming": "Message", |
| 69 | + "CounterResolution": "Lifetime", |
| 70 | + } |
| 71 | + return mailjet_v3.statcounters.get(filters=filters) |
| 72 | + |
| 73 | + |
| 74 | +def test_parse_api(): |
| 75 | + """Test 4: Email API v3 (Parse API)""" |
| 76 | + return mailjet_v3.parseroute.get(filters={"limit": 2}) |
| 77 | + |
| 78 | + |
| 79 | +def test_segmentation(): |
| 80 | + """Test 5: Email API v3 (Segmentation)""" |
| 81 | + return mailjet_v3.contactfilter.get(filters={"limit": 2}) |
| 82 | + |
| 83 | + |
| 84 | +def test_content_api_templates(): |
| 85 | + """Test 6: Content API v1 (Templates)""" |
| 86 | + return mailjet_v1.templates.get(filters={"limit": 2}) |
| 87 | + |
| 88 | + |
| 89 | +def test_content_api_images_negative(): |
| 90 | + """Test 7: Negative test (verifies server validation for missing multipart).""" |
| 91 | + client_logger = logging.getLogger("mailjet_rest.client") |
| 92 | + previous_level = client_logger.level |
| 93 | + # Temporarily hide the "ERROR - API Error 400" log since we expect a failure |
| 94 | + client_logger.setLevel(logging.CRITICAL) |
| 95 | + try: |
| 96 | + data = {"name": "test.png", "image_data": "iVBORw0KGgo="} |
| 97 | + return mailjet_v1.data_images.create(data=data) |
| 98 | + finally: |
| 99 | + client_logger.setLevel(previous_level) |
| 100 | + |
| 101 | + |
| 102 | +def test_content_api_images_real_upload(): |
| 103 | + """Test 8: REAL file upload via multipart/form-data with mandatory metadata.""" |
| 104 | + # 1x1 Transparent PNG in Base64 |
| 105 | + b64_string = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=" |
| 106 | + image_bytes = base64.b64decode(b64_string) |
| 107 | + |
| 108 | + # Status must be "open" or "locked" according to the documentation |
| 109 | + metadata_json = '{"name": "smoke_test_logo.png", "Status": "open"}' |
| 110 | + |
| 111 | + files_payload = { |
| 112 | + "metadata": (None, metadata_json, "application/json"), |
| 113 | + "file": ("smoke_test_logo.png", image_bytes, "image/png"), |
| 114 | + } |
| 115 | + |
| 116 | + # Erase default JSON Content-Type to allow requests to build multipart boundaries |
| 117 | + return mailjet_v1.data_images.create(headers={"Content-Type": None}, files=files_payload) |
| 118 | + |
| 119 | + |
| 120 | +if __name__ == "__main__": |
| 121 | + if not API_KEY or not API_SECRET: |
| 122 | + print("⚠️ MJ_APIKEY_PUBLIC and/or MJ_APIKEY_PRIVATE not found.") |
| 123 | + |
| 124 | + run_test("1. Send API v3.1 (Sandbox)", test_send_sandbox) |
| 125 | + run_test("2. Email API v3 (Contacts)", test_get_contacts) |
| 126 | + run_test("3. Email API v3 (Statistics)", test_get_statistics) |
| 127 | + run_test("4. Email API v3 (Parse API)", test_parse_api) |
| 128 | + run_test("5. Email API v3 (Segmentation)", test_segmentation) |
| 129 | + run_test("6. Content API v1 (Templates)", test_content_api_templates) |
| 130 | + |
| 131 | + # We only explicitly pass expected_status when it deviates from the (200,) default |
| 132 | + run_test("7. Content API v1 (Negative Upload)", test_content_api_images_negative, expected_status=(400,)) |
| 133 | + run_test("8. Content API v1 (Real Multipart Upload)", test_content_api_images_real_upload, expected_status=(201,)) |
0 commit comments