To fix chatbot streaming request response with TLS communication enabled (#1732)

* fix streaming with TLS

* add UTC for added changes

* fix utc failure

Author: justjais

ansible_ai_connect/ai/api/model_pipelines/http/pipelines.py

@@ -262,7 +262,11 @@ def send_schema1_event(self, ev):
 
     async def async_invoke(self, params: StreamingChatBotParameters) -> AsyncGenerator:
 
-        async with aiohttp.ClientSession(raise_for_status=True) as session:
+        # Configure SSL context based on verify_ssl setting
+        ssl_context = self.config.verify_ssl
+        connector = aiohttp.TCPConnector(ssl=ssl_context)
+
+        async with aiohttp.ClientSession(raise_for_status=True, connector=connector) as session:
             headers = {
                 "Content-Type": "application/json",
                 "Accept": "application/json,text/event-stream",

ansible_ai_connect/ai/api/model_pipelines/http/tests/test_pipelines.py

@@ -14,9 +14,13 @@
 #  limitations under the License.
 import json
 import logging
+from typing import cast
 from unittest import IsolatedAsyncioTestCase
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
+from ansible_ai_connect.ai.api.model_pipelines.http.configuration import (
+    HttpConfiguration,
+)
 from ansible_ai_connect.ai.api.model_pipelines.http.pipelines import (
     HttpStreamingChatBotPipeline,
 )
@@ -105,7 +109,9 @@ class TestHttpStreamingChatBotPipeline(IsolatedAsyncioTestCase, WisdomLogAwareMi
     ]
 
     def setUp(self):
-        self.pipeline = HttpStreamingChatBotPipeline(mock_pipeline_config("http"))
+        self.pipeline = HttpStreamingChatBotPipeline(
+            cast(HttpConfiguration, mock_pipeline_config("http"))
+        )
         self.call_counter = 0
 
     def assertInLog(self, s, logs, number_of_matches_expected=None):
@@ -139,7 +145,7 @@ def get_params(self) -> StreamingChatBotParameters:
             provider="",
             model_id="",
             conversation_id=None,
-            system_prompt=None,
+            system_prompt="You are a helpful assistant",
             media_type="application/json",
         )
 
@@ -200,3 +206,91 @@ async def test_async_invoke_error_with_no_data(self, mock_post):
             async for _ in self.pipeline.async_invoke(self.get_params()):
                 pass
             self.assertInLog("(not provided)", log)
+
+    @patch("aiohttp.ClientSession.post")
+    @patch("aiohttp.TCPConnector")
+    async def test_ssl_context_verify_ssl_false(self, mock_tcp_connector, mock_post):
+        """Test that SSL context is correctly configured when verify_ssl=False"""
+        # Setup pipeline with verify_ssl=False
+        config = cast(HttpConfiguration, mock_pipeline_config("http", verify_ssl=False))
+        pipeline = HttpStreamingChatBotPipeline(config)
+        # Mock the connector
+        mock_connector_instance = MagicMock()
+        mock_tcp_connector.return_value = mock_connector_instance
+        # Mock the post method to return our test data
+        mock_post.return_value = self.get_return_value(self.STREAM_DATA)
+        # Execute the async_invoke method
+        params = self.get_params()
+        async for _ in pipeline.async_invoke(params):
+            pass
+        # Verify TCPConnector was created with ssl=False
+        mock_tcp_connector.assert_called_once_with(ssl=False)
+
+    @patch("aiohttp.ClientSession.post")
+    @patch("aiohttp.TCPConnector")
+    async def test_ssl_context_verify_ssl_true(self, mock_tcp_connector, mock_post):
+        """Test that SSL context is correctly configured when verify_ssl=True"""
+        # Setup pipeline with verify_ssl=True
+        config = cast(HttpConfiguration, mock_pipeline_config("http", verify_ssl=True))
+        pipeline = HttpStreamingChatBotPipeline(config)
+        # Mock the connector
+        mock_connector_instance = MagicMock()
+        mock_tcp_connector.return_value = mock_connector_instance
+        # Mock the post method to return our test data
+        mock_post.return_value = self.get_return_value(self.STREAM_DATA)
+        # Execute the async_invoke method
+        params = self.get_params()
+        async for _ in pipeline.async_invoke(params):
+            pass
+        # Verify TCPConnector was created with ssl=True
+        mock_tcp_connector.assert_called_once_with(ssl=True)
+
+    @patch("aiohttp.ClientSession.post")
+    @patch("aiohttp.TCPConnector")
+    async def test_ssl_context_uses_config_value(self, mock_tcp_connector, mock_post):
+        """Test that SSL context directly uses the config.verify_ssl value"""
+        # Setup pipeline with a specific verify_ssl value
+        config = cast(HttpConfiguration, mock_pipeline_config("http", verify_ssl=False))
+        pipeline = HttpStreamingChatBotPipeline(config)
+        # Mock the connector
+        mock_connector_instance = MagicMock()
+        mock_tcp_connector.return_value = mock_connector_instance
+        # Mock the post method to return our test data
+        mock_post.return_value = self.get_return_value(self.STREAM_DATA)
+        # Execute the async_invoke method
+        params = self.get_params()
+        async for _ in pipeline.async_invoke(params):
+            pass
+        # Verify that ssl_context was set to the exact config value
+        # The ssl parameter should match config.verify_ssl exactly
+        expected_ssl_value = config.verify_ssl  # False in this case
+        mock_tcp_connector.assert_called_once_with(ssl=expected_ssl_value)
+
+    @patch("aiohttp.ClientSession.post")
+    @patch("aiohttp.TCPConnector")
+    async def test_ssl_context_integration_with_existing_flow(self, mock_tcp_connector, mock_post):
+        """Test that SSL changes don't break existing functionality"""
+        # Test with both verify_ssl values to ensure no regression
+        for verify_ssl_value in [True, False]:
+            with self.subTest(verify_ssl=verify_ssl_value):
+                config = cast(
+                    HttpConfiguration, mock_pipeline_config("http", verify_ssl=verify_ssl_value)
+                )
+                pipeline = HttpStreamingChatBotPipeline(config)
+                # Mock the connector
+                mock_connector_instance = MagicMock()
+                mock_tcp_connector.return_value = mock_connector_instance
+                # Mock the post method to return our test data
+                mock_post.return_value = self.get_return_value(self.STREAM_DATA)
+                # Execute the async_invoke method
+                params = self.get_params()
+                result_count = 0
+                async for _ in pipeline.async_invoke(params):
+                    result_count += 1
+                # Verify that streaming still works
+                self.assertGreater(result_count, 0, "Streaming should return data")
+                # Verify SSL configuration is correct
+                mock_tcp_connector.assert_called_with(ssl=verify_ssl_value)
+                # Reset mocks for next iteration
+                mock_tcp_connector.reset_mock()
+                mock_post.reset_mock()