tests: integration: fix http server response overrides

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

tests/integration/scenarios/out_http/tests/test_out_http_001.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+import time
 
 import requests
 
@@ -9,11 +10,39 @@
     configure_oauth_token_response,
     data_storage,
     http_server_run,
+    server_instances,
 )
 from utils.test_service import FluentBitTestService
 
 logger = logging.getLogger(__name__)
 
+def _wait_for_http_server(port, timeout=5):
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        try:
+            response = requests.get(f"http://127.0.0.1:{port}/ping", timeout=1)
+            if response.status_code == 200:
+                return
+        except requests.RequestException:
+            pass
+
+        time.sleep(0.1)
+
+    raise TimeoutError(f"Timed out waiting for HTTP server on port {port}")
+
+
+def _wait_for_http_server_port(timeout=5):
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        if server_instances:
+            return server_instances[-1].server_port
+
+        time.sleep(0.1)
+
+    raise TimeoutError("Timed out waiting for HTTP server port assignment")
+
 
 class Service:
     def __init__(self, config_file, *, response_setup=None, use_tls=False):
@@ -281,3 +310,93 @@ def test_out_http_tls_read_idle_timeout_retries_partial_response():
 
     assert len(requests_seen) >= 2
     assert "read idle timeout reached" in log_text
+
+
+def test_http_server_configure_helpers_allow_clearing_nullable_fields():
+    http_server_run(0, reset_state=True)
+    port = _wait_for_http_server_port()
+    _wait_for_http_server(port)
+
+    try:
+        configure_http_response(
+            stream_fragments=["part"],
+            hang_after_fragment_index=0,
+        )
+        configure_http_response(
+            stream_fragments=None,
+            hang_after_fragment_index=None,
+        )
+
+        response = requests.post(
+            f"http://127.0.0.1:{port}/data",
+            json={"test": "clear"},
+            timeout=5,
+        )
+
+        assert response.status_code == 200
+        assert response.json() == {"status": "received"}
+    finally:
+        try:
+            requests.post(f"http://127.0.0.1:{port}/shutdown", timeout=2)
+        except requests.RequestException:
+            pass
+
+
+def test_http_server_oauth_token_honors_explicit_content_type_and_raw_body():
+    http_server_run(0, reset_state=True)
+    port = _wait_for_http_server_port()
+    _wait_for_http_server(port)
+
+    try:
+        configure_oauth_token_response(
+            stream_fragments=["partial-token"],
+            hang_after_fragment_index=0,
+        )
+        configure_oauth_token_response(
+            body="not-json",
+            content_type="text/plain",
+            stream_fragments=None,
+            hang_after_fragment_index=None,
+        )
+
+        response = requests.post(
+            f"http://127.0.0.1:{port}/oauth/token",
+            data="grant_type=client_credentials",
+            timeout=5,
+        )
+
+        assert response.status_code == 200
+        assert response.text == "not-json"
+        assert response.headers["Content-Type"].startswith("text/plain")
+    finally:
+        try:
+            requests.post(f"http://127.0.0.1:{port}/shutdown", timeout=2)
+        except requests.RequestException:
+            pass
+
+
+def test_http_server_oauth_token_honors_explicit_json_content_type():
+    http_server_run(0, reset_state=True)
+    port = _wait_for_http_server_port()
+    _wait_for_http_server(port)
+
+    try:
+        configure_oauth_token_response(
+            body={"access_token": "json-token", "token_type": "Bearer"},
+            content_type="application/json; charset=utf-8",
+        )
+
+        response = requests.post(
+            f"http://127.0.0.1:{port}/oauth/token",
+            data="grant_type=client_credentials",
+            timeout=5,
+        )
+
+        assert response.status_code == 200
+        assert response.json()["access_token"] == "json-token"
+        assert "application/json" in response.headers["Content-Type"]
+    finally:
+        try:
+            requests.post(f"http://127.0.0.1:{port}/shutdown", timeout=2)
+        except requests.RequestException:
+            pass

tests/integration/src/server/http_server.py

@@ -63,6 +63,7 @@
 server_thread = None
 server_instances = []
 shutdown_event = threading.Event()
+UNSET = object()
 
 
 def _sleep_interruptible(seconds):
@@ -115,51 +116,51 @@ def reset_http_server_state():
     )
 
 
-def configure_http_response(*, status_code=None, body=None, content_type=None,
-                            delay_seconds=None, stream_fragments=None,
-                            fragment_delay_seconds=None,
-                            hang_before_response=None,
-                            hang_after_fragment_index=None):
-    if status_code is not None:
+def configure_http_response(*, status_code=UNSET, body=UNSET, content_type=UNSET,
+                            delay_seconds=UNSET, stream_fragments=UNSET,
+                            fragment_delay_seconds=UNSET,
+                            hang_before_response=UNSET,
+                            hang_after_fragment_index=UNSET):
+    if status_code is not UNSET:
         response_config["status_code"] = status_code
-    if body is not None:
+    if body is not UNSET:
         response_config["body"] = body
-    if content_type is not None:
+    if content_type is not UNSET:
         response_config["content_type"] = content_type
-    if delay_seconds is not None:
+    if delay_seconds is not UNSET:
         response_config["delay_seconds"] = delay_seconds
-    if stream_fragments is not None:
-        response_config["stream_fragments"] = list(stream_fragments)
-    if fragment_delay_seconds is not None:
+    if stream_fragments is not UNSET:
+        response_config["stream_fragments"] = None if stream_fragments is None else list(stream_fragments)
+    if fragment_delay_seconds is not UNSET:
         response_config["fragment_delay_seconds"] = fragment_delay_seconds
-    if hang_before_response is not None:
+    if hang_before_response is not UNSET:
         response_config["hang_before_response"] = hang_before_response
-    if hang_after_fragment_index is not None:
+    if hang_after_fragment_index is not UNSET:
         response_config["hang_after_fragment_index"] = hang_after_fragment_index
 
 
-def configure_oauth_token_response(*, status_code=None, body=None,
-                                   content_type=None,
-                                   delay_seconds=None,
-                                   hang_before_response=None,
-                                   stream_fragments=None,
-                                   fragment_delay_seconds=None,
-                                   hang_after_fragment_index=None):
-    if status_code is not None:
+def configure_oauth_token_response(*, status_code=UNSET, body=UNSET,
+                                   content_type=UNSET,
+                                   delay_seconds=UNSET,
+                                   hang_before_response=UNSET,
+                                   stream_fragments=UNSET,
+                                   fragment_delay_seconds=UNSET,
+                                   hang_after_fragment_index=UNSET):
+    if status_code is not UNSET:
         oauth_token_response["status_code"] = status_code
-    if body is not None:
+    if body is not UNSET:
         oauth_token_response["body"] = body
-    if content_type is not None:
+    if content_type is not UNSET:
         oauth_token_response["content_type"] = content_type
-    if delay_seconds is not None:
+    if delay_seconds is not UNSET:
         oauth_token_response["delay_seconds"] = delay_seconds
-    if hang_before_response is not None:
+    if hang_before_response is not UNSET:
         oauth_token_response["hang_before_response"] = hang_before_response
-    if stream_fragments is not None:
-        oauth_token_response["stream_fragments"] = list(stream_fragments)
-    if fragment_delay_seconds is not None:
+    if stream_fragments is not UNSET:
+        oauth_token_response["stream_fragments"] = None if stream_fragments is None else list(stream_fragments)
+    if fragment_delay_seconds is not UNSET:
         oauth_token_response["fragment_delay_seconds"] = fragment_delay_seconds
-    if hang_after_fragment_index is not None:
+    if hang_after_fragment_index is not UNSET:
         oauth_token_response["hang_after_fragment_index"] = hang_after_fragment_index
 
 
@@ -284,7 +285,26 @@ def oauth_token():
         return Response(status=503)
     if oauth_token_response["stream_fragments"] is not None:
         return _build_streaming_response(oauth_token_response)
-    return jsonify(oauth_token_response["body"]), oauth_token_response["status_code"]
+
+    body = oauth_token_response["body"]
+    content_type = oauth_token_response.get("content_type")
+    normalized_content_type = None
+
+    if content_type is not None:
+        normalized_content_type = content_type.split(";", 1)[0].strip().lower()
+
+    if isinstance(body, (dict, list)) and (
+        normalized_content_type is None or
+        normalized_content_type == "application/json" or
+        normalized_content_type.endswith("+json")
+    ):
+        return jsonify(body), oauth_token_response["status_code"]
+
+    return Response(
+        body,
+        status=oauth_token_response["status_code"],
+        content_type=content_type,
+    )
 
 
 @app.route('/ping', methods=['GET'])