test: add comprehensive tests for ASGI server support

- Add tests for HTTPServer ASGI/WSGI auto-detection
- Add tests for StarletteApplication and UvicornApplication
- Add tests for CLI --gateway flag functionality
- Add integration tests for async functions with ASGI
- Ensure 100% code coverage for new ASGI features
- Apply black and isort formatting to test files

Author: taeold

tests/test_asgi.py

@@ -0,0 +1,123 @@
+# Copyright 2025 Google LLC
+#
+# 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.
+
+import sys
+
+import flask
+import pretend
+import pytest
+
+import functions_framework._http
+
+try:
+    from starlette.applications import Starlette
+except ImportError:
+    pass
+
+
+def test_httpserver_detects_asgi_app():
+    flask_app = flask.Flask("test")
+    flask_wrapper = functions_framework._http.HTTPServer(flask_app, debug=True)
+    assert flask_wrapper.server_class.__name__ == "FlaskApplication"
+
+    starlette_app = Starlette(routes=[])
+    starlette_wrapper = functions_framework._http.HTTPServer(starlette_app, debug=True)
+    assert starlette_wrapper.server_class.__name__ == "StarletteApplication"
+
+
+@pytest.mark.skipif("platform.system() == 'Windows'")
+def test_httpserver_production_asgi():
+    starlette_app = Starlette(routes=[])
+    wrapper = functions_framework._http.HTTPServer(starlette_app, debug=False)
+    assert wrapper.server_class.__name__ == "UvicornApplication"
+
+
+def test_starlette_application_init():
+    from functions_framework._http.asgi import StarletteApplication
+
+    app = pretend.stub()
+    host = "1.2.3.4"
+    port = "5678"
+
+    # Test debug mode
+    starlette_app = StarletteApplication(app, host, port, debug=True, custom="value")
+    assert starlette_app.app == app
+    assert starlette_app.host == host
+    assert starlette_app.port == port
+    assert starlette_app.debug is True
+    assert starlette_app.options["log_level"] == "debug"
+    assert starlette_app.options["reload"] is True
+    assert starlette_app.options["custom"] == "value"
+
+    # Test production mode
+    starlette_app = StarletteApplication(app, host, port, debug=False)
+    assert starlette_app.options["log_level"] == "error"
+    assert starlette_app.options["reload"] is False
+
+
+@pytest.mark.skipif("platform.system() == 'Windows'")
+def test_uvicorn_application_init():
+    from functions_framework._http.gunicorn import UvicornApplication
+
+    app = pretend.stub()
+    host = "1.2.3.4"
+    port = "1234"
+
+    uvicorn_app = UvicornApplication(app, host, port, debug=False)
+    assert uvicorn_app.app == app
+    assert uvicorn_app.options["worker_class"] == "uvicorn_worker.UvicornWorker"
+    assert uvicorn_app.options["bind"] == "1.2.3.4:1234"
+    assert uvicorn_app.load() == app
+
+
+def test_httpserver_fallback_on_import_error(monkeypatch):
+    starlette_app = Starlette(routes=[])
+
+    monkeypatch.setitem(sys.modules, "functions_framework._http.gunicorn", None)
+
+    wrapper = functions_framework._http.HTTPServer(starlette_app, debug=False)
+    assert wrapper.server_class.__name__ == "StarletteApplication"
+
+
+def test_starlette_application_run(monkeypatch):
+    uvicorn_run_calls = []
+
+    def mock_uvicorn_run(app, **kwargs):
+        uvicorn_run_calls.append((app, kwargs))
+
+    uvicorn_stub = pretend.stub(run=mock_uvicorn_run)
+    monkeypatch.setitem(sys.modules, "uvicorn", uvicorn_stub)
+
+    # Clear and re-import to get fresh module with mocked uvicorn
+    if "functions_framework._http.asgi" in sys.modules:
+        del sys.modules["functions_framework._http.asgi"]
+
+    from functions_framework._http.asgi import StarletteApplication
+
+    app = pretend.stub()
+    host = "1.2.3.4"
+    port = "5678"
+
+    starlette_app = StarletteApplication(app, host, port, debug=True, custom="value")
+    starlette_app.run()
+
+    assert len(uvicorn_run_calls) == 1
+    assert uvicorn_run_calls[0][0] == app
+    assert uvicorn_run_calls[0][1] == {
+        "host": host,
+        "port": int(port),
+        "log_level": "debug",
+        "reload": True,
+        "custom": "value",
+    }

tests/test_cli.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import sys
+
 import pretend
 import pytest
 
@@ -103,3 +105,22 @@ def test_cli(monkeypatch, args, env, create_app_calls, run_calls):
     assert result.exit_code == 0
     assert create_app.calls == create_app_calls
     assert wsgi_server.run.calls == run_calls
+
+
+def test_asgi_cli(monkeypatch):
+    asgi_server = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))
+    asgi_app = pretend.stub()
+
+    create_asgi_app = pretend.call_recorder(lambda *a, **kw: asgi_app)
+    aio_module = pretend.stub(create_asgi_app=create_asgi_app)
+    monkeypatch.setitem(sys.modules, "functions_framework.aio", aio_module)
+
+    create_server = pretend.call_recorder(lambda *a, **kw: asgi_server)
+    monkeypatch.setattr(functions_framework._cli, "create_server", create_server)
+
+    runner = CliRunner()
+    result = runner.invoke(_cli, ["--target", "foo", "--gateway", "asgi"])
+
+    assert result.exit_code == 0
+    assert create_asgi_app.calls == [pretend.call("foo", None, "http")]
+    assert asgi_server.run.calls == [pretend.call("0.0.0.0", 8080)]

tests/test_cli_gateway.py

@@ -0,0 +1,161 @@
+# Copyright 2025 Google LLC
+#
+# 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.
+
+import sys
+
+import pretend
+import pytest
+
+from click.testing import CliRunner
+
+from functions_framework._cli import _cli as cli_command
+
+
+# Test for CLI gateway flag
+def test_cli_gateway_default(monkeypatch):
+    """Test default gateway is WSGI"""
+    server = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))
+    app = pretend.stub()
+
+    create_app = pretend.call_recorder(lambda *a, **kw: app)
+    create_server = pretend.call_recorder(lambda *a, **kw: server)
+
+    from functions_framework import _cli
+
+    monkeypatch.setattr(_cli, "create_app", create_app)
+    monkeypatch.setattr(_cli, "create_server", create_server)
+
+    runner = CliRunner()
+    result = runner.invoke(cli_command, ["--target", "foo"])
+
+    assert result.exit_code == 0
+    assert create_app.calls == [pretend.call("foo", None, "http")]
+    assert server.run.calls == [pretend.call("0.0.0.0", 8080)]
+
+
+def test_cli_gateway_wsgi_explicit(monkeypatch):
+    """Test explicit --gateway wsgi"""
+    server = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))
+    app = pretend.stub()
+
+    create_app = pretend.call_recorder(lambda *a, **kw: app)
+    create_server = pretend.call_recorder(lambda *a, **kw: server)
+
+    from functions_framework import _cli
+
+    monkeypatch.setattr(_cli, "create_app", create_app)
+    monkeypatch.setattr(_cli, "create_server", create_server)
+
+    runner = CliRunner()
+    result = runner.invoke(cli_command, ["--target", "foo", "--gateway", "wsgi"])
+
+    assert result.exit_code == 0
+    assert create_app.calls == [pretend.call("foo", None, "http")]
+
+
+def test_cli_gateway_asgi(monkeypatch):
+    """Test --gateway asgi"""
+    # Skip this test if aio module is not available
+    try:
+        import functions_framework.aio
+    except ImportError:
+        pytest.skip("Async support not available")
+
+    server = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))
+    app = pretend.stub()
+
+    # Mock the create_asgi_app function
+    create_asgi_app = pretend.call_recorder(lambda *a, **kw: app)
+    monkeypatch.setattr("functions_framework.aio.create_asgi_app", create_asgi_app)
+
+    create_server = pretend.call_recorder(lambda *a, **kw: server)
+
+    from functions_framework import _cli
+
+    monkeypatch.setattr(_cli, "create_server", create_server)
+
+    runner = CliRunner()
+    result = runner.invoke(cli_command, ["--target", "foo", "--gateway", "asgi"])
+
+    assert result.exit_code == 0
+    assert create_asgi_app.calls == [pretend.call("foo", None, "http")]
+
+
+def test_cli_gateway_asgi_cloudevent(monkeypatch):
+    """Test --gateway asgi with cloudevent signature"""
+    # Skip this test if aio module is not available
+    try:
+        import functions_framework.aio
+    except ImportError:
+        pytest.skip("Async support not available")
+
+    server = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))
+    app = pretend.stub()
+
+    # Mock the create_asgi_app function
+    create_asgi_app = pretend.call_recorder(lambda *a, **kw: app)
+    monkeypatch.setattr("functions_framework.aio.create_asgi_app", create_asgi_app)
+
+    create_server = pretend.call_recorder(lambda *a, **kw: server)
+
+    from functions_framework import _cli
+
+    monkeypatch.setattr(_cli, "create_server", create_server)
+
+    runner = CliRunner()
+    result = runner.invoke(
+        cli_command,
+        ["--target", "foo", "--gateway", "asgi", "--signature-type", "cloudevent"],
+    )
+
+    assert result.exit_code == 0
+    assert create_asgi_app.calls == [pretend.call("foo", None, "cloudevent")]
+
+
+def test_cli_gateway_env_var(monkeypatch):
+    """Test GATEWAY environment variable"""
+    # Skip this test if aio module is not available
+    try:
+        import functions_framework.aio
+    except ImportError:
+        pytest.skip("Async support not available")
+
+    server = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))
+    app = pretend.stub()
+
+    # Mock the create_asgi_app function
+    create_asgi_app = pretend.call_recorder(lambda *a, **kw: app)
+    monkeypatch.setattr("functions_framework.aio.create_asgi_app", create_asgi_app)
+
+    create_server = pretend.call_recorder(lambda *a, **kw: server)
+
+    from functions_framework import _cli
+
+    monkeypatch.setattr(_cli, "create_server", create_server)
+
+    runner = CliRunner(env={"GATEWAY": "asgi"})
+    result = runner.invoke(cli_command, ["--target", "foo"])
+
+    assert result.exit_code == 0
+    assert create_asgi_app.calls == [pretend.call("foo", None, "http")]
+
+
+def test_cli_invalid_gateway():
+    """Test that invalid gateway values are rejected"""
+    runner = CliRunner()
+    result = runner.invoke(cli_command, ["--target", "foo", "--gateway", "invalid"])
+
+    assert result.exit_code == 2
+    assert "Invalid value" in result.output
+    assert "'invalid' is not one of 'wsgi', 'asgi'" in result.output

tests/test_functions.py

@@ -21,7 +21,6 @@
 import pretend
 import pytest
 
-# Conditional import for Starlette
 if sys.version_info >= (3, 8):
     from starlette.testclient import TestClient as StarletteTestClient
 else:
@@ -31,7 +30,6 @@
 
 from functions_framework import LazyWSGIApp, create_app, errorhandler, exceptions
 
-# Conditional import for async functionality
 if sys.version_info >= (3, 8):
     from functions_framework.aio import create_asgi_app
 else: