feat(https_server): Added Server Sent Events demo

This commit adds the demo of Server Sent Events
functionality in the https_server/simple example
Closes https://github.com/espressif/esp-idf/issues/13603

Co-authored-by: default avatarAditya Patwardhan <aditya.patwardhan@espressif.com>

examples/protocols/http_server/simple/pytest_http_server_simple.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# SPDX-FileCopyrightText: 2018-2022 Espressif Systems (Shanghai) CO LTD
+# SPDX-FileCopyrightText: 2018-2024 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Apache-2.0
 import logging
 import os
@@ -14,6 +14,7 @@ import time
 import pytest
 
 try:
+    import http.client
     from idf_http_server_test import client
 except ModuleNotFoundError:
     sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', 'tools', 'ci', 'python_packages'))
@@ -171,3 +172,58 @@ def test_examples_protocol_http_server_lru_purge_enable(dut: Dut) -> None:
 
     for t in threads:
         t.join()
+
+
+@pytest.mark.esp32
+@pytest.mark.wifi_router
+@pytest.mark.parametrize('config', ['sse',], indirect=True)
+def test_examples_protocol_http_server_sse(dut: Dut) -> None:
+    # Get binary file
+    binary_file = os.path.join(dut.app.binary_path, 'simple.bin')
+    bin_size = os.path.getsize(binary_file)
+    logging.info('http_server_bin_size : {}KB'.format(bin_size // 1024))
+
+    # Upload binary and start testing
+    logging.info('Starting http_server simple test app')
+
+    # Parse IP address of STA
+    logging.info('Waiting to connect with AP')
+    if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
+        dut.expect('Please input ssid password:')
+        env_name = 'wifi_router'
+        ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
+        ap_password = get_env_config_variable(env_name, 'ap_password')
+        dut.write(f'{ap_ssid} {ap_password}')
+    got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode()
+    got_port = int(dut.expect(r"(?:[\s\S]*)Starting server on port: '(\d+)'", timeout=30)[1].decode())
+
+    logging.info('Got IP   : {}'.format(got_ip))
+    logging.info('Got Port : {}'.format(got_port))
+
+    # Expected Logs
+    dut.expect('Registering URI handlers', timeout=30)
+
+    logging.info('Test /sse GET handler')
+    try:
+        logging.info(f'Connecting to {got_ip}:{got_port}')
+        conn = http.client.HTTPConnection(got_ip, got_port, timeout=15)
+        conn.request('GET', url='/sse')  # Ensure the URL path is correct
+        response = conn.getresponse()
+
+        # Process and verify only the first 5 lines of the response as the response is continuous
+        response_data = ''
+        for i, line in enumerate(response):
+            if i >= 5:
+                break
+            decoded_line = line.decode('utf-8').strip()
+            response_data += decoded_line
+
+        conn.close()
+
+        # Verify the format of the line
+        if 'data: Time since boot:' not in response_data:
+            raise RuntimeError(f'Unexpected line format: {response_data}')
+
+    except Exception as e:
+        logging.error(f'Error during SSE GET request: {e}')
+        raise RuntimeError('SSE handler test failed due to connection error')