IoSL/service-aware-frontend
3
0
Fork 0
generated from Luis/nextjs-python-web-template
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Better emulate metadata
Some checks failed
checks-impure / test (pull_request) Successful in 26s
Details
checks / test (pull_request) Failing after 1m47s
Details

Browse Source
This commit is contained in:
Luis-Hebendanz
2024-01-13 20:54:56 +01:00
parent 4ae4557602
commit f4cf817bf1
6 changed files with 67 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
pkgs/clan-cli/clan_cli/config.py
View File

@@ -1,12 +1,8 @@
import dataclasses
@dataclasses.dataclass
class Config:
host: str
port_dlg: int
port_ap: int
port_client_base: int
config = Config(host="127.0.0.1", port_dlg=6000, port_ap=6600, port_client_base=7000)
host = "127.0.0.1"
port_dlg = 7000
port_ap = 7500
port_client_base = 8000
dlg_url = f"http://{host}:{port_dlg}/docs"
ap_url = f"http://{host}:{port_ap}/docs"
c1_url = f"http://{host}:{port_client_base}/docs"
c2_url = f"http://{host}:{port_client_base + 1}/docs"

31
pkgs/clan-cli/clan_cli/emulate_fastapi.py
View File

@@ -6,14 +6,13 @@ from datetime import datetime
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse
import clan_cli.config as config
from clan_cli.webui.schemas import Resolution
from .config import config
app_dlg = FastAPI()
app_ap = FastAPI()
app_c1 = FastAPI()
app_c2 = FastAPI()
app_dlg = FastAPI(swagger_ui_parameters={"tryItOutEnabled": True})
app_ap = FastAPI(swagger_ui_parameters={"tryItOutEnabled": True})
app_c1 = FastAPI(swagger_ui_parameters={"tryItOutEnabled": True})
app_c2 = FastAPI(swagger_ui_parameters={"tryItOutEnabled": True})
apps = [
(app_dlg, config.port_dlg),
@@ -24,21 +23,41 @@ apps = [
#### HEALTHCHECK
@app_c1.get("/")
async def root_c1() -> str:
return "C1 is alive"
@app_c1.get("/health")
async def healthcheck_c1() -> str:
return "200 OK"
@app_c2.get("/")
async def root_c2() -> str:
return "C2 is alive"
@app_c2.get("/health")
async def healthcheck_c2() -> str:
return "200 OK"
@app_dlg.get("/")
async def root_dlg() -> str:
return "DLG is alive"
@app_dlg.get("/health")
async def healthcheck_dlg() -> str:
return "200 OK"
@app_ap.get("/")
async def root_ap() -> str:
return "AP is alive"
@app_ap.get("/health")
async def healthcheck_ap() -> str:
return "200 OK"

29
pkgs/clan-cli/clan_cli/webui/routers/endpoints.py
View File

@@ -4,6 +4,7 @@ from typing import List, Optional
import httpx
from fastapi import APIRouter, BackgroundTasks, Depends, Query
from fastapi.responses import HTMLResponse
from sqlalchemy.orm import Session
from ...errors import ClanError
@@ -291,3 +292,31 @@ def get_all_eventmessages(
) -> List[sql_models.Eventmessage]:
eventmessages = sql_crud.get_eventmessages(db, skip=skip, limit=limit)
return eventmessages
##############################
# #
# EMULATED API ENDPOINTS #
# #
##############################
@router.get("/emulate", response_class=HTMLResponse)
@router.get("/emu", response_class=HTMLResponse)
def get_emulated_enpoints() -> HTMLResponse:
from clan_cli.config import ap_url, c1_url, c2_url, dlg_url
html_content = f"""
<html>
<head>
<title>Emulated API</title>
</head>
<body>
<h1>Emulated API</h1>
<p>Emulated API endpoints for testing purposes.</p>
<p>DLG: <a href="{dlg_url}" >{dlg_url} </a></p>
<p>AP: <a href="{ap_url}">{ap_url}</a></p>
<p>C1: <a href="{c1_url}">{c1_url} </a></p>
<p>C2: <a href="{c2_url}">{c2_url}</a></p>
</body>
</html>
"""
return HTMLResponse(content=html_content, status_code=200)

7
pkgs/clan-cli/clan_cli/webui/server.py
View File

@@ -15,6 +15,7 @@ import uvicorn
from pydantic import AnyUrl, IPvAnyAddress
from pydantic.tools import parse_obj_as
import clan_cli.config as config
from clan_cli.emulate_fastapi import apps, get_health
from clan_cli.errors import ClanError
@@ -128,8 +129,8 @@ def start_server(args: argparse.Namespace) -> None:
cmd = ["pytest", "-s", str(test_db_api)]
subprocess.run(cmd, check=True)
config.host = args.host
if args.emulate:
urls = list()
# start servers as processes (dlg, ap, c1 and c2 for tests)
for app, port in apps:
proc = mp.Process(
@@ -143,9 +144,7 @@ def start_server(args: argparse.Namespace) -> None:
daemon=True,
)
proc.start()
urls.append(f"http://{args.host}:{port}")
# check server health
for url in urls:
url = f"http://{args.host}:{port}"
res = get_health(url=url + "/health")
if res is None:
raise Exception(f"Couldn't reach {url} after starting server")