Add runtime-checker to program cmds

Author: philogicae

src/aleph_client/commands/program.py

@@ -2,12 +2,14 @@
 
 import json
 import logging
+import re
 from base64 import b16decode, b32encode
 from collections.abc import Mapping
 from pathlib import Path
 from typing import List, Optional, cast
 from zipfile import BadZipFile
 
+import aiohttp
 import typer
 from aleph.sdk import AlephHttpClient, AuthenticatedAlephHttpClient
 from aleph.sdk.account import _load_account
@@ -171,14 +173,14 @@ async def upload(
         hash_base32 = b32encode(b16decode(item_hash.upper())).strip(b"=").lower().decode()
 
         if verbose:
-        typer.echo(
-            f"Your program has been uploaded on aleph.im\n\n"
-            "Available on:\n"
-            f"  {settings.VM_URL_PATH.format(hash=item_hash)}\n"
-            f"  {settings.VM_URL_HOST.format(hash_base32=hash_base32)}\n"
-            "Visualise on:\n  https://explorer.aleph.im/address/"
-            f"{message.chain.value}/{message.sender}/message/PROGRAM/{item_hash}\n"
-        )
+            typer.echo(
+                f"Your program has been uploaded on aleph.im\n\n"
+                "Available on:\n"
+                f"  {settings.VM_URL_PATH.format(hash=item_hash)}\n"
+                f"  {settings.VM_URL_HOST.format(hash_base32=hash_base32)}\n"
+                "Visualise on:\n  https://explorer.aleph.im/address/"
+                f"{message.chain.value}/{message.sender}/message/PROGRAM/{item_hash}\n"
+            )
         return item_hash
 
 
@@ -286,11 +288,11 @@ async def delete(
                 hashes=[ItemHash(code_volume.item_hash)], reason=f"Deletion of program {item_hash}"
             )
             if verbose:
-            typer.echo(f"Code volume {code_volume.item_hash} has been deleted.")
+                typer.echo(f"Code volume {code_volume.item_hash} has been deleted.")
         if print_message:
             typer.echo(f"{message.json(indent=4)}")
         if verbose:
-        typer.echo(f"Program {item_hash} has been deleted.")
+            typer.echo(f"Program {item_hash} has been deleted.")
 
 
 @app.command(name="list")
@@ -469,3 +471,80 @@ async def logs(
             log_entries = await response.json()
             for log in log_entries:
                 echo(f'{log["__REALTIME_TIMESTAMP"]}>  {log["MESSAGE"]}')
+
+
+@app.command()
+async def runtime_checker(
+    item_hash: str = typer.Argument(..., help="Item hash of the runtime to check"),
+    private_key: Optional[str] = settings.PRIVATE_KEY_STRING,
+    private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE,
+    verbose: bool = False,
+    debug: bool = False,
+):
+    """Check versions used by a runtime (distribution, python, nodejs, etc)"""
+
+    setup_logging(debug)
+
+    echo("Deploy runtime checker program...")
+    try:
+        program_hash = await upload(
+            path=Path(__file__).resolve().parent / "program_utils/runtime_checker.squashfs",
+            entrypoint="main:app",
+            channel=settings.DEFAULT_CHANNEL,
+            memory=settings.DEFAULT_VM_MEMORY,
+            vcpus=settings.DEFAULT_VM_VCPUS,
+            timeout_seconds=settings.DEFAULT_VM_TIMEOUT,
+            name="runtime_checker",
+            runtime=item_hash,
+            beta=False,
+            persistent=False,
+            skip_volume=True,
+            private_key=private_key,
+            private_key_file=private_key_file,
+            print_messages=False,
+            print_code_message=False,
+            print_program_message=False,
+            verbose=verbose,
+            debug=debug,
+        )
+    except Exception as e:
+        echo(f"Failed to deploy the runtime checker program: {e}")
+        raise typer.Exit(code=1)
+
+    program_url = settings.VM_URL_PATH.format(hash=program_hash)
+    versions: dict
+    echo("Query runtime checker to retrieve versions...")
+    try:
+        timeout = aiohttp.ClientTimeout(total=settings.HTTP_REQUEST_TIMEOUT)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.get(program_url) as resp:
+                resp.raise_for_status()
+                versions = await resp.json()
+    except Exception as e:
+        logger.debug(f"Unexpected error when calling {program_url}: {e}")
+        raise typer.Exit(code=1)
+
+    echo("Delete runtime checker...")
+    try:
+        await delete(
+            item_hash=program_hash,
+            reason="Automatic deletion of the runtime checker program",
+            delete_code=True,
+            private_key=private_key,
+            private_key_file=private_key_file,
+            print_message=False,
+            verbose=verbose,
+            debug=debug,
+        )
+    except Exception as e:
+        echo(f"Failed to delete the runtime checker program: {e}")
+        raise typer.Exit(code=1)
+
+    console = Console()
+    infos = [Text.from_markup(f"[bold]Ref:[/bold] [bright_cyan]{item_hash}[/bright_cyan]")]
+    for label, version in versions.items():
+        color = "green" if bool(re.search(r"\d", version)) else "red"
+        infos.append(Text.from_markup(f"\n[bold]{label}:[/bold] [{color}]{version}[/{color}]"))
+    console.print(
+        Panel(Text.assemble(*infos), title="Runtime Infos", border_style="violet", expand=False, title_align="left")
+    )

src/aleph_client/commands/program_utils/runtime_checker.squashfs

@@ -0,0 +1,36 @@
+import platform
+import subprocess
+from typing import Dict
+
+from fastapi import FastAPI
+
+app = FastAPI()
+
+extra_checks = dict(
+    Nodejs="node --version",
+    Rust="rustc --version",
+    Go="go version",
+)
+
+
+@app.get("/")
+async def versions() -> Dict[str, str]:
+    results = dict()
+
+    # Distribution
+    try:
+        results["Distribution"] = platform.freedesktop_os_release()["PRETTY_NAME"]  # type: ignore
+    except Exception:
+        results["Distribution"] = "Not available"
+
+    # Python
+    results["Python"] = platform.python_version()
+
+    # Others
+    for label, command in extra_checks.items():
+        try:
+            results[label] = subprocess.check_output(command.split(" ")).decode("utf-8").strip()
+        except Exception:
+            results[label] = "Not installed"
+
+    return results