improved Fast_API_Server with support for capturing Stdout and Stderr

osbot_fast_api/utils/Fast_API_Server.py

@@ -3,29 +3,46 @@
 from urllib.parse           import urljoin
 from threading              import Thread
 from fastapi                import FastAPI
-from osbot_utils.utils.Http import wait_for_port, wait_for_port_closed, is_port_open
+from osbot_utils.base_classes.Type_Safe import Type_Safe
+from osbot_utils.testing.Stderr import Stderr
+from osbot_utils.testing.Stdout import Stdout
+from osbot_utils.utils.Http import wait_for_port, wait_for_port_closed, is_port_open, url_join_safe
+from osbot_utils.utils.Objects import base_types
 from uvicorn                import Config, Server
 from osbot_utils.utils.Misc import random_port
 
 FAST_API__HOST      = "127.0.0.1"
 FAST_API__LOG_LEVEL = "error"
 
-class Fast_API_Server:
-    def __init__(self, app, port=None, log_level=None):
-        self.app       : FastAPI = app
-        self.port      : int     = port or random_port()
-        self.log_level : str     = log_level or FAST_API__LOG_LEVEL
-        self.config    : Config  = Config(app=self.app, host=FAST_API__HOST, port=self.port, log_level=self.log_level)
-        self.server    : Server  = None
-        self.thread    : Thread  = None
-        self.running   : bool    = False
+class Fast_API_Server(Type_Safe):
+    app       : FastAPI
+    port      : int
+    log_level : str     = FAST_API__LOG_LEVEL
+    config    : Config  = None
+    server    : Server  = None
+    thread    : Thread  = None
+    running   : bool    = False
+    stdout    : Stdout
+    stderr    : Stderr
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        if self.port == 0:
+            self.port = random_port()
+        if self.config is None:
+            self.config = Config(app=self.app, host=FAST_API__HOST, port=self.port, log_level=self.log_level)
 
     def __enter__(self):
+        self.stderr.start()
+        self.stdout.start()
         self.start()
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.stop()
+        self.stdout.stop()
+        self.stderr.stop()
+        pass
 
     def is_port_open(self):
         return is_port_open(host=FAST_API__HOST, port=self.port)
@@ -50,8 +67,18 @@ def stop(self):
         return result
 
     def requests_get(self, path=''):
-        url = urljoin(self.url(), path)
+        url = url_join_safe(self.url(), path)
         return requests.get(url)
 
+    def requests_post(self, path='', data=None):
+        if Type_Safe in base_types(data):
+            json_data = data.json()
+        elif type(data) is dict:
+            json_data = data
+        else:
+            raise ValueError("data must be a Type_Safe or a dict")
+        url = urljoin(self.url(), path)
+        return requests.post(url, json=json_data)
+
     def url(self):
         return f'http://{FAST_API__HOST}:{self.port}/'

tests/integration/utils/test_Fast_API_Server.py

@@ -1,9 +1,8 @@
-from threading import Thread
-from unittest import TestCase
-
-from fastapi import FastAPI
-from uvicorn import Config, Server
-
+import sys
+from threading                            import Thread
+from unittest                             import TestCase
+from fastapi                              import FastAPI
+from uvicorn                              import Config, Server
 from osbot_fast_api.utils.Fast_API_Server import Fast_API_Server, FAST_API__LOG_LEVEL, FAST_API__HOST
 
 
@@ -43,10 +42,25 @@ def test_start__start(self):
         assert self.fast_api_server.stop()                is True
         assert self.fast_api_server.server.should_exit    is True
 
-        # this also works ok :)
-        # assert self.fast_api_server.start() is True
-        # pprint(self.fast_api_server.requests_get('docs').text)
-        # assert self.fast_api_server.stop() is True
 
+    def test__start_using_context(self):
+        with Fast_API_Server() as _:
+            assert _.requests_get('openapi.json').json() == { 'info'   : {'title': 'FastAPI', 'version': '0.1.0'},
+                                                              'openapi': '3.1.0',
+                                                              'paths'  : {}}
+
+    def test_capture_logs(self):
+        app = FastAPI()
+
+        @app.get("/ping")
+        def ping():
+            print("in ping")
+            print("error in ping", file=sys.stderr)
+            return {"it's": "pong"}
+
+        with Fast_API_Server(app=app) as _:
+            assert _.requests_get('ping').json() == {"it's": "pong"}
+            assert _.stdout.value() == 'in ping\n'
+            assert _.stderr.value() == 'error in ping\n'
 
 

tests/unit/api/routes/http_shell/test_Http_Shell__Client.py

@@ -1,4 +1,6 @@
 import os
+
+import pytest
 import requests
 from unittest                                           import TestCase
 
@@ -24,6 +26,8 @@ def setUpClass(cls) -> None:
         cls.fast_api            = Fast_API().setup()
         cls.fast_api_server     = Fast_API_Server(app=cls.fast_api.app())
         cls.auth_key            = os.environ.get(ENV__HTTP_SHELL_AUTH_KEY)
+        if cls.auth_key is None:
+            pytest.skip(f"env var {ENV__HTTP_SHELL_AUTH_KEY} not set, so skipping tests")
         cls.server_endpoint     = cls.fast_api_server.url() + 'http-shell-server'
         cls.client              = Http_Shell__Client(server_endpoint=cls.server_endpoint, auth_key=cls.auth_key, return_value_if_ok=False)
         cls.fast_api.add_route_post(cls.http_shell_server)

tests/unit/api/test_Fast_API.py

@@ -1,13 +1,11 @@
 from unittest import TestCase
 
+import pytest
 from fastapi import FastAPI
-from flask import Flask
 from starlette.testclient                       import TestClient
-from osbot_fast_api.api.Fast_API                import Fast_API, DEFAULT__NAME__FAST_API
 from osbot_fast_api.api.routes.Routes_Config    import ROUTES__CONFIG
 from osbot_fast_api.utils.Fast_API_Utils        import FAST_API_DEFAULT_ROUTES
 from osbot_fast_api.utils.Fast_API_Utils        import Fast_API_Utils
-from osbot_utils.utils.Dev import pprint
 from tests.unit.fast_api__for_tests import fast_api, fast_api_client
 
 EXPECTED_ROUTES_METHODS = ['redirect_to_docs', 'status', 'version']
@@ -23,7 +21,10 @@ def setUp(self):
     def test__init__(self):
         assert type(self.fast_api.app()) is FastAPI
 
+
     def test_add_flask_app(self):
+        flask = pytest.importorskip("flask", reason="Flask is not installed")
+        from flask import Flask
         path      = '/flask-app'
         flask_app = Flask(__name__)