diff --git a/tests/conftest.py b/tests/conftest.py index e281d1f7fa..f41f932a0a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -702,6 +702,18 @@ def request_parameter_with_value(context, name, value, path_parameters): path_parameters[param_name] = json.loads(tpl) +@given(parsers.parse('the request and response use "{compression}" compression')) +def request_and_response_use_compression(compression): + """Defer request and response compression to the generated replay server.""" + assert compression + + +@given(parsers.parse('the client selects "{compression}" compression')) +def client_selects_compression(compression): + """Pass the selected compression from the generated request plan.""" + assert compression + + def assert_no_unparsed(data): if isinstance(data, list): for item in data: @@ -732,6 +744,9 @@ def prepare_test_runner_request(context, client, api_version, request, path_para body = _materialize_test_value(request_plan["body"]["value"], context) api_request["kwargs"]["body"] = json.dumps(body) + if request_plan.get("selected_compression") is not None: + api_request["kwargs"]["content_encoding"] = json.dumps(request_plan["selected_compression"]) + for parameter in request_plan["parameters"]: source = parameter["source"] if source["type"] == "fixture": diff --git a/tests/generated-test/test-runner-data/manifest.json b/tests/generated-test/test-runner-data/manifest.json index c770c3463b..c8a3cdd73c 100644 --- a/tests/generated-test/test-runner-data/manifest.json +++ b/tests/generated-test/test-runner-data/manifest.json @@ -1211,6 +1211,20 @@ "scenario": "Search test logs returns \"OK\" response", "version": "v1" }, + { + "feature": "Logs", + "feature_file": "../../v1/features/logs.feature", + "file": "v1/logs/send-deflate-logs-returns-response-from-server-always-200-empty-json-response.json", + "scenario": "Send deflate logs returns \"Response from server (always 200 empty JSON).\" response", + "version": "v1" + }, + { + "feature": "Logs", + "feature_file": "../../v1/features/logs.feature", + "file": "v1/logs/send-gzip-logs-returns-response-from-server-always-200-empty-json-response.json", + "scenario": "Send gzip logs returns \"Response from server (always 200 empty JSON).\" response", + "version": "v1" + }, { "feature": "Logs", "feature_file": "../../v1/features/logs.feature", @@ -6391,6 +6405,20 @@ "scenario": "Search logs returns \"OK\" response with pagination", "version": "v2" }, + { + "feature": "Logs", + "feature_file": "../../v2/features/logs.feature", + "file": "v2/logs/send-deflate-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json", + "scenario": "Send deflate logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "version": "v2" + }, + { + "feature": "Logs", + "feature_file": "../../v2/features/logs.feature", + "file": "v2/logs/send-gzip-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json", + "scenario": "Send gzip logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "version": "v2" + }, { "feature": "Logs", "feature_file": "../../v2/features/logs.feature", diff --git a/tests/generated-test/test-runner-data/v1/logs/send-deflate-logs-returns-response-from-server-always-200-empty-json-response.json b/tests/generated-test/test-runner-data/v1/logs/send-deflate-logs-returns-response-from-server-always-200-empty-json-response.json new file mode 100644 index 0000000000..2d2ffc1076 --- /dev/null +++ b/tests/generated-test/test-runner-data/v1/logs/send-deflate-logs-returns-response-from-server-always-200-empty-json-response.json @@ -0,0 +1,38 @@ +{ + "api": "Logs", + "expected_status": 200, + "feature": "Logs", + "id": "v1/Logs/Send deflate logs returns \"Response from server (always 200 empty JSON).\" response", + "operation_id": "SubmitLog", + "request": { + "body": { + "schema": { + "format": null, + "items": { + "format": null, + "ref": "HTTPLogItem", + "type": "object" + }, + "ref": "HTTPLog", + "type": "array" + }, + "source": "inline", + "value": [ + { + "ddtags": "host:{{ unique_alnum }}", + "message": "{{ unique }}" + } + ] + }, + "compression": "deflate", + "content_type": "application/json", + "method": "POST", + "pagination": false, + "parameters": [], + "path": "/v1/input", + "selected_compression": "deflate" + }, + "scenario": "Send deflate logs returns \"Response from server (always 200 empty JSON).\" response", + "schema_version": 1, + "version": "v1" +} diff --git a/tests/generated-test/test-runner-data/v1/logs/send-gzip-logs-returns-response-from-server-always-200-empty-json-response.json b/tests/generated-test/test-runner-data/v1/logs/send-gzip-logs-returns-response-from-server-always-200-empty-json-response.json new file mode 100644 index 0000000000..9488adfae2 --- /dev/null +++ b/tests/generated-test/test-runner-data/v1/logs/send-gzip-logs-returns-response-from-server-always-200-empty-json-response.json @@ -0,0 +1,38 @@ +{ + "api": "Logs", + "expected_status": 200, + "feature": "Logs", + "id": "v1/Logs/Send gzip logs returns \"Response from server (always 200 empty JSON).\" response", + "operation_id": "SubmitLog", + "request": { + "body": { + "schema": { + "format": null, + "items": { + "format": null, + "ref": "HTTPLogItem", + "type": "object" + }, + "ref": "HTTPLog", + "type": "array" + }, + "source": "inline", + "value": [ + { + "ddtags": "host:{{ unique_alnum }}", + "message": "{{ unique }}" + } + ] + }, + "compression": "gzip", + "content_type": "application/json", + "method": "POST", + "pagination": false, + "parameters": [], + "path": "/v1/input", + "selected_compression": "gzip" + }, + "scenario": "Send gzip logs returns \"Response from server (always 200 empty JSON).\" response", + "schema_version": 1, + "version": "v1" +} diff --git a/tests/generated-test/test-runner-data/v2/logs/send-deflate-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json b/tests/generated-test/test-runner-data/v2/logs/send-deflate-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json new file mode 100644 index 0000000000..dec29d1204 --- /dev/null +++ b/tests/generated-test/test-runner-data/v2/logs/send-deflate-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json @@ -0,0 +1,41 @@ +{ + "api": "Logs", + "expected_status": 202, + "feature": "Logs", + "id": "v2/Logs/Send deflate logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "operation_id": "SubmitLog", + "request": { + "body": { + "schema": { + "format": null, + "items": { + "format": null, + "ref": "HTTPLogItem", + "type": "object" + }, + "ref": "HTTPLog", + "type": "array" + }, + "source": "inline", + "value": [ + { + "ddsource": "nginx", + "ddtags": "env:staging,version:5.1", + "hostname": "i-012345678", + "message": "2019-11-19T14:37:58,995 INFO [process.name][20081] Hello World", + "service": "payment" + } + ] + }, + "compression": "deflate", + "content_type": "application/json", + "method": "POST", + "pagination": false, + "parameters": [], + "path": "/api/v2/logs", + "selected_compression": "deflate" + }, + "scenario": "Send deflate logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "schema_version": 1, + "version": "v2" +} diff --git a/tests/generated-test/test-runner-data/v2/logs/send-gzip-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json b/tests/generated-test/test-runner-data/v2/logs/send-gzip-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json new file mode 100644 index 0000000000..06a238d744 --- /dev/null +++ b/tests/generated-test/test-runner-data/v2/logs/send-gzip-logs-returns-request-accepted-for-processing-always-202-empty-json-response.json @@ -0,0 +1,41 @@ +{ + "api": "Logs", + "expected_status": 202, + "feature": "Logs", + "id": "v2/Logs/Send gzip logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "operation_id": "SubmitLog", + "request": { + "body": { + "schema": { + "format": null, + "items": { + "format": null, + "ref": "HTTPLogItem", + "type": "object" + }, + "ref": "HTTPLog", + "type": "array" + }, + "source": "inline", + "value": [ + { + "ddsource": "nginx", + "ddtags": "env:staging,version:5.1", + "hostname": "i-012345678", + "message": "2019-11-19T14:37:58,995 INFO [process.name][20081] Hello World", + "service": "payment" + } + ] + }, + "compression": "gzip", + "content_type": "application/json", + "method": "POST", + "pagination": false, + "parameters": [], + "path": "/api/v2/logs", + "selected_compression": "gzip" + }, + "scenario": "Send gzip logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "schema_version": 1, + "version": "v2" +} diff --git a/tests/generated-test/test-server b/tests/generated-test/test-server index 417fcc086c..ff395f697d 100755 --- a/tests/generated-test/test-server +++ b/tests/generated-test/test-server @@ -10,12 +10,15 @@ from __future__ import annotations import argparse import base64 +import ctypes import json import os import re import tempfile import threading import uuid +import zlib +from ctypes.util import find_library from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -44,6 +47,18 @@ SAFE_BROWSER_HEADERS = { "content-security-policy": "default-src 'none'; sandbox", "x-content-type-options": "nosniff", } +ZSTD_CONTENTSIZE_UNKNOWN = (1 << 64) - 1 +ZSTD_CONTENTSIZE_ERROR = (1 << 64) - 2 +ZSTD_MAGIC = b"\x28\xb5\x2f\xfd" +ZSTD_SINGLE_SEGMENT_FLAG = 0x20 +ZSTD_DESCRIPTOR_LOW_BITS_MASK = 0x3F +ZSTD_MIN_RAW_FRAME_SIZE = 6 +ZSTD_FCS_TWO_BYTE_OFFSET = 0x100 +UINT8_MAX = (1 << 8) - 1 +UINT16_WITH_OFFSET_MAX = (1 << 16) - 1 + ZSTD_FCS_TWO_BYTE_OFFSET +UINT32_MAX = (1 << 32) - 1 +ZSTD_RAW_BLOCK_SIZE = 128 * 1024 +MAX_DECOMPRESSED_BODY_SIZE = 256 * 1024 * 1024 class RecordingDatabase: @@ -52,9 +67,11 @@ class RecordingDatabase: self.lock = threading.RLock() self.shards: dict[tuple[str, str], dict[str, Any]] = {} self.shard_paths: dict[tuple[str, str], Path] = {} + self.request_plans: dict[tuple[str, str, str], dict[str, Any]] = {} self.sessions: dict[str, dict[str, Any]] = {} self.fallback_consumed: set[tuple[str, str, str, int]] = set() self._load() + self._load_request_plans() def _load(self) -> None: manifest_path = self.root / "manifest.json" @@ -70,6 +87,17 @@ class RecordingDatabase: self.shards[key] = shard self.shard_paths[key] = path + def _load_request_plans(self) -> None: + root = self.root.parent / "test-runner-data" + manifest_path = root / "manifest.json" + if not manifest_path.exists(): + return + manifest = _read_json(manifest_path) + for item in manifest.get("scenarios", []): + plan = _read_json(root / item["file"]) + key = (item["version"], item["feature"], item["scenario"]) + self.request_plans[key] = plan.get("request", {}) + def start(self, version: str, feature: str, scenario: str, mode: str) -> dict[str, Any]: with self.lock: key = (version, feature) @@ -91,13 +119,14 @@ class RecordingDatabase: "key": key, "scenario": scenario, "recording": recording, + "request_plan": self.request_plans.get((version, feature, scenario)), "cursor": 0, "captures": [], "frozen_at": frozen_at, } return {"session": session_id, "frozen_at": frozen_at} - def replay(self, session_id: str | None, actual: dict[str, Any]) -> dict[str, Any]: + def replay(self, session_id: str | None, actual: dict[str, Any]) -> tuple[dict[str, Any], str | None]: with self.lock: if session_id: session = self.sessions.get(session_id) @@ -109,20 +138,32 @@ class RecordingDatabase: if cursor >= len(interactions): raise LookupError(f"Recording has no interaction #{cursor + 1}") expected = interactions[cursor] - if not _requests_match(expected["request"], actual): + request_plan = session["request_plan"] + if not _requests_match(expected["request"], actual, request_plan): raise RequestMismatchError(expected["request"], actual, cursor) session["cursor"] += 1 - return expected["response"] + compression = ( + request_plan.get("compression") + if request_plan and _request_matches_plan(expected["request"], request_plan) + else expected["request"].get("compression") + ) + return expected["response"], compression for (version, feature), shard in sorted(self.shards.items()): for recording in shard["recordings"]: for index, interaction in enumerate(recording["interactions"]): consumed_key = (version, feature, recording["scenario"], index) + request_plan = self.request_plans.get((version, feature, recording["scenario"])) if consumed_key not in self.fallback_consumed and _requests_match( - interaction["request"], actual + interaction["request"], actual, request_plan ): self.fallback_consumed.add(consumed_key) - return interaction["response"] + compression = ( + request_plan.get("compression") + if request_plan and _request_matches_plan(interaction["request"], request_plan) + else interaction["request"].get("compression") + ) + return interaction["response"], compression raise LookupError("No unconsumed interaction matches this request") def next_request(self, session_id: str) -> dict[str, Any]: @@ -298,14 +339,21 @@ class TestRequestHandler(BaseHTTPRequestHandler): def _handle_api_request(self) -> None: body = self._read_body() - actual = _normalise_request(self.command, self.path, self.headers.get("content-type", ""), body) + actual = _normalise_request( + self.command, + self.path, + self.headers.get("content-type", ""), + self.headers.get("content-encoding", ""), + body, + ) session_id = self.headers.get(SESSION_HEADER) if self.server.mode == "replay": - response = self.server.database.replay(session_id, actual) + response, compression = self.server.database.replay(session_id, actual) else: response = self._forward(body) self.server.database.capture(session_id, actual, response) - self._send_recorded_response(response) + compression = None + self._send_recorded_response(response, compression=compression) def _forward(self, body: bytes) -> dict[str, Any]: if not self.server.upstream: @@ -341,17 +389,37 @@ class TestRequestHandler(BaseHTTPRequestHandler): length = int(self.headers.get("content-length", "0")) return self.rfile.read(length) if length else b"" - def _send_recorded_response(self, response: dict[str, Any]) -> None: + def _send_recorded_response(self, response: dict[str, Any], *, compression: str | None = None) -> None: body_data = response.get("body", {}) if body_data.get("encoding") == "base64": body = base64.b64decode(body_data.get("value", "")) else: body = body_data.get("value", "").encode("utf-8") status = response["status"] + response_headers = response.get("headers", {}) + recorded_encoding = next( + (str(value) for key, value in response_headers.items() if key.lower() == "content-encoding"), + "", + ) + recorded_compression = recorded_encoding.strip().casefold() + if compression and _status_allows_message_content(status): + if recorded_compression: + decoded = _decompress_body(body, recorded_compression) + if decoded is None: + raise ValueError(f"Unsupported or invalid recorded response compression: {recorded_encoding}") + body = decoded + body = _compress_body(body, compression) self.send_response(status, response.get("reason")) - for key, value in response.get("headers", {}).items(): - if key.lower() not in HOP_BY_HOP_HEADERS | {"content-length"} | SAFE_BROWSER_HEADERS.keys(): + for key, value in response_headers.items(): + if ( + key.lower() + not in HOP_BY_HOP_HEADERS | {"content-encoding", "content-length"} | SAFE_BROWSER_HEADERS.keys() + ): self.send_header(key, value) + if compression and _status_allows_message_content(status): + self.send_header("content-encoding", compression) + elif recorded_encoding and _status_allows_message_content(status): + self.send_header("content-encoding", recorded_encoding) for key, value in SAFE_BROWSER_HEADERS.items(): self.send_header(key, value) if _status_allows_message_content(status): @@ -372,21 +440,50 @@ class TestRequestHandler(BaseHTTPRequestHandler): self.wfile.write(body) -def _normalise_request(method: str, raw_path: str, content_type: str, body: bytes) -> dict[str, Any]: +def _normalise_request( + method: str, + raw_path: str, + content_type: str, + content_encoding: str, + body: bytes, +) -> dict[str, Any]: parsed = urlsplit(raw_path) - normalised_body = _normalise_body(body, content_type) - return { + compression = content_encoding.strip().casefold() + normalised_body = _normalise_body(body, content_type, compression) + request = { "method": method.upper(), "path": _normalise_path(parsed.path), "query": sorted([list(pair) for pair in parse_qsl(parsed.query, keep_blank_values=True)]), "content_type": _normalise_content_type(content_type, normalised_body), "body": normalised_body, } + if compression: + request["compression"] = compression + return request -def _normalise_body(body: bytes, content_type: str) -> dict[str, Any]: +def _normalise_body(body: bytes, content_type: str, compression: str = "") -> dict[str, Any]: if not body: + if compression in {"gzip", "deflate", "zstd1"}: + return {"type": "invalid-compression", "value": ""} return {"type": "empty", "value": None} + if compression == "zstd1": + decompressed = _decompress_zstd(body) + if decompressed is None: + return { + "type": "invalid-compression", + "value": base64.b64encode(body).decode("ascii"), + } + body = decompressed + elif compression in {"gzip", "deflate"}: + wbits = zlib.MAX_WBITS | 16 if compression == "gzip" else zlib.MAX_WBITS + decompressed = _decompress_zlib(body, wbits=wbits) + if decompressed is None: + return { + "type": "invalid-compression", + "value": base64.b64encode(body).decode("ascii"), + } + body = decompressed media_type = _media_type(content_type) text = body.decode("utf-8", errors="surrogateescape") if media_type.endswith("json"): @@ -401,6 +498,224 @@ def _normalise_body(body: bytes, content_type: str) -> dict[str, Any]: return {"type": "text", "value": text} +def _decompress_zlib( + body: bytes, + *, + wbits: int, + max_size: int = MAX_DECOMPRESSED_BODY_SIZE, +) -> bytes | None: + """Decompress exactly one complete zlib stream with bounded output.""" + decompressor = zlib.decompressobj(wbits) + output = bytearray() + pending = body + try: + while pending: + remaining = max_size - len(output) + 1 + chunk = decompressor.decompress(pending, remaining) + output.extend(chunk) + if len(output) > max_size: + return None + unconsumed = decompressor.unconsumed_tail + if not unconsumed: + break + if len(unconsumed) == len(pending) and not chunk: + return None + pending = unconsumed + except zlib.error: + return None + if not decompressor.eof or decompressor.unused_data or decompressor.unconsumed_tail: + return None + return bytes(output) + + +def _compress_body(body: bytes, compression: str) -> bytes: + if compression == "gzip": + compressor = zlib.compressobj(wbits=zlib.MAX_WBITS | 16) + return compressor.compress(body) + compressor.flush() + if compression == "deflate": + return zlib.compress(body) + if compression == "zstd1": + return _compress_zstd(body) + raise ValueError(f"Unsupported response compression: {compression}") + + +def _decompress_body(body: bytes, compression: str) -> bytes | None: + if compression == "gzip": + return _decompress_zlib(body, wbits=zlib.MAX_WBITS | 16) + if compression == "deflate": + return _decompress_zlib(body, wbits=zlib.MAX_WBITS) + if compression == "zstd1": + return _decompress_zstd(body) + return None + + +def _compress_zstd(body: bytes) -> bytes: + """Compress one Zstandard frame with the platform libzstd.""" + library = _load_zstd() + if library is None: + return _compress_zstd_raw_frame(body) + try: + library.ZSTD_isError.argtypes = [ctypes.c_size_t] + library.ZSTD_isError.restype = ctypes.c_uint + library.ZSTD_compressBound.argtypes = [ctypes.c_size_t] + library.ZSTD_compressBound.restype = ctypes.c_size_t + library.ZSTD_compress.argtypes = [ + ctypes.c_void_p, + ctypes.c_size_t, + ctypes.c_void_p, + ctypes.c_size_t, + ctypes.c_int, + ] + library.ZSTD_compress.restype = ctypes.c_size_t + + source = ctypes.create_string_buffer(body) + capacity = library.ZSTD_compressBound(len(body)) + destination = ctypes.create_string_buffer(max(1, capacity)) + compressed_size = library.ZSTD_compress(destination, capacity, source, len(body), 3) + if library.ZSTD_isError(compressed_size): + return _compress_zstd_raw_frame(body) + return destination.raw[:compressed_size] + except (AttributeError, OSError, OverflowError, TypeError): + return _compress_zstd_raw_frame(body) + + +def _compress_zstd_raw_frame(body: bytes) -> bytes: + """Create a valid Zstandard frame made only of dependency-free raw blocks.""" + size = len(body) + if size <= UINT8_MAX: + descriptor = 0x20 + content_size = size.to_bytes(1, "little") + elif size <= UINT16_WITH_OFFSET_MAX: + descriptor = 0x60 + content_size = (size - ZSTD_FCS_TWO_BYTE_OFFSET).to_bytes(2, "little") + elif size <= UINT32_MAX: + descriptor = 0xA0 + content_size = size.to_bytes(4, "little") + else: + descriptor = 0xE0 + content_size = size.to_bytes(8, "little") + + frame = bytearray(ZSTD_MAGIC) + frame.append(descriptor) + frame.extend(content_size) + offset = 0 + while offset < size: + chunk = body[offset : offset + ZSTD_RAW_BLOCK_SIZE] + offset += len(chunk) + header = (len(chunk) << 3) | int(offset == size) + frame.extend(header.to_bytes(3, "little")) + frame.extend(chunk) + if not body: + frame.extend(b"\x01\x00\x00") + return bytes(frame) + + +def _load_zstd() -> Any | None: + candidates = ( + find_library("zstd"), + "libzstd.so.1", + "libzstd.dylib", + "/opt/homebrew/lib/libzstd.dylib", + "/usr/local/lib/libzstd.dylib", + "libzstd.dll", + "zstd.dll", + ) + for candidate in dict.fromkeys(item for item in candidates if item): + try: + return ctypes.CDLL(candidate) + except OSError: + continue + return None + + +def _decompress_zstd(body: bytes) -> bytes | None: + """Decompress one complete Zstandard frame with the platform libzstd.""" + library = _load_zstd() + if library is None: + return _decompress_zstd_raw_frame(body) + + try: + library.ZSTD_isError.argtypes = [ctypes.c_size_t] + library.ZSTD_isError.restype = ctypes.c_uint + library.ZSTD_findFrameCompressedSize.argtypes = [ctypes.c_void_p, ctypes.c_size_t] + library.ZSTD_findFrameCompressedSize.restype = ctypes.c_size_t + library.ZSTD_getFrameContentSize.argtypes = [ctypes.c_void_p, ctypes.c_size_t] + library.ZSTD_getFrameContentSize.restype = ctypes.c_ulonglong + library.ZSTD_decompressBound.argtypes = [ctypes.c_void_p, ctypes.c_size_t] + library.ZSTD_decompressBound.restype = ctypes.c_ulonglong + library.ZSTD_decompress.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_void_p, ctypes.c_size_t] + library.ZSTD_decompress.restype = ctypes.c_size_t + + source = ctypes.create_string_buffer(body) + source_pointer = ctypes.cast(source, ctypes.c_void_p) + compressed_size = library.ZSTD_findFrameCompressedSize(source_pointer, len(body)) + if library.ZSTD_isError(compressed_size) or compressed_size != len(body): + return None + + capacity = library.ZSTD_getFrameContentSize(source_pointer, len(body)) + if capacity == ZSTD_CONTENTSIZE_ERROR: + return None + if capacity == ZSTD_CONTENTSIZE_UNKNOWN: + capacity = library.ZSTD_decompressBound(source_pointer, len(body)) + if capacity > MAX_DECOMPRESSED_BODY_SIZE: + return None + + destination = ctypes.create_string_buffer(max(1, capacity)) + decompressed_size = library.ZSTD_decompress(destination, capacity, source_pointer, len(body)) + if library.ZSTD_isError(decompressed_size) or decompressed_size > capacity: + return None + return destination.raw[:decompressed_size] + except (AttributeError, OSError, OverflowError, TypeError): + return None + + +def _decompress_zstd_raw_frame(body: bytes) -> bytes | None: + """Decode the raw-block Zstandard frames emitted by the dependency-free fallback.""" + if len(body) < ZSTD_MIN_RAW_FRAME_SIZE or body[:4] != ZSTD_MAGIC: + return None + descriptor = body[4] + if descriptor & ZSTD_DESCRIPTOR_LOW_BITS_MASK != ZSTD_SINGLE_SEGMENT_FLAG: + return None + size_flag = descriptor >> 6 + size_bytes = (1, 2, 4, 8)[size_flag] + cursor = 5 + if len(body) < cursor + size_bytes: + return None + expected_size = int.from_bytes(body[cursor : cursor + size_bytes], "little") + if size_flag == 1: + expected_size += ZSTD_FCS_TWO_BYTE_OFFSET + if expected_size > MAX_DECOMPRESSED_BODY_SIZE: + return None + cursor += size_bytes + output = bytearray() + last_block = False + while not last_block: + if len(body) < cursor + 3: + return None + header = int.from_bytes(body[cursor : cursor + 3], "little") + cursor += 3 + last_block = bool(header & 1) + block_type = (header >> 1) & 0x3 + block_size = header >> 3 + if block_type == 0: + if len(body) < cursor + block_size: + return None + output.extend(body[cursor : cursor + block_size]) + cursor += block_size + elif block_type == 1: + if cursor >= len(body): + return None + output.extend(body[cursor : cursor + 1] * block_size) + cursor += 1 + else: + return None + if len(output) > MAX_DECOMPRESSED_BODY_SIZE: + return None + if cursor != len(body) or len(output) != expected_size: + return None + return bytes(output) + + def _normalise_json(value: Any) -> Any: if isinstance(value, dict): return {key: _normalise_json(item) for key, item in value.items()} @@ -415,13 +730,33 @@ def _normalise_content_type(content_type: str, body: dict[str, Any]) -> str: return "" if body["type"] == "empty" else _media_type(content_type) -def _requests_match(expected: dict[str, Any], actual: dict[str, Any]) -> bool: +def _requests_match( + expected: dict[str, Any], + actual: dict[str, Any], + request_plan: dict[str, Any] | None = None, +) -> bool: comparable_fields = ("method", "path", "query", "content_type") if any(expected[field] != actual[field] for field in comparable_fields): return False + expected_compression = expected.get("compression") + if request_plan and _request_matches_plan(expected, request_plan): + expected_compression = request_plan.get("compression", expected_compression) + if expected_compression is not None and actual.get("compression", "") != expected_compression: + return False return _bodies_match(expected["body"], actual["body"]) +def _request_matches_plan(request: dict[str, Any], plan: dict[str, Any]) -> bool: + if request["method"] != plan.get("method"): + return False + path = plan.get("path") + if not path: + return False + parts = re.split(r"(\{[^/{}]+\})", path) + pattern = "".join(r"[^/]+" if part.startswith("{") else re.escape(part) for part in parts) + return re.fullmatch(pattern, request["path"]) is not None + + def _bodies_match(expected: dict[str, Any], actual: dict[str, Any]) -> bool: if expected == actual: return True diff --git a/tests/generated-test/test-server-data/v1/logs.json b/tests/generated-test/test-server-data/v1/logs.json index 3ade4a2a92..d52869f93d 100644 --- a/tests/generated-test/test-server-data/v1/logs.json +++ b/tests/generated-test/test-server-data/v1/logs.json @@ -41,6 +41,80 @@ "scenario": "Search test logs returns \"OK\" response", "version": "v1" }, + { + "feature": "Logs", + "frozen_at": "2026-02-02T02:40:00.000Z", + "interactions": [ + { + "request": { + "body": { + "type": "json", + "value": [ + { + "ddtags": "host:TestSenddeflatelogsreturnsResponsefromserveralways200emptyJSONresponse1770000000", + "message": "Test-Send_deflate_logs_returns_Response_from_server_always_200_empty_JSON_response-1770000000" + } + ] + }, + "compression": "deflate", + "content_type": "application/json", + "method": "POST", + "path": "/v1/input", + "query": [] + }, + "response": { + "body": { + "encoding": "text", + "value": "{}" + }, + "headers": { + "content-type": "application/json" + }, + "reason": "OK", + "status": 200 + } + } + ], + "scenario": "Send deflate logs returns \"Response from server (always 200 empty JSON).\" response", + "version": "v1" + }, + { + "feature": "Logs", + "frozen_at": "2026-02-02T02:40:00.000Z", + "interactions": [ + { + "request": { + "body": { + "type": "json", + "value": [ + { + "ddtags": "host:TestSendgziplogsreturnsResponsefromserveralways200emptyJSONresponse1770000000", + "message": "Test-Send_gzip_logs_returns_Response_from_server_always_200_empty_JSON_response-1770000000" + } + ] + }, + "compression": "gzip", + "content_type": "application/json", + "method": "POST", + "path": "/v1/input", + "query": [] + }, + "response": { + "body": { + "encoding": "text", + "value": "{}" + }, + "headers": { + "content-type": "application/json" + }, + "reason": "OK", + "status": 200 + } + } + ], + "scenario": "Send gzip logs returns \"Response from server (always 200 empty JSON).\" response", + "version": "v1" + }, { "feature": "Logs", "frozen_at": "2022-04-12T14:46:46.337Z", diff --git a/tests/generated-test/test-server-data/v2/logs.json b/tests/generated-test/test-server-data/v2/logs.json index d4b92521f2..16edc4255f 100644 --- a/tests/generated-test/test-server-data/v2/logs.json +++ b/tests/generated-test/test-server-data/v2/logs.json @@ -467,6 +467,86 @@ "scenario": "Search logs returns \"OK\" response with pagination", "version": "v2" }, + { + "feature": "Logs", + "frozen_at": "2026-02-02T02:40:00.000Z", + "interactions": [ + { + "request": { + "body": { + "type": "json", + "value": [ + { + "ddsource": "nginx", + "ddtags": "env:staging,version:5.1", + "hostname": "i-012345678", + "message": "2019-11-19T14:37:58,995 INFO [process.name][20081] Hello World", + "service": "payment" + } + ] + }, + "compression": "deflate", + "content_type": "application/json", + "method": "POST", + "path": "/api/v2/logs", + "query": [] + }, + "response": { + "body": { + "encoding": "text", + "value": "{}" + }, + "headers": { + "content-type": "application/json" + }, + "reason": "Accepted", + "status": 202 + } + } + ], + "scenario": "Send deflate logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "version": "v2" + }, + { + "feature": "Logs", + "frozen_at": "2026-02-02T02:40:00.000Z", + "interactions": [ + { + "request": { + "body": { + "type": "json", + "value": [ + { + "ddsource": "nginx", + "ddtags": "env:staging,version:5.1", + "hostname": "i-012345678", + "message": "2019-11-19T14:37:58,995 INFO [process.name][20081] Hello World", + "service": "payment" + } + ] + }, + "compression": "gzip", + "content_type": "application/json", + "method": "POST", + "path": "/api/v2/logs", + "query": [] + }, + "response": { + "body": { + "encoding": "text", + "value": "{}" + }, + "headers": { + "content-type": "application/json" + }, + "reason": "Accepted", + "status": 202 + } + } + ], + "scenario": "Send gzip logs returns \"Request accepted for processing (always 202 empty JSON).\" response", + "version": "v2" + }, { "feature": "Logs", "frozen_at": "2024-10-01T15:36:43.563Z", diff --git a/tests/v1/features/logs.feature b/tests/v1/features/logs.feature index 5ae909c3aa..51f1619013 100644 --- a/tests/v1/features/logs.feature +++ b/tests/v1/features/logs.feature @@ -32,19 +32,21 @@ Feature: Logs When the request is sent Then the response status is 200 OK - @integration-only @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake + @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake Scenario: Send deflate logs returns "Response from server (always 200 empty JSON)." response Given new "SubmitLog" request And body with value [{"message": "{{ unique }}", "ddtags": "host:{{ unique_alnum }}"}] - And request contains "Content-Encoding" parameter with value "deflate" + And the client selects "deflate" compression + And the request and response use "deflate" compression When the request is sent Then the response status is 200 Response from server (always 200 empty JSON). - @integration-only @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake + @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake Scenario: Send gzip logs returns "Response from server (always 200 empty JSON)." response Given new "SubmitLog" request And body with value [{"message": "{{ unique }}", "ddtags": "host:{{ unique_alnum }}"}] - And request contains "Content-Encoding" parameter with value "gzip" + And the client selects "gzip" compression + And the request and response use "gzip" compression When the request is sent Then the response status is 200 Response from server (always 200 empty JSON). diff --git a/tests/v2/features/logs.feature b/tests/v2/features/logs.feature index b7accd07dd..84c58b2094 100644 --- a/tests/v2/features/logs.feature +++ b/tests/v2/features/logs.feature @@ -128,19 +128,21 @@ Feature: Logs Then the response status is 200 OK And the response has 3 items - @integration-only @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake @team:DataDog/logs-backend @team:DataDog/logs-ingestion + @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake @team:DataDog/logs-backend @team:DataDog/logs-ingestion Scenario: Send deflate logs returns "Request accepted for processing (always 202 empty JSON)." response Given new "SubmitLog" request And body with value [{"ddsource": "nginx", "ddtags": "env:staging,version:5.1", "hostname": "i-012345678", "message": "2019-11-19T14:37:58,995 INFO [process.name][20081] Hello World", "service": "payment"}] - And request contains "Content-Encoding" parameter with value "deflate" + And the client selects "deflate" compression + And the request and response use "deflate" compression When the request is sent Then the response status is 202 Response from server (always 202 empty JSON). - @integration-only @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake @team:DataDog/logs-backend @team:DataDog/logs-ingestion + @skip-terraform-config @skip-validation @team:DataDog/event-platform-intake @team:DataDog/logs-backend @team:DataDog/logs-ingestion Scenario: Send gzip logs returns "Request accepted for processing (always 202 empty JSON)." response Given new "SubmitLog" request And body with value [{"ddsource": "nginx", "ddtags": "env:staging,version:5.1", "hostname": "i-012345678", "message": "2019-11-19T14:37:58,995 INFO [process.name][20081] Hello World", "service": "payment"}] - And request contains "Content-Encoding" parameter with value "gzip" + And the client selects "gzip" compression + And the request and response use "gzip" compression When the request is sent Then the response status is 202 Request accepted for processing (always 202 empty JSON).