# Copyright (c) Streamlit Inc. (2018-2022) Snowflake Inc. (2022-2026)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unit tests for the Streamlit CLI."""

from __future__ import annotations

import contextlib
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock, patch

import pytest
import requests
import requests_mock
from click.testing import CliRunner
from parameterized import parameterized
from requests.adapters import HTTPAdapter, Retry
from testfixtures import tempdir

import streamlit.web.bootstrap
from streamlit import config
from streamlit.config_option import ConfigOption
from streamlit.runtime.credentials import Credentials
from streamlit.web import cli
from streamlit.web.cli import _convert_config_option_to_click_option
from tests import testutil


class CliTest(unittest.TestCase):
    """Unit tests for the cli."""

    def setUp(self):
        # Credentials._singleton should be None here, but a mis-behaving
        # test may have left it intact.
        Credentials._singleton = None

        cli.name = "streamlit"
        self.runner = CliRunner()

        self.patches = [
            patch.object(config._on_config_parsed, "send"),
            # Make sure the calls to `streamlit run` in this file don't unset
            # the config options loaded in conftest.py.
            patch.object(streamlit.web.bootstrap, "load_config_options"),
        ]

        for p in self.patches:
            p.start()

    def tearDown(self):
        Credentials._singleton = None

        for p in self.patches:
            p.stop()

    def test_run_no_file_argument_but_default_exists(self):
        """streamlit run should succeed when run with no arguments and the default file exists."""
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run") as mock_main_run,
            patch("pathlib.Path.exists", return_value=True),
        ):
            result = self.runner.invoke(cli, ["run"])
        assert result.exit_code == 0

        mock_main_run.assert_called_once()
        positional_args = mock_main_run.call_args[0]
        assert positional_args[0] == "streamlit_app.py"

    def test_run_no_file_argument_and_default_doesnt_exist(self):
        """streamlit run should fail if run with no arguments and default file doesn't exist."""
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run"),
            patch("pathlib.Path.exists", return_value=False),
        ):
            result = self.runner.invoke(cli, ["run", "file_name.py"])
        assert result.exit_code != 0

    def test_run_existing_file_argument(self):
        """streamlit run succeeds if an existing file is passed."""
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run") as mock_main_run,
            patch("pathlib.Path.exists", return_value=True),
        ):
            result = self.runner.invoke(cli, ["run", "file_name.py"])
        assert result.exit_code == 0

        mock_main_run.assert_called_once()
        positional_args = mock_main_run.call_args[0]
        assert positional_args[0] == "file_name.py"

    def test_run_existing_path_argument(self):
        """streamlit run succeeds if an existing path is passed."""
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run") as mock_main_run,
            patch("pathlib.Path.exists", return_value=True),
            patch("pathlib.Path.is_dir", return_value=True),
        ):
            result = self.runner.invoke(cli, ["run", "foo/bar"])
        assert result.exit_code == 0

        mock_main_run.assert_called_once()
        positional_args = mock_main_run.call_args[0]
        assert positional_args[0] == "foo/bar/streamlit_app.py"

    def test_run_non_existing_file_argument(self):
        """streamlit run should fail if a non existing file is passed."""

        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run"),
            patch("pathlib.Path.exists", return_value=False),
        ):
            result = self.runner.invoke(cli, ["run", "file_name.py"])
        assert result.exit_code != 0
        assert "File does not exist" in result.output

    def test_run_not_allowed_file_extension(self):
        """streamlit run should fail if a not allowed file extension is passed."""

        result = self.runner.invoke(cli, ["run", "file_name.doc"])

        assert result.exit_code != 0
        assert "Streamlit requires raw Python (.py) files, not .doc." in result.output

    @tempdir()
    def test_run_valid_url(self, temp_dir):
        """streamlit run succeeds if an existing url is passed."""

        with (
            patch("streamlit.url_util.is_url", return_value=True),
            patch("streamlit.web.cli._main_run"),
            requests_mock.mock() as m,
        ):
            file_content = b"content"
            m.get("http://url/app.py", content=file_content)
            with patch("streamlit.temporary_directory.TemporaryDirectory") as mock_tmp:
                mock_tmp.return_value.__enter__.return_value = temp_dir.path
                result = self.runner.invoke(cli, ["run", "http://url/app.py"])
                with open(os.path.join(temp_dir.path, "app.py"), "rb") as f:
                    assert file_content == f.read()

        assert result.exit_code == 0

    @tempdir()
    def test_run_non_existing_url(self, temp_dir):
        """streamlit run should fail if a non existing but valid
        url is passed.
        """

        with (
            patch("streamlit.url_util.is_url", return_value=True),
            patch("streamlit.web.cli._main_run"),
            requests_mock.mock() as m,
        ):
            m.get("http://url/app.py", exc=requests.exceptions.RequestException)
            with patch("streamlit.temporary_directory.TemporaryDirectory") as mock_tmp:
                mock_tmp.return_value.__enter__.return_value = temp_dir.path
                result = self.runner.invoke(cli, ["run", "http://url/app.py"])

        assert result.exit_code != 0
        assert "Unable to fetch" in result.output

    def test_run_arguments(self):
        """The correct command line should be passed downstream."""
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli._main_run") as mock_main_run,
        ):
            result = self.runner.invoke(
                cli,
                [
                    "run",
                    "some script.py",
                    "argument with space",
                    "argument with another space",
                ],
            )
        mock_main_run.assert_called_once()
        positional_args = mock_main_run.call_args[0]
        assert positional_args[0] == "some script.py"
        assert positional_args[1] == (
            "argument with space",
            "argument with another space",
        )
        assert result.exit_code == 0

    def test_run_command_with_flag_config_options(self):
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.bootstrap.run"),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli.check_credentials"),
        ):
            result = self.runner.invoke(
                cli, ["run", "file_name.py", "--server.port=8502"]
            )

        streamlit.web.bootstrap.load_config_options.assert_called_once()
        _args, kwargs = streamlit.web.bootstrap.load_config_options.call_args
        assert kwargs["flag_options"]["server_port"] == 8502
        assert result.exit_code == 0

    def test_run_command_with_multiple_secrets_path_single_value(self):
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.bootstrap.run"),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli.check_credentials"),
        ):
            result = self.runner.invoke(
                cli, ["run", "file_name.py", "--secrets.files=secrets1.toml"]
            )

        streamlit.web.bootstrap.load_config_options.assert_called_once()
        _args, kwargs = streamlit.web.bootstrap.load_config_options.call_args
        assert kwargs["flag_options"]["secrets_files"] == ("secrets1.toml",)
        assert result.exit_code == 0

    def test_run_command_with_multiple_secrets_path_multiple_value(self):
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.bootstrap.run"),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli.check_credentials"),
        ):
            result = self.runner.invoke(
                cli,
                [
                    "run",
                    "file_name.py",
                    "--secrets.files=secrets1.toml",
                    "--secrets.files=secrets2.toml",
                ],
            )

        streamlit.web.bootstrap.load_config_options.assert_called_once()
        _args, kwargs = streamlit.web.bootstrap.load_config_options.call_args
        assert kwargs["flag_options"]["secrets_files"] == (
            "secrets1.toml",
            "secrets2.toml",
        )
        assert result.exit_code == 0

    @parameterized.expand(["mapbox.token", "server.cookieSecret"])
    def test_run_command_with_sensitive_options_as_flag(self, sensitive_option):
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run"),
            patch("pathlib.Path.exists", return_value=True),
        ):
            result = self.runner.invoke(
                cli, ["run", "file_name.py", f"--{sensitive_option}=TESTSECRET"]
            )

        assert "option using the CLI flag is not allowed" in result.output
        assert result.exit_code == 1

    def test_get_command_line(self):
        """Test that _get_command_line_as_string correctly concatenates values
        from click.
        """
        mock_context = MagicMock()
        mock_context.parent.command_path = "streamlit"
        with patch("click.get_current_context", return_value=mock_context):
            with patch.object(sys, "argv", ["", "os_arg1", "os_arg2"]):
                result = cli._get_command_line_as_string()
                assert result == "streamlit os_arg1 os_arg2"

    def test_get_command_line_without_parent_context(self):
        """Test that _get_command_line_as_string correctly returns None when
        there is no context parent
        """
        mock_context = MagicMock()
        mock_context.parent = None
        with patch("click.get_current_context", return_value=mock_context):
            result = cli._get_command_line_as_string()
            assert result is None

    def test_convert_config_option_to_click_option(self):
        """Test that configurator_options adds dynamic commands based on a
        config lists.
        """
        config_option = ConfigOption(
            "server.customKey",
            description="Custom description.\n\nLine one.",
            deprecated=False,
            type_=int,
        )

        result = _convert_config_option_to_click_option(config_option)

        assert result["option"] == "--server.customKey"
        assert result["param"] == "server_customKey"
        assert result["type"] == config_option.type
        assert result["description"] == config_option.description
        assert result["envvar"] == "STREAMLIT_SERVER_CUSTOM_KEY"

    def test_convert_depecated_config_option_to_click_option(self):
        """Test that configurator_options adds extra deprecation information
        to config option's description
        """
        config_option = ConfigOption(
            "deprecated.customKey",
            description="Custom description.\n\nLine one.",
            deprecated=True,
            deprecation_text="Foo",
            expiration_date="Bar",
            type_=int,
        )

        result = _convert_config_option_to_click_option(config_option)

        assert result["description"] == "Custom description.\n\nLine one.\n Foo - Bar"

    def test_credentials_headless_no_config(self):
        """If headless mode and no config is present,
        activation should be None."""
        with testutil.patch_config_options({"server.headless": True}):
            with (
                patch("streamlit.url_util.is_url", return_value=False),
                patch("streamlit.web.bootstrap.run"),
                patch("pathlib.Path.exists", return_value=True),
                patch(
                    "streamlit.runtime.credentials._check_credential_file_exists",
                    return_value=False,
                ),
            ):
                result = self.runner.invoke(cli, ["run", "some script.py"])
            from streamlit.runtime.credentials import Credentials

            credentials = Credentials.get_current()
            assert credentials.activation is None
            assert result.exit_code == 0

    @parameterized.expand([(True,), (False,)])
    def test_credentials_headless_with_config(self, headless_mode):
        """If headless, but a config file is present, activation should be
        defined.
        So we call `_check_activated`.
        """
        with testutil.patch_config_options({"server.headless": headless_mode}):
            with (
                patch("streamlit.url_util.is_url", return_value=False),
                patch("streamlit.web.bootstrap.run"),
                patch("pathlib.Path.exists", return_value=True),
                mock.patch(
                    "streamlit.runtime.credentials.Credentials._check_activated"
                ) as mock_check,
                patch(
                    "streamlit.runtime.credentials._check_credential_file_exists",
                    return_value=True,
                ),
            ):
                result = self.runner.invoke(cli, ["run", "some script.py"])
            assert mock_check.called
            assert result.exit_code == 0

    @parameterized.expand([(True,), (False,)])
    def test_headless_telemetry_message(self, headless_mode):
        """If headless mode, show a message about usage metrics gathering."""

        with testutil.patch_config_options({"server.headless": headless_mode}):
            with (
                patch("streamlit.url_util.is_url", return_value=False),
                patch("pathlib.Path.exists", return_value=True),
                patch("streamlit.config.is_manually_set", return_value=False),
                patch(
                    "streamlit.runtime.credentials._check_credential_file_exists",
                    return_value=False,
                ),
                patch("streamlit.web.bootstrap.run"),
            ):
                result = self.runner.invoke(cli, ["run", "file_name.py"])

            assert ("Collecting usage statistics" in result.output) == headless_mode, (
                f"Telemetry message mode is {headless_mode} "
                f"yet output is: {result.output}"
            )

    @parameterized.expand([(False, False), (False, True), (True, False), (True, True)])
    def test_prompt_welcome_message(self, prompt_mode, headless_mode):
        """If prompt is true, show a welcome prompt, unless headless."""

        with testutil.patch_config_options(
            {"server.showEmailPrompt": prompt_mode, "server.headless": headless_mode}
        ):
            with (
                patch("streamlit.url_util.is_url", return_value=False),
                patch("pathlib.Path.exists", return_value=True),
                patch("streamlit.config.is_manually_set", return_value=False),
                patch(
                    "streamlit.runtime.credentials._check_credential_file_exists",
                    return_value=False,
                ),
                patch("streamlit.web.bootstrap.run"),
            ):
                result = self.runner.invoke(cli, ["run", "file_name.py"])

            assert (prompt_mode and not headless_mode) == (
                "like to receive helpful onboarding emails, news, offers, promotions,"
                in result.output
            ), (
                f"Welcome message mode is {prompt_mode} "
                f"and headless mode is {headless_mode} "
                f"yet output is: {result.output}"
            )

    def test_streamlit_folder_not_created_when_show_email_prompt_false(self):
        """Test that ~/.streamlit directory is not created when server.showEmailPrompt=False."""
        with testutil.patch_config_options(
            {"server.showEmailPrompt": False, "server.headless": False}
        ):
            with (
                patch("streamlit.url_util.is_url", return_value=False),
                patch("pathlib.Path.exists", return_value=True),
                patch("streamlit.config.is_manually_set", return_value=False),
                patch(
                    "streamlit.runtime.credentials._check_credential_file_exists",
                    return_value=False,
                ),
                patch("streamlit.runtime.credentials.os.makedirs") as mock_makedirs,
                patch("streamlit.web.bootstrap.run"),
            ):
                result = self.runner.invoke(cli, ["run", "file_name.py"])

            # Assert that makedirs was never called to create ~/.streamlit directory
            mock_makedirs.assert_not_called()
            assert result.exit_code == 0

    def test_streamlit_folder_created_when_show_email_prompt_true(self):
        """Test that ~/.streamlit directory is created when server.showEmailPrompt=True."""
        with testutil.patch_config_options(
            {"server.showEmailPrompt": True, "server.headless": False}
        ):
            with (
                patch("streamlit.url_util.is_url", return_value=False),
                patch("pathlib.Path.exists", return_value=True),
                patch("streamlit.config.is_manually_set", return_value=False),
                patch(
                    "streamlit.runtime.credentials._check_credential_file_exists",
                    return_value=False,
                ),
                patch("streamlit.runtime.credentials.os.makedirs") as mock_makedirs,
                patch("streamlit.web.bootstrap.run"),
                patch("click.prompt", return_value="test@example.com") as mock_prompt,
                patch(
                    "streamlit.runtime.credentials.open", mock.mock_open(), create=True
                ),
                patch("streamlit.runtime.credentials._send_email"),
            ):
                result = self.runner.invoke(cli, ["run", "file_name.py"])

            # Assert that makedirs was called to create ~/.streamlit directory
            mock_makedirs.assert_called_once()
            # Assert that the email prompt was shown
            mock_prompt.assert_called_once()
            assert result.exit_code == 0

    def test_help_command(self):
        """Tests the help command redirects to using the --help flag"""
        with patch.object(sys, "argv", ["streamlit", "help"]) as args:
            self.runner.invoke(cli, ["help"])
            assert args[1] == "--help"

    def test_version_command(self):
        """Tests the version command redirects to using the --version flag"""
        with patch.object(sys, "argv", ["streamlit", "version"]) as args:
            self.runner.invoke(cli, ["version"])
            assert args[1] == "--version"

    def test_docs_command(self):
        """Tests the docs command opens the browser"""
        with patch("streamlit.cli_util.open_browser") as mock_open_browser:
            self.runner.invoke(cli, ["docs"])
            mock_open_browser.assert_called_once_with("https://docs.streamlit.io")

    def test_hello_command(self):
        """Tests the hello command runs the hello script in streamlit"""
        from streamlit.hello import streamlit_app

        with patch("streamlit.web.cli._main_run") as mock_main_run:
            self.runner.invoke(cli, ["hello"])

            mock_main_run.assert_called_once()
            positional_args = mock_main_run.call_args[0]
            assert positional_args[0] == streamlit_app.__file__

    @patch("streamlit.logger.get_logger")
    def test_hello_command_with_logs(self, get_logger):
        """Tests setting log level using --log_level prints a warning."""

        with patch("streamlit.web.cli._main_run"):
            self.runner.invoke(cli, ["--log_level", "error", "hello"])

            mock_logger = get_logger()
            mock_logger.warning.assert_called_once()

    def test_hello_command_with_flag_config_options(self):
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.bootstrap.run"),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli.check_credentials"),
        ):
            result = self.runner.invoke(cli, ["hello", "--server.port=8502"])

        streamlit.web.bootstrap.load_config_options.assert_called_once()
        _args, kwargs = streamlit.web.bootstrap.load_config_options.call_args
        assert kwargs["flag_options"]["server_port"] == 8502
        assert result.exit_code == 0

    def test_config_show_command(self):
        """Tests the config show command calls the corresponding method in
        config
        """
        with patch("streamlit.config.show_config") as mock_config:
            self.runner.invoke(cli, ["config", "show"])
            mock_config.assert_called()

    def test_config_show_command_with_flag_config_options(self):
        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("streamlit.web.cli._main_run"),
            patch("pathlib.Path.exists", return_value=True),
        ):
            result = self.runner.invoke(cli, ["config", "show", "--server.port=8502"])

        streamlit.web.bootstrap.load_config_options.assert_called_once()
        _args, kwargs = streamlit.web.bootstrap.load_config_options.call_args
        assert kwargs["flag_options"]["server_port"] == 8502
        assert result.exit_code == 0

    @patch(
        "streamlit.runtime.caching.storage.local_disk_cache_storage.LocalDiskCacheStorageManager.clear_all"
    )
    @patch("streamlit.runtime.caching.cache_resource.clear")
    def test_cache_clear_all_caches(self, clear_resource_caches, clear_data_caches):
        """cli.clear_cache should clear st.cache_data and st.cache_resource"""
        self.runner.invoke(cli, ["cache", "clear"])
        clear_resource_caches.assert_called_once()
        clear_data_caches.assert_called_once()

    def test_activate_command(self):
        """Tests activating a credential"""
        mock_credential = MagicMock()
        with mock.patch(
            "streamlit.runtime.credentials.Credentials.get_current",
            return_value=mock_credential,
        ):
            self.runner.invoke(cli, ["activate"])
            mock_credential.activate.assert_called()

    def test_activate_without_command(self):
        """Tests that it doesn't activate the credential when not specified"""
        mock_credential = MagicMock()
        with mock.patch(
            "streamlit.runtime.credentials.Credentials.get_current",
            return_value=mock_credential,
        ):
            self.runner.invoke(cli)
            mock_credential.activate.assert_not_called()

    def test_reset_command(self):
        """Tests resetting a credential"""
        mock_credential = MagicMock()
        with mock.patch(
            "streamlit.runtime.credentials.Credentials.get_current",
            return_value=mock_credential,
        ):
            self.runner.invoke(cli, ["activate", "reset"])
            mock_credential.reset.assert_called()

    def test_init_command(self):
        """Test creating a new project in current directory."""
        with tempfile.TemporaryDirectory() as tmpdir:
            orig_dir = os.getcwd()
            os.chdir(tmpdir)
            try:
                result = self.runner.invoke(cli, ["init"], input="n\n")

                # Check command output
                assert result.exit_code == 0

                # Check created files
                assert Path(tmpdir, "requirements.txt").exists()
                assert Path(tmpdir, "streamlit_app.py").exists()

                # Check file contents
                assert "streamlit" in Path(tmpdir, "requirements.txt").read_text(
                    encoding="utf-8"
                )
                assert "import streamlit as st" in Path(
                    tmpdir, "streamlit_app.py"
                ).read_text(encoding="utf-8")
            finally:
                os.chdir(orig_dir)

    def test_init_command_with_directory(self):
        """Test creating a new project in specified directory."""
        with tempfile.TemporaryDirectory() as tmpdir:
            orig_dir = os.getcwd()
            os.chdir(tmpdir)
            try:
                result = self.runner.invoke(cli, ["init", "new-project"], input="n\n")

                # Check command output
                assert result.exit_code == 0

                # Check created files
                project_dir = Path(tmpdir) / "new-project"
                assert (project_dir / "requirements.txt").exists()
                assert (project_dir / "streamlit_app.py").exists()
            finally:
                os.chdir(orig_dir)

    def test_run_detects_st_app_and_calls_asgi_bootstrap(self):
        """Test that _main_run detects st.App and calls run_asgi_app."""
        from streamlit.web.server.app_discovery import AppDiscoveryResult

        mock_discovery_result = AppDiscoveryResult(
            is_asgi_app=True,
            app_name="app",
            import_string="mymodule:app",
        )

        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli.check_credentials"),
            patch(
                "streamlit.web.server.app_discovery.discover_asgi_app",
                return_value=mock_discovery_result,
            ),
            patch("streamlit.web.bootstrap.run_asgi_app") as mock_run_asgi,
            patch("streamlit.web.bootstrap.run") as mock_run,
        ):
            result = self.runner.invoke(cli, ["run", "mymodule.py"])

        # Should call run_asgi_app, not run
        mock_run_asgi.assert_called_once()
        mock_run.assert_not_called()
        assert result.exit_code == 0

    def test_run_uses_regular_bootstrap_for_non_st_app(self):
        """Test that _main_run uses regular bootstrap for non-st.App scripts."""
        from streamlit.web.server.app_discovery import AppDiscoveryResult

        mock_discovery_result = AppDiscoveryResult(
            is_asgi_app=False,
            app_name=None,
            import_string=None,
        )

        with (
            patch("streamlit.url_util.is_url", return_value=False),
            patch("pathlib.Path.exists", return_value=True),
            patch("streamlit.web.cli.check_credentials"),
            patch(
                "streamlit.web.server.app_discovery.discover_asgi_app",
                return_value=mock_discovery_result,
            ),
            patch("streamlit.web.bootstrap.run_asgi_app") as mock_run_asgi,
            patch("streamlit.web.bootstrap.run") as mock_run,
        ):
            result = self.runner.invoke(cli, ["run", "regular_script.py"])

        # Should call run, not run_asgi_app
        mock_run.assert_called_once()
        mock_run_asgi.assert_not_called()
        assert result.exit_code == 0


class HTTPServerIntegrationTest(unittest.TestCase):
    def tearDown(self) -> None:
        from streamlit.watcher.event_based_path_watcher import EventBasedPathWatcher

        EventBasedPathWatcher.close_all()

    def get_http_session(self) -> requests.Session:
        http_session = requests.Session()
        http_session.mount(
            "https://", HTTPAdapter(max_retries=Retry(total=10, backoff_factor=0.2))
        )
        http_session.mount("http://", HTTPAdapter(max_retries=None))
        return http_session

    @unittest.skipIf(
        "win32" in sys.platform, "openssl not present on windows by default"
    )
    def test_ssl(self):
        with contextlib.ExitStack() as exit_stack:
            tmp_home = exit_stack.enter_context(tempfile.TemporaryDirectory())
            (Path(tmp_home) / ".streamlit").mkdir()
            (Path(tmp_home) / ".streamlit" / "credentials.toml").write_text(
                '[general]\nemail = ""', encoding="utf-8"
            )
            cert_file = Path(tmp_home) / "cert.cert"
            key_file = Path(tmp_home) / "key.key"
            pem_file = Path(tmp_home) / "public.pem"

            subprocess.check_call(
                [
                    "openssl",
                    "req",
                    "-x509",
                    "-newkey",
                    "rsa:4096",
                    "-keyout",
                    str(key_file),
                    "-out",
                    str(cert_file),
                    "-sha256",
                    "-days",
                    "365",
                    "-nodes",
                    "-subj",
                    "/CN=localhost",
                    # sublectAltName is required by modern browsers
                    # See: https://github.com/urllib3/urllib3/issues/497
                    "-addext",
                    "subjectAltName = DNS:localhost",
                ]
            )
            subprocess.check_call(
                [
                    "openssl",
                    "x509",
                    "-inform",
                    "PEM",
                    "-in",
                    str(cert_file),
                    "-out",
                    str(pem_file),
                ]
            )
            https_session = exit_stack.enter_context(self.get_http_session())
            proc = exit_stack.enter_context(
                subprocess.Popen(
                    [
                        sys.executable,
                        "-m",
                        "streamlit",
                        "hello",
                        "--global.developmentMode=False",
                        "--server.sslCertFile",
                        str(cert_file),
                        "--server.sslKeyFile",
                        str(key_file),
                        "--server.headless",
                        "true",
                        "--server.port=8510",
                    ],
                    env={**os.environ, "HOME": tmp_home},
                )
            )
            try:
                response = https_session.get(
                    "https://localhost:8510/healthz", verify=str(pem_file)
                )
                response.raise_for_status()
                assert response.text == "ok"
                # HTTP traffic is restricted
                with pytest.raises(requests.exceptions.ConnectionError):
                    response = https_session.get("http://localhost:8510/healthz")
                    response.raise_for_status()
            finally:
                proc.kill()
