{#- Python HTTP test function template for pytest/urllib
Context variables:
- fn_name: sanitized test function name
- description: fixture description with period appended
- method: HTTP method (uppercase)
- path: fixture path
- headers_py: Python dict literal for headers
- has_body: bool
- body_py: Python literal for JSON body
- expected_status: HTTP status code
- has_text_body: bool
- text_py: Python literal for expected text body
- has_json_body: bool
- json_py: Python literal for expected JSON body
- has_partial_body: bool
- partial_body_checks: array of {key, py_val} dicts
- header_assertions: array of {name, assertion_type, value} dicts
- has_validation_errors: bool
- validation_errors: array of {loc_py, escaped_msg} dicts
#}
{%- set body_used = has_text_body or has_json_body or has_partial_body or has_validation_errors -%}
{%- set headers_used = header_assertions | length > 0 -%}
def test_{{ fn_name }}() -> None:
"""{{ description }}"""
base = os.environ.get("SUT_URL", "http://127.0.0.1:8000")
url = f"{base}{{ path }}"
_headers = {{ headers_py }}
{% if has_body %}
{% if is_multipart and body_is_bytes_literal %}
_body = {{ body_py }}
_headers.setdefault("Content-Type", "multipart/form-data; boundary=alef-boundary")
{% elif is_multipart and body_is_string %}
_body = ({{ body_py }}).encode()
_headers.setdefault("Content-Type", "multipart/form-data; boundary=alef-boundary")
{% elif is_form_body and body_is_string %}
_body = ({{ body_py }}).encode()
{% else %}
_headers.setdefault("Content-Type", "application/json")
_body = json.dumps({{ body_py }}).encode()
{% endif %}
_req = urllib.request.Request(url, data=_body, headers=_headers, method="{{ method }}") # noqa: S310
{% else %}
_req = urllib.request.Request(url, headers=_headers, method="{{ method }}") # noqa: S310
{% endif %}
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs): # noqa: ANN002, ANN003, ANN202, ARG002
return None
_opener = urllib.request.build_opener(_NoRedirect())
try:
response = _opener.open(_req)
status_code = response.status
resp_body = response.read()
resp_headers = dict(response.headers)
except urllib.error.HTTPError as _exc:
status_code = _exc.code
{% if body_used %}
resp_body = _exc.read()
{% else %}
resp_body = _exc.read() # noqa: F841
{% endif %}
{% if headers_used %}
resp_headers = dict(_exc.headers)
{% else %}
resp_headers = dict(_exc.headers) # noqa: F841
{% endif %}
assert status_code == {{ expected_status }}
{% if has_text_body %}
assert resp_body.decode() == {{ text_py }}
{% elif has_json_body %}
data = json.loads(resp_body)
assert data == {{ json_py }}
{% elif has_partial_body %}
data = json.loads(resp_body)
{% for check in partial_body_checks %}
assert data["{{ check.key }}"] == {{ check.py_val }}
{% endfor %}
{% endif %}
{% for assertion in header_assertions %}
{% if assertion.assertion_type == "present" %}
assert "{{ assertion.name }}" in resp_headers
{% elif assertion.assertion_type == "absent" %}
assert resp_headers.get("{{ assertion.name }}") is None
{% elif assertion.assertion_type == "uuid" %}
assert re.match(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', resp_headers["{{ assertion.name }}"])
{% else %}
assert resp_headers["{{ assertion.name }}"] == "{{ assertion.value }}"
{% endif %}
{% endfor %}
{% if has_validation_errors %}
_data = json.loads(resp_body)
errors = _data.get("errors", [])
{% for ve in validation_errors %}
assert any(e["loc"] == [{{ ve.loc_py }}] and "{{ ve.escaped_msg }}" in e["msg"] for e in errors)
{% endfor %}
{% endif %}