Show percent progress for fastcdn offline and add unittest for it

fastapi_cdn_host/cli.py

@@ -2,15 +2,15 @@
 import os
 import shlex
 import subprocess
-from contextlib import contextmanager
+from contextlib import asynccontextmanager, contextmanager
 from datetime import datetime
 from pathlib import Path
-from typing import Generator, Union
+from typing import AsyncGenerator, Generator, Union
 
 import anyio
 import typer
 from rich import print
-from rich.progress import Progress, SpinnerColumn, TextColumn
+from rich.progress import Progress, SpinnerColumn
 from typing_extensions import Annotated
 
 from .client import CdnHostBuilder, HttpSniff
@@ -72,12 +72,46 @@ def patch_app(path: Union[str, Path], remove=True) -> Generator[Path, None, None
             print(f"Auto remove temp file: {app_file}")
 
 
+@asynccontextmanager
+async def percentbar(
+    msg: str, seconds=5, color: str = "cyan", transient=False
+) -> AsyncGenerator[None, None]:
+    """Progressbar with custom font color
+
+    :param msg: prompt message.
+    :param seconds: max seconds of tasks.
+    :param color: font color，e.g.: 'blue'.
+    :param transient: whether clean progressbar after finished.
+    """
+    total = seconds * 100
+
+    async def play(progress, task, expected=1 / 2, thod=0.8) -> None:
+        cost = seconds * expected
+        quick = int(total * thod)
+        delay = cost / quick
+        for i in range(quick):
+            await anyio.sleep(delay)
+            progress.advance(task)
+        cost = seconds - cost
+        slow = total - quick
+        delay = cost / slow
+        for i in range(slow):
+            await anyio.sleep(delay)
+            progress.advance(task)
+
+    with Progress(transient=transient) as progress:
+        task = progress.add_task(f"[{color}]{msg}:", total=total)
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(play, progress, task)
+            yield
+            tg.cancel_scope.cancel()
+            progress.update(task, completed=total)
+
+
 @contextmanager
-def progressbar(msg, color="cyan", transient=True) -> Generator[None, None, None]:
+def spinnerbar(msg, color="yellow", transient=True) -> Generator[None, None, None]:
     with Progress(
-        SpinnerColumn(),
-        TextColumn("[progress.description]{task.description}"),
-        transient=transient,
+        SpinnerColumn(), *Progress.get_default_columns(), transient=transient
     ) as progress:
         progress.add_task(f"[{color}]{msg}...", total=None)
         yield
@@ -94,10 +128,10 @@ async def download_offline_assets(dirname="static") -> None:
             relative_path = p.relative_to(cwd)
             print(f"{relative_path} already exists. abort!")
             return
-    with progressbar("Comparing cdn host response speed", transient=False):
+    async with percentbar("Comparing cdn hosts response speed"):
         urls = await CdnHostBuilder.sniff_the_fastest()
     print("Result:", urls)
-    with progressbar("Fetching files from cdn", color="blue"):
+    with spinnerbar("Fetching files from cdn"):
         url_list = [urls.js, urls.css, urls.redoc]
         contents = await HttpSniff.bulk_fetch(url_list, get_content=True)
         for url, content in zip(url_list, contents):

tests/fastcdn_offline/conftest.py

@@ -0,0 +1,22 @@
+import os
+import shutil
+from pathlib import Path
+
+import pytest
+
+
+def ensure_not_exist(path: Path) -> None:
+    if path.exists():
+        shutil.rmtree(path)
+        print(f"{path} removed.")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def prepare():
+    static_root = Path(__file__).parent / "static"
+    ensure_not_exist(static_root)
+    rc = os.system("fastcdn offline")
+    assert rc == 0
+    assert static_root.exists()
+    yield
+    ensure_not_exist(static_root)

tests/fastcdn_offline/main.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+
+from fastapi import FastAPI, Request
+from fastapi.responses import RedirectResponse
+
+import fastapi_cdn_host
+
+app = FastAPI(title="FastAPI CDN host test")
+
+
+@app.get("/", include_in_schema=False)
+async def to_docs():
+    return RedirectResponse("/docs")
+
+
+@app.get("/app")
+async def get_app(request: Request) -> dict:
+    return {"routes": str(request.app.routes)}
+
+
+fastapi_cdn_host.patch_docs(app)

tests/fastcdn_offline/test_fastcdn_offline.py

@@ -0,0 +1,39 @@
+# mypy: no-disallow-untyped-decorators
+import pytest
+from httpx import AsyncClient
+
+from fastapi_cdn_host.client import CdnHostBuilder
+from fastapi_cdn_host.utils import TestClient
+
+default_favicon_url = "https://fastapi.tiangolo.com/img/favicon.png"
+
+
+@pytest.fixture(scope="module")
+async def client():
+    from main import app
+
+    async with TestClient(app) as c:
+        yield c
+
+
+@pytest.mark.anyio
+async def test_docs(client: AsyncClient):  # nosec
+    swagger_ui = CdnHostBuilder.swagger_files
+    js_url = "/static/" + swagger_ui["js"]
+    css_url = "/static/" + swagger_ui["css"]
+    response = await client.get(css_url)
+    assert response.status_code == 200, f"{response.url=};{response.text=}"
+    response = await client.get(js_url)
+    assert response.status_code == 200, response.text
+    response = await client.get("/docs")
+    text = response.text
+    assert response.status_code == 200, text
+    assert default_favicon_url in text
+    assert js_url in text
+    assert css_url in text
+    response = await client.get("/redoc")
+    text = response.text
+    assert response.status_code == 200, text
+    assert "/static/redoc" in text
+    response = await client.get("/app")
+    assert response.status_code == 200