mirror of
https://github.com/arsvendg/Stirling-PDF.git
synced 2026-07-16 19:33:11 +02:00
cucumber for days (#5766)
This commit is contained in:
Anthony Stirling
@@ -0,0 +1,499 @@
|
||||
"""
|
||||
Step definitions for JWT authentication end-to-end tests.
|
||||
|
||||
Covers:
|
||||
- Login / logout (POST /api/v1/auth/login, /logout)
|
||||
- Token refresh (POST /api/v1/auth/refresh)
|
||||
- Current user (GET /api/v1/auth/me)
|
||||
- Role-based access (admin-only endpoints)
|
||||
- API key authentication (X-API-KEY header)
|
||||
- MFA endpoints
|
||||
- User registration
|
||||
"""
|
||||
|
||||
import json as json_module
|
||||
|
||||
import requests
|
||||
from behave import given, then, when
|
||||
|
||||
BASE_URL = "http://localhost:8080"
|
||||
|
||||
# Default test credentials (set in docker-compose-security-with-login.yml)
|
||||
ADMIN_USERNAME = "admin"
|
||||
ADMIN_PASSWORD = "stirling"
|
||||
GLOBAL_API_KEY = "123456789"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper utilities
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _jwt_headers(context):
|
||||
"""Return Authorization: Bearer headers using the stored JWT token."""
|
||||
token = getattr(context, "jwt_token", None)
|
||||
assert token, "No JWT token stored in context – did you use 'Given I am logged in as admin'?"
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
def _api_key_headers(api_key):
|
||||
"""Return X-API-KEY headers for a given key."""
|
||||
return {"X-API-KEY": api_key}
|
||||
|
||||
|
||||
def _do_login(username, password):
|
||||
"""Perform a login POST and return the raw response."""
|
||||
payload = {"username": username, "password": password}
|
||||
return requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/login",
|
||||
json=payload,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GIVEN – session setup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("I am logged in as admin")
|
||||
def step_logged_in_as_admin(context):
|
||||
"""Login as the default admin user and store the JWT token in context."""
|
||||
response = _do_login(ADMIN_USERNAME, ADMIN_PASSWORD)
|
||||
assert response.status_code == 200, (
|
||||
f"Admin login failed (status {response.status_code}): {response.text}"
|
||||
)
|
||||
data = response.json()
|
||||
context.jwt_token = data["session"]["access_token"]
|
||||
|
||||
|
||||
@given("I store the JWT token")
|
||||
def step_store_current_jwt(context):
|
||||
"""Store the currently held jwt_token into context for later comparison."""
|
||||
assert hasattr(context, "jwt_token") and context.jwt_token, (
|
||||
"No JWT token available – did you log in first?"
|
||||
)
|
||||
context.original_jwt_token = context.jwt_token
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – login
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I login with username "{username}" and password "{password}"')
|
||||
def step_login(context, username, password):
|
||||
"""Send a login request with the given username and password."""
|
||||
payload = {"username": username, "password": password}
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/login",
|
||||
json=payload,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
@when('I login with only username "{username}"')
|
||||
def step_login_only_username(context, username):
|
||||
"""Send a login request with only a username field (no password key)."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/login",
|
||||
json={"username": username},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
@when('I login with only password "{password}"')
|
||||
def step_login_only_password(context, password):
|
||||
"""Send a login request with only a password field (no username key)."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/login",
|
||||
json={"password": password},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
@when('I login with an empty username and password "{password}"')
|
||||
def step_login_empty_username(context, password):
|
||||
"""Send a login request with an explicit empty string as the username."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/login",
|
||||
json={"username": "", "password": password},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
@when('I login with username "{username}" and an empty password')
|
||||
def step_login_empty_password(context, username):
|
||||
"""Send a login request with an explicit empty string as the password."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/login",
|
||||
json={"username": username, "password": ""},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – GET with various authentication methods
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with JWT authentication')
|
||||
def step_get_with_jwt(context, endpoint):
|
||||
"""Send GET request using the stored JWT token in the Authorization header."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with no authentication')
|
||||
def step_get_no_auth(context, endpoint):
|
||||
"""Send GET request with no authentication headers whatsoever."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with an invalid JWT token "{token_value}"')
|
||||
def step_get_with_invalid_jwt(context, endpoint, token_value):
|
||||
"""Send GET request with a specific invalid JWT string."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers={"Authorization": f"Bearer {token_value}"},
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with a malformed authorization header')
|
||||
def step_get_with_malformed_auth(context, endpoint):
|
||||
"""Send GET request with a non-Bearer authorization header."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers={"Authorization": "Basic dXNlcjpwYXNz"},
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with API key "{api_key}"')
|
||||
def step_get_with_api_key(context, endpoint, api_key):
|
||||
"""Send GET request using an X-API-KEY header."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_api_key_headers(api_key),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with Authorization header value "{header_value}"')
|
||||
def step_get_with_auth_header_value(context, endpoint, header_value):
|
||||
"""Send GET request with an arbitrary Authorization header value."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers={"Authorization": header_value},
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with an empty Authorization header')
|
||||
def step_get_with_empty_auth_header(context, endpoint):
|
||||
"""Send GET request with an explicitly empty Authorization header."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers={"Authorization": ""},
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with the stored JWT token')
|
||||
def step_get_with_stored_jwt(context, endpoint):
|
||||
"""Send GET request using the JWT token currently stored in context."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – POST with various authentication methods
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I send a POST request to "{endpoint}" with JWT authentication')
|
||||
def step_post_with_jwt(context, endpoint):
|
||||
"""Send POST request (no body) using the stored JWT token."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a POST request to "{endpoint}" with no authentication')
|
||||
def step_post_no_auth(context, endpoint):
|
||||
"""Send POST request with no authentication headers."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a POST request to "{endpoint}" with an invalid JWT token "{token_value}"')
|
||||
def step_post_with_invalid_jwt(context, endpoint, token_value):
|
||||
"""Send POST request with a specific invalid JWT string."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers={"Authorization": f"Bearer {token_value}"},
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
'I send a JSON POST request to "{endpoint}" with JWT authentication and body \'{json_body}\''
|
||||
)
|
||||
def step_json_post_with_jwt(context, endpoint, json_body):
|
||||
"""Send JSON POST request using the stored JWT token and a JSON body."""
|
||||
headers = {
|
||||
"Authorization": f"Bearer {context.jwt_token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=headers,
|
||||
data=json_body,
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
'I send a JSON POST request to "{endpoint}" with API key "{api_key}" and body \'{json_body}\''
|
||||
)
|
||||
def step_json_post_with_api_key(context, endpoint, api_key, json_body):
|
||||
"""Send JSON POST request using X-API-KEY header and a JSON body."""
|
||||
headers = {
|
||||
"X-API-KEY": api_key,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=headers,
|
||||
data=json_body,
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – token refresh
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I refresh the JWT token")
|
||||
def step_refresh_jwt(context):
|
||||
"""Send POST /api/v1/auth/refresh with the stored JWT token."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/refresh",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when("I refresh the stored JWT token")
|
||||
def step_refresh_stored_jwt(context):
|
||||
"""Send POST /api/v1/auth/refresh with the stored JWT token (alias)."""
|
||||
step_refresh_jwt(context)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – logout
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I logout with JWT authentication")
|
||||
def step_logout_with_jwt(context):
|
||||
"""Send POST /api/v1/auth/logout using the stored JWT token."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}/api/v1/auth/logout",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN – status code variants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('the response status code should be one of "{codes}"')
|
||||
def step_status_code_one_of(context, codes):
|
||||
"""Assert that the response status code matches any one of a comma-separated list."""
|
||||
allowed = [int(c.strip()) for c in codes.split(",")]
|
||||
actual = context.response.status_code
|
||||
assert actual in allowed, (
|
||||
f"Expected status code to be one of {allowed} but got {actual}. "
|
||||
f"Body: {context.response.text[:500]}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN – JWT structure assertions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("the response should contain a JWT access token")
|
||||
def step_response_contains_jwt(context):
|
||||
"""Assert the response has a session.access_token that looks like a JWT."""
|
||||
data = context.response.json()
|
||||
assert "session" in data, f"No 'session' key in response: {data}"
|
||||
assert "access_token" in data["session"], (
|
||||
f"No 'access_token' in session: {data['session']}"
|
||||
)
|
||||
token = data["session"]["access_token"]
|
||||
assert token, "access_token is empty"
|
||||
parts = token.split(".")
|
||||
assert len(parts) == 3, (
|
||||
f"JWT should have 3 dot-separated parts but got {len(parts)}: {token[:60]}..."
|
||||
)
|
||||
|
||||
|
||||
@then("the JWT access token should have three dot-separated parts")
|
||||
def step_jwt_three_parts(context):
|
||||
"""Assert the access_token in the response is a three-part JWT."""
|
||||
data = context.response.json()
|
||||
token = data.get("session", {}).get("access_token", "")
|
||||
assert token, "No access_token found in response"
|
||||
parts = token.split(".")
|
||||
assert len(parts) == 3, (
|
||||
f"JWT must have 3 parts (header.payload.signature) but got {len(parts)}: {token[:60]}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN – JSON field assertions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("the response JSON should have field \"{field}\"")
|
||||
def step_json_has_field(context, field):
|
||||
"""Assert the top-level response JSON contains the specified field."""
|
||||
data = context.response.json()
|
||||
assert field in data, (
|
||||
f"Expected field '{field}' in response JSON but only found: {list(data.keys())}"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON should have a user with username "{username}"')
|
||||
def step_json_user_username(context, username):
|
||||
"""Assert response.user.username equals the expected value."""
|
||||
data = context.response.json()
|
||||
assert "user" in data, f"No 'user' in response: {list(data.keys())}"
|
||||
actual = data["user"].get("username") or data["user"].get("email", "")
|
||||
assert actual == username, f"Expected username '{username}' but got '{actual}'"
|
||||
|
||||
|
||||
@then('the response JSON should have a user with role "{role}"')
|
||||
def step_json_user_role(context, role):
|
||||
"""Assert response.user.role equals the expected role string."""
|
||||
data = context.response.json()
|
||||
assert "user" in data, f"No 'user' in response: {list(data.keys())}"
|
||||
actual = data["user"].get("role", "")
|
||||
assert actual == role, f"Expected role '{role}' but got '{actual}'"
|
||||
|
||||
|
||||
@then('the response JSON user field "{field}" should not be empty')
|
||||
def step_json_user_field_not_empty(context, field):
|
||||
"""Assert response.user.<field> exists and is non-empty."""
|
||||
data = context.response.json()
|
||||
assert "user" in data, f"No 'user' in response: {list(data.keys())}"
|
||||
value = data["user"].get(field)
|
||||
assert value is not None and str(value) != "", (
|
||||
f"Expected user field '{field}' to be non-empty, got: {value!r}"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON user field "{field}" should equal "{expected}"')
|
||||
def step_json_user_field_equals(context, field, expected):
|
||||
"""Assert response.user.<field> equals the expected string value.
|
||||
|
||||
JSON booleans are compared as lowercase strings ("true"/"false").
|
||||
"""
|
||||
data = context.response.json()
|
||||
assert "user" in data, f"No 'user' in response: {list(data.keys())}"
|
||||
value = data["user"].get(field, "")
|
||||
actual = str(value).lower() if isinstance(value, bool) else str(value)
|
||||
assert actual == expected, (
|
||||
f"Expected user field '{field}' == '{expected}' but got '{actual}'"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON field "{field}" should equal "{expected}"')
|
||||
def step_json_top_field_equals(context, field, expected):
|
||||
"""Assert a top-level JSON field equals the expected string value.
|
||||
|
||||
JSON booleans (true/false) are compared as lowercase strings to match
|
||||
JSON serialisation ("true"/"false"), not Python's "True"/"False".
|
||||
"""
|
||||
data = context.response.json()
|
||||
value = data.get(field, "")
|
||||
actual = str(value).lower() if isinstance(value, bool) else str(value)
|
||||
assert actual == expected, (
|
||||
f"Expected JSON field '{field}' == '{expected}' but got '{actual}'. "
|
||||
f"Full response: {data}"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON session field "{field}" should be positive')
|
||||
def step_json_session_field_positive(context, field):
|
||||
"""Assert response.session.<field> is a number greater than zero."""
|
||||
data = context.response.json()
|
||||
assert "session" in data, f"No 'session' in response: {list(data.keys())}"
|
||||
value = data["session"].get(field)
|
||||
assert value is not None, f"Field '{field}' not found in session: {data['session']}"
|
||||
assert int(value) > 0, f"Expected session field '{field}' > 0 but got {value}"
|
||||
|
||||
|
||||
@then('the response JSON error should contain "{error_text}"')
|
||||
def step_json_error_contains(context, error_text):
|
||||
"""Assert the error/message/detail field contains the expected substring (case-insensitive)."""
|
||||
data = context.response.json()
|
||||
error = (
|
||||
data.get("error")
|
||||
or data.get("message")
|
||||
or data.get("detail")
|
||||
or ""
|
||||
)
|
||||
assert error_text.lower() in str(error).lower(), (
|
||||
f"Expected '{error_text}' (case-insensitive) in error response but got: '{error}'. "
|
||||
f"Full response: {data}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN – token storage and chaining
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("I store the JWT token from the login response")
|
||||
def step_store_jwt_from_login(context):
|
||||
"""Extract and store access_token from the login response."""
|
||||
data = context.response.json()
|
||||
assert "session" in data and "access_token" in data["session"], (
|
||||
f"No access_token in login response: {data}"
|
||||
)
|
||||
context.jwt_token = data["session"]["access_token"]
|
||||
assert context.jwt_token, "Stored JWT token is empty"
|
||||
|
||||
|
||||
@then("I update the stored JWT token from the response")
|
||||
def step_update_stored_jwt(context):
|
||||
"""Replace the stored JWT token with the new one from the current response."""
|
||||
data = context.response.json()
|
||||
assert "session" in data and "access_token" in data["session"], (
|
||||
f"No access_token in response: {data}"
|
||||
)
|
||||
new_token = data["session"]["access_token"]
|
||||
assert new_token, "New JWT token from response is empty"
|
||||
context.jwt_token = new_token
|
||||
@@ -0,0 +1,263 @@
|
||||
"""
|
||||
Step definitions for enterprise and proprietary API endpoints.
|
||||
|
||||
Covers:
|
||||
- User management (password change, API keys, admin CRUD)
|
||||
- Admin settings
|
||||
- Audit log
|
||||
- Invite links
|
||||
- Mobile scanner sessions
|
||||
- Signatures
|
||||
- Teams
|
||||
"""
|
||||
|
||||
import requests
|
||||
from behave import given, then, when
|
||||
|
||||
BASE_URL = "http://localhost:8080"
|
||||
|
||||
|
||||
def _jwt_headers(context):
|
||||
"""Return Authorization: Bearer headers using the stored JWT token."""
|
||||
token = getattr(context, "jwt_token", None)
|
||||
assert token, "No JWT token stored in context – did you use 'Given I am logged in as admin'?"
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
def _parse_params(params_str):
|
||||
"""Parse 'key1=val1&key2=val2' into a dict, supporting values that contain '='."""
|
||||
result = {}
|
||||
for part in params_str.split("&"):
|
||||
if "=" in part:
|
||||
k, v = part.split("=", 1)
|
||||
result[k.strip()] = v.strip()
|
||||
return result
|
||||
|
||||
|
||||
def _expand_stored(template, context):
|
||||
"""Replace '{stored}' in an endpoint template with context.stored_value."""
|
||||
return template.replace("{stored}", getattr(context, "stored_value", ""))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – POST with query params
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I send a POST request to "{endpoint}" with JWT authentication and params "{params}"')
|
||||
def step_post_with_jwt_and_params(context, endpoint, params):
|
||||
"""Send POST request with query parameters using the stored JWT token."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
params=_parse_params(params),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a POST request to "{endpoint}" with no authentication and params "{params}"')
|
||||
def step_post_no_auth_and_params(context, endpoint, params):
|
||||
"""Send POST request with query parameters and no authentication."""
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
params=_parse_params(params),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – GET with query params
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I send a GET request to "{endpoint}" with JWT authentication and params "{params}"')
|
||||
def step_get_with_jwt_and_params(context, endpoint, params):
|
||||
"""Send GET request with query parameters using the stored JWT token."""
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
params=_parse_params(params),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – DELETE
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I send a DELETE request to "{endpoint}" with JWT authentication')
|
||||
def step_delete_with_jwt(context, endpoint):
|
||||
"""Send DELETE request using the stored JWT token."""
|
||||
context.response = requests.delete(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a DELETE request to "{endpoint}" with no authentication')
|
||||
def step_delete_no_auth(context, endpoint):
|
||||
"""Send DELETE request with no authentication headers."""
|
||||
context.response = requests.delete(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a DELETE request to "{endpoint}" with JWT authentication and params "{params}"')
|
||||
def step_delete_with_jwt_and_params(context, endpoint, params):
|
||||
"""Send DELETE request with query parameters using the stored JWT token."""
|
||||
context.response = requests.delete(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
params=_parse_params(params),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when('I send a DELETE request to "{endpoint}" with no authentication and params "{params}"')
|
||||
def step_delete_no_auth_and_params(context, endpoint, params):
|
||||
"""Send DELETE request with query parameters and no authentication."""
|
||||
context.response = requests.delete(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
params=_parse_params(params),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN – steps that use a previously stored value in the URL path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when(
|
||||
'I use the stored value to send a GET request to "{endpoint_template}" with JWT authentication'
|
||||
)
|
||||
def step_get_stored_jwt(context, endpoint_template):
|
||||
"""Send GET request substituting {stored} in the path with context.stored_value."""
|
||||
endpoint = _expand_stored(endpoint_template, context)
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
'I use the stored value to send a GET request to "{endpoint_template}" with no authentication'
|
||||
)
|
||||
def step_get_stored_no_auth(context, endpoint_template):
|
||||
"""Send GET request substituting {stored} in the path with no authentication."""
|
||||
endpoint = _expand_stored(endpoint_template, context)
|
||||
context.response = requests.get(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
'I use the stored value to send a DELETE request to "{endpoint_template}" with JWT authentication'
|
||||
)
|
||||
def step_delete_stored_jwt(context, endpoint_template):
|
||||
"""Send DELETE request substituting {stored} in the path with context.stored_value."""
|
||||
endpoint = _expand_stored(endpoint_template, context)
|
||||
context.response = requests.delete(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
'I use the stored value to send a POST request to "{endpoint_template}" with JWT authentication'
|
||||
)
|
||||
def step_post_stored_jwt(context, endpoint_template):
|
||||
"""Send POST request substituting {stored} in the path with context.stored_value."""
|
||||
endpoint = _expand_stored(endpoint_template, context)
|
||||
context.response = requests.post(
|
||||
f"{BASE_URL}{endpoint}",
|
||||
headers=_jwt_headers(context),
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN – store response value for later steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('I store the response JSON field "{field}"')
|
||||
def step_store_response_field(context, field):
|
||||
"""Store a top-level JSON field from the current response into context.stored_value."""
|
||||
data = context.response.json()
|
||||
value = data.get(field)
|
||||
assert value is not None, f"Field '{field}' not found in response: {data}"
|
||||
context.stored_value = str(value)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN – response body / field assertions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('the response JSON field "{field}" should not be empty')
|
||||
def step_json_top_field_not_empty(context, field):
|
||||
"""Assert a top-level JSON field is present and non-empty."""
|
||||
data = context.response.json()
|
||||
value = data.get(field)
|
||||
assert value is not None and str(value) != "", (
|
||||
f"Expected field '{field}' to be non-empty, got: {value!r}. "
|
||||
f"Full response: {data}"
|
||||
)
|
||||
|
||||
|
||||
@then("the response body should not be empty")
|
||||
def step_response_body_not_empty(context):
|
||||
"""Assert that the raw response body contains at least one byte."""
|
||||
assert len(context.response.content) > 0, "Response body is empty"
|
||||
|
||||
|
||||
@then("the response JSON should be a list")
|
||||
def step_response_is_list(context):
|
||||
"""Assert that the top-level response JSON value is a list."""
|
||||
data = context.response.json()
|
||||
assert isinstance(data, list), (
|
||||
f"Expected response to be a JSON list but got: {type(data).__name__}. "
|
||||
f"Content: {str(data)[:200]}"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON field "{field}" should be a list')
|
||||
def step_json_field_is_list(context, field):
|
||||
"""Assert a top-level JSON field is a list."""
|
||||
data = context.response.json()
|
||||
value = data.get(field)
|
||||
assert isinstance(value, list), (
|
||||
f"Expected field '{field}' to be a list but got: {type(value).__name__}. "
|
||||
f"Full response: {data}"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON field "{field}" should be true')
|
||||
def step_json_field_is_true(context, field):
|
||||
"""Assert a top-level JSON boolean field is true."""
|
||||
data = context.response.json()
|
||||
value = data.get(field)
|
||||
actual = str(value).lower() if isinstance(value, bool) else str(value).lower()
|
||||
assert actual == "true", (
|
||||
f"Expected field '{field}' to be true but got: {value!r}. "
|
||||
f"Full response: {data}"
|
||||
)
|
||||
|
||||
|
||||
@then('the response JSON field "{field}" should be false')
|
||||
def step_json_field_is_false(context, field):
|
||||
"""Assert a top-level JSON boolean field is false."""
|
||||
data = context.response.json()
|
||||
value = data.get(field)
|
||||
actual = str(value).lower() if isinstance(value, bool) else str(value).lower()
|
||||
assert actual == "false", (
|
||||
f"Expected field '{field}' to be false but got: {value!r}. "
|
||||
f"Full response: {data}"
|
||||
)
|
||||
@@ -1,3 +1,4 @@
|
||||
import json as json_module
|
||||
import os
|
||||
import requests
|
||||
from behave import given, when, then
|
||||
@@ -289,6 +290,273 @@ def save_generated_pdf(context, filename):
|
||||
print(f"Saved generated PDF content to {filename}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Multi-file accumulation steps (same parameter key sent multiple times)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('I also generate a PDF file as "{param}"')
|
||||
def step_also_generate_pdf(context, param):
|
||||
"""Add an additional generated PDF under the given parameter name (supports duplicate keys)."""
|
||||
count = sum(1 for k, _ in getattr(context, "multi_files", []) if k == param)
|
||||
file_name = f"genericNonCustomisableName_extra_{param}_{count}.pdf"
|
||||
writer = PdfWriter()
|
||||
writer.add_blank_page(width=72, height=72)
|
||||
with open(file_name, "wb") as f:
|
||||
writer.write(f)
|
||||
if not hasattr(context, "multi_files"):
|
||||
context.multi_files = []
|
||||
context.multi_files.append((param, open(file_name, "rb")))
|
||||
|
||||
|
||||
@given('I also use an example file at "{filePath}" as parameter "{param}"')
|
||||
def step_also_use_example_file(context, filePath, param):
|
||||
"""Add an additional file from exampleFiles under the given parameter name."""
|
||||
if not hasattr(context, "multi_files"):
|
||||
context.multi_files = []
|
||||
try:
|
||||
context.multi_files.append((param, open(filePath, "rb")))
|
||||
except FileNotFoundError:
|
||||
raise FileNotFoundError(f"The example file '{filePath}' does not exist.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Non-PDF file generation steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('I generate a PNG image file as "{param}"')
|
||||
def step_generate_png(context, param):
|
||||
"""Generate a simple coloured PNG and register it under the given parameter name."""
|
||||
file_name = f"genericNonCustomisableName_{param}.png"
|
||||
img = Image.new("RGB", (200, 200), color=(73, 109, 137))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.rectangle([50, 50, 150, 150], fill=(255, 165, 0))
|
||||
draw.ellipse([75, 75, 125, 125], fill=(255, 255, 255))
|
||||
img.save(file_name, format="PNG")
|
||||
if not hasattr(context, "files"):
|
||||
context.files = {}
|
||||
context.files[param] = open(file_name, "rb")
|
||||
context.param_name = param
|
||||
context.file_name = file_name
|
||||
|
||||
|
||||
@given('I also generate a PNG image file as "{param}"')
|
||||
def step_also_generate_png(context, param):
|
||||
"""Add an additional PNG image under the given parameter name (supports duplicate keys)."""
|
||||
count = sum(1 for k, _ in getattr(context, "multi_files", []) if k == param)
|
||||
file_name = f"genericNonCustomisableName_extra_{param}_{count}.png"
|
||||
img = Image.new("RGB", (200, 200), color=(count * 60 + 40, 100, 180))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.rectangle([30, 30, 170, 170], fill=(200 - count * 30, 150, 50))
|
||||
img.save(file_name, format="PNG")
|
||||
if not hasattr(context, "multi_files"):
|
||||
context.multi_files = []
|
||||
context.multi_files.append((param, open(file_name, "rb")))
|
||||
|
||||
|
||||
@given('I generate an SVG file as "{param}"')
|
||||
def step_generate_svg(context, param):
|
||||
"""Generate a minimal SVG file and register it under the given parameter name."""
|
||||
file_name = f"genericNonCustomisableName_{param}.svg"
|
||||
svg_content = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
'<svg xmlns="http://www.w3.org/2000/svg" width="200" height="200" viewBox="0 0 200 200">\n'
|
||||
' <rect width="200" height="200" fill="#4a90e2"/>\n'
|
||||
' <circle cx="100" cy="100" r="60" fill="#f5a623"/>\n'
|
||||
' <text x="100" y="107" font-size="16" text-anchor="middle" fill="white">Test SVG</text>\n'
|
||||
"</svg>\n"
|
||||
)
|
||||
with open(file_name, "w", encoding="utf-8") as f:
|
||||
f.write(svg_content)
|
||||
if not hasattr(context, "files"):
|
||||
context.files = {}
|
||||
context.files[param] = open(file_name, "rb")
|
||||
context.param_name = param
|
||||
context.file_name = file_name
|
||||
|
||||
|
||||
@given('I also generate an SVG file as "{param}"')
|
||||
def step_also_generate_svg(context, param):
|
||||
"""Add an additional SVG under the given parameter name (supports duplicate keys)."""
|
||||
count = sum(1 for k, _ in getattr(context, "multi_files", []) if k == param)
|
||||
file_name = f"genericNonCustomisableName_extra_{param}_{count}.svg"
|
||||
svg_content = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
'<svg xmlns="http://www.w3.org/2000/svg" width="150" height="150" viewBox="0 0 150 150">\n'
|
||||
f' <rect width="150" height="150" fill="hsl({count * 60}, 70%, 50%)"/>\n'
|
||||
' <polygon points="75,20 140,130 10,130" fill="white" opacity="0.6"/>\n'
|
||||
"</svg>\n"
|
||||
)
|
||||
with open(file_name, "w", encoding="utf-8") as f:
|
||||
f.write(svg_content)
|
||||
if not hasattr(context, "multi_files"):
|
||||
context.multi_files = []
|
||||
context.multi_files.append((param, open(file_name, "rb")))
|
||||
|
||||
|
||||
@given('I generate an EML email file as "{param}"')
|
||||
def step_generate_eml(context, param):
|
||||
"""Generate a minimal RFC-2822 EML file and register it under the given parameter name."""
|
||||
file_name = f"genericNonCustomisableName_{param}.eml"
|
||||
eml_content = (
|
||||
"MIME-Version: 1.0\r\n"
|
||||
"Date: Thu, 19 Feb 2026 10:00:00 +0000\r\n"
|
||||
"Message-ID: <[email protected]>\r\n"
|
||||
"From: [email protected]\r\n"
|
||||
"To: [email protected]\r\n"
|
||||
"Subject: Test Email for PDF Conversion\r\n"
|
||||
"Content-Type: text/plain; charset=UTF-8\r\n"
|
||||
"\r\n"
|
||||
"This is a test email body.\r\n"
|
||||
"It contains multiple lines of text.\r\n"
|
||||
"Used for EML to PDF conversion testing.\r\n"
|
||||
)
|
||||
with open(file_name, "w", encoding="utf-8") as f:
|
||||
f.write(eml_content)
|
||||
if not hasattr(context, "files"):
|
||||
context.files = {}
|
||||
context.files[param] = open(file_name, "rb")
|
||||
context.param_name = param
|
||||
context.file_name = file_name
|
||||
|
||||
|
||||
@given('I generate a CBZ comic archive file as "{param}"')
|
||||
def step_generate_cbz(context, param):
|
||||
"""Generate a CBZ file (ZIP of PNG images) and register it under the given parameter name."""
|
||||
file_name = f"genericNonCustomisableName_{param}.cbz"
|
||||
with zipfile.ZipFile(file_name, "w") as cbz:
|
||||
for i in range(3):
|
||||
img = Image.new("RGB", (200, 300), color=(i * 60 + 40, 120, 200 - i * 50))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.rectangle([20, 20, 180, 280], outline=(0, 0, 0), width=3)
|
||||
draw.rectangle([40, 40, 160, 100], fill=(200, 200, 255))
|
||||
img_bytes = io.BytesIO()
|
||||
img.save(img_bytes, format="PNG")
|
||||
cbz.writestr(f"page_{i + 1:03d}.png", img_bytes.getvalue())
|
||||
if not hasattr(context, "files"):
|
||||
context.files = {}
|
||||
context.files[param] = open(file_name, "rb")
|
||||
context.param_name = param
|
||||
context.file_name = file_name
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PDF modification steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("the pdf has form fields")
|
||||
def step_pdf_has_form_fields(context):
|
||||
"""Create a PDF with a basic AcroForm text field on each page."""
|
||||
reader = PdfReader(context.file_name)
|
||||
page_count = len(reader.pages)
|
||||
buffer = io.BytesIO()
|
||||
c = canvas.Canvas(buffer, pagesize=letter)
|
||||
w, h = letter
|
||||
for i in range(page_count):
|
||||
c.acroForm.textfield(
|
||||
name=f"field_{i + 1}",
|
||||
tooltip=f"Field {i + 1}",
|
||||
x=72,
|
||||
y=h - 72,
|
||||
width=200,
|
||||
height=20,
|
||||
forceBorder=True,
|
||||
)
|
||||
c.showPage()
|
||||
c.save()
|
||||
with open(context.file_name, "wb") as f:
|
||||
f.write(buffer.getvalue())
|
||||
context.files[context.param_name].close()
|
||||
context.files[context.param_name] = open(context.file_name, "rb")
|
||||
|
||||
|
||||
@given('the pdf has an attachment named "{attachment_name}"')
|
||||
def step_pdf_has_attachment(context, attachment_name):
|
||||
"""Embed a small text attachment into the current PDF."""
|
||||
reader = PdfReader(context.file_name)
|
||||
writer = PdfWriter()
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
attachment_bytes = (
|
||||
f"Attachment: {attachment_name}\nThis is test attachment content.".encode("utf-8")
|
||||
)
|
||||
writer.add_attachment(attachment_name, attachment_bytes)
|
||||
with open(context.file_name, "wb") as f:
|
||||
writer.write(f)
|
||||
context.files[context.param_name].close()
|
||||
context.files[context.param_name] = open(context.file_name, "rb")
|
||||
|
||||
|
||||
@given("the pdf has bookmarks")
|
||||
def step_pdf_has_bookmarks(context):
|
||||
"""Add one top-level outline/bookmark entry per page to the current PDF."""
|
||||
reader = PdfReader(context.file_name)
|
||||
writer = PdfWriter()
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
for i in range(len(reader.pages)):
|
||||
writer.add_outline_item(f"Chapter {i + 1}", i)
|
||||
with open(context.file_name, "wb") as f:
|
||||
writer.write(f)
|
||||
context.files[context.param_name].close()
|
||||
context.files[context.param_name] = open(context.file_name, "rb")
|
||||
|
||||
|
||||
@given("the pdf has a Stirling-PDF QR code split marker on page {page_num:d}")
|
||||
def step_pdf_has_qr_split_marker(context, page_num):
|
||||
"""Replace page page_num (1-indexed) with a page containing a Stirling-PDF QR code."""
|
||||
try:
|
||||
import qrcode as _qrcode
|
||||
except ImportError:
|
||||
raise ImportError(
|
||||
"qrcode package is required for this step. "
|
||||
"Install with: pip install 'qrcode[pil]'"
|
||||
)
|
||||
reader = PdfReader(context.file_name)
|
||||
qr = _qrcode.QRCode(box_size=4, border=2)
|
||||
qr.add_data("https://github.com/Stirling-Tools/Stirling-PDF")
|
||||
qr.make(fit=True)
|
||||
qr_img = qr.make_image(fill_color="black", back_color="white")
|
||||
qr_bytes = io.BytesIO()
|
||||
qr_img.save(qr_bytes, format="PNG")
|
||||
qr_bytes.seek(0)
|
||||
writer = PdfWriter()
|
||||
for i, page in enumerate(reader.pages):
|
||||
if i + 1 == page_num:
|
||||
packet = io.BytesIO()
|
||||
can = canvas.Canvas(packet, pagesize=letter)
|
||||
w, h = letter
|
||||
can.drawImage(
|
||||
ImageReader(qr_bytes), (w - 100) / 2, (h - 100) / 2, width=100, height=100
|
||||
)
|
||||
can.showPage()
|
||||
can.save()
|
||||
packet.seek(0)
|
||||
qr_pdf = PdfReader(packet)
|
||||
writer.add_page(qr_pdf.pages[0])
|
||||
else:
|
||||
writer.add_page(page)
|
||||
with open(context.file_name, "wb") as f:
|
||||
writer.write(f)
|
||||
context.files[context.param_name].close()
|
||||
context.files[context.param_name] = open(context.file_name, "rb")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# JSON multipart part steps (@RequestPart endpoints like /form/fill)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('the request includes a JSON part "{part_name}" with content "{json_content}"')
|
||||
def step_request_json_part(context, part_name, json_content):
|
||||
"""Register a JSON multipart part (sent with Content-Type: application/json)."""
|
||||
if not hasattr(context, "json_parts"):
|
||||
context.json_parts = {}
|
||||
context.json_parts[part_name] = json_content
|
||||
|
||||
|
||||
########
|
||||
# WHEN #
|
||||
########
|
||||
@@ -337,6 +605,17 @@ def step_send_api_request(context, endpoint):
|
||||
print(f"form_data {file.name} with {mime_type}")
|
||||
form_data.append((key, (file.name, file, mime_type)))
|
||||
|
||||
# Multi-file entries (duplicate keys for MultipartFile[] endpoints, e.g. merge-pdfs)
|
||||
for key, file in getattr(context, "multi_files", []):
|
||||
mime_type, _ = mimetypes.guess_type(file.name)
|
||||
mime_type = mime_type or "application/octet-stream"
|
||||
print(f"form_data (multi) {file.name} with {mime_type}")
|
||||
form_data.append((key, (file.name, file, mime_type)))
|
||||
|
||||
# JSON multipart parts for @RequestPart endpoints (e.g. /form/fill)
|
||||
for part_name, json_content in getattr(context, "json_parts", {}).items():
|
||||
form_data.append((part_name, (None, json_content, "application/json")))
|
||||
|
||||
# Set timeout to 300 seconds (5 minutes) to prevent infinite hangs
|
||||
print(f"Sending POST request to {endpoint} with timeout=300s")
|
||||
response = requests.post(url, files=form_data, headers=API_HEADERS, timeout=300)
|
||||
|
||||
Reference in New Issue
Block a user