diff --git a/pyproject.toml b/pyproject.toml index a092d381f..7337da7d1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ dependencies = [ "pycurl>=7.43", "click>=7.1.2", "rich", + "textual>=0.50.0", # Storage & Local "minio==5.0.10", diff --git a/requirements.txt b/requirements.txt index c3e9e504c..0d3e73985 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,6 +20,7 @@ Pillow pycurl>=7.43 click>=7.1.2 rich +textual>=0.50.0 patch-ng interrogate==1.7.0 diff --git a/sebs/cache.py b/sebs/cache.py index 098aa6eac..92305549a 100644 --- a/sebs/cache.py +++ b/sebs/cache.py @@ -23,6 +23,7 @@ """ import collections.abc +import copy import docker import datetime import json @@ -158,16 +159,25 @@ class Cache(LoggingBase): _lock_registry_guard = threading.Lock() _lock_registry: Dict[str, threading.RLock] = {} - def __init__(self, cache_dir: str, docker_client: docker.DockerClient) -> None: + # Cloud platforms that own a per-cloud cache file (`.json`). + SUPPORTED_CLOUDS: List[str] = ["azure", "aws", "gcp", "openwhisk", "local"] + + def __init__(self, cache_dir: str, docker_client: Optional[docker.DockerClient] = None) -> None: """Initialize the Cache with directory and Docker client. Sets up the cache directory structure and loads existing configurations. Creates the cache directory if it doesn't exist, otherwise loads existing cached configurations. + The Docker client is optional so that read-only consumers (for example, + the cache inspection command) can open a cache without a running Docker + daemon. It is only required for operations that copy container images, + such as `add_code_package`. + Args: cache_dir (str): Path to the cache directory. - docker_client (docker.DockerClient): Docker client for container operations. + docker_client (Optional[docker.DockerClient]): Docker client for + container operations. May be None for read-only usage. """ super().__init__() self.cached_config: Dict[str, Any] = {} @@ -199,6 +209,24 @@ def _cache_dir_lock(cls, cache_dir: str) -> threading.RLock: cls._lock_registry[cache_dir] = threading.RLock() return cls._lock_registry[cache_dir] + def _require_docker_client(self) -> docker.DockerClient: + """Return the Docker client, raising if it is unavailable. + + The Docker client is optional to allow read-only consumers to open a + cache without a running Docker daemon. Operations that copy container + images require it, so they call this helper to fail loudly (and, unlike + an ``assert``, reliably under ``python -O``) when it is missing. + + Returns: + docker.DockerClient: The configured Docker client. + + Raises: + RuntimeError: If no Docker client was provided to the cache. + """ + if self.docker_client is None: + raise RuntimeError("A Docker client is required to cache container images.") + return self.docker_client + @staticmethod def _write_json_atomic(path: str, data: Any) -> None: """Atomically replace a JSON file after fully writing it to a temp file.""" @@ -240,7 +268,7 @@ def load_config(self) -> None: the cache directory and loads them into memory. """ with self._lock: - for cloud in ["azure", "aws", "gcp", "openwhisk", "local"]: + for cloud in self.SUPPORTED_CLOUDS: cloud_config_file = os.path.join(self.cache_dir, "{}.json".format(cloud)) if os.path.exists(cloud_config_file): with open(cloud_config_file, "r") as f: @@ -288,7 +316,7 @@ def shutdown(self) -> None: """ if self.config_updated: with self._lock: - for cloud in ["azure", "aws", "gcp", "openwhisk", "local"]: + for cloud in self.SUPPORTED_CLOUDS: if cloud in self.cached_config: cloud_config_file = os.path.join(self.cache_dir, "{}.json".format(cloud)) self.logging.info("Update cached config {}".format(cloud_config_file)) @@ -498,6 +526,217 @@ def get_all_functions(self, deployment: str) -> Dict[str, Any]: return result + def get_deployed_benchmarks(self, deployment: Optional[str] = None) -> List[Dict[str, Any]]: + """Collect every deployed benchmark entry recorded in the cache. + + Walks all per-benchmark `config.json` files and flattens them into a + list of rows describing what is deployed where. Each row corresponds to + a single (benchmark, platform, language) combination and reports the + deployed functions, packaging type, and cached storage/nosql tables. + + This is a purely read-only view intended for inspection tooling; it does + not query any cloud provider and does not require a Docker client. + + Args: + deployment (Optional[str]): Restrict the result to a single platform + (e.g. 'aws'). When None, entries for all platforms are returned. + + Returns: + List[Dict[str, Any]]: One dictionary per (benchmark, platform, + language) combination. Keys: 'benchmark', 'platform', 'language', + 'packaging', 'functions' (list of function names), + 'function_details' (per-function name/hash/triggers), 'triggers' + (sorted list of trigger types across all functions), 'storage' + (list of bucket names), and 'nosql' (list of table names). + """ + + def summarize_functions(functions: Dict[str, Any]) -> List[Dict[str, Any]]: + """Reduce a cached `functions` mapping to per-function summaries. + + Args: + functions (Dict[str, Any]): Mapping of function name to the + cached function configuration. + + Returns: + List[Dict[str, Any]]: Summaries with 'name', 'hash', and the + sorted list of trigger types for each function. + """ + summaries: List[Dict[str, Any]] = [] + for name, cfg in functions.items(): + triggers = cfg.get("triggers", []) if isinstance(cfg, dict) else [] + trigger_types = sorted( + {str(t["type"]) for t in triggers if isinstance(t, dict) and t.get("type")} + ) + # Preserve per-trigger details (type, URL, implementation) so the + # inspection UI can surface endpoints, not just trigger types. + trigger_details = [ + { + "type": str(t.get("type")), + "url": t.get("url"), + "implementation": t.get("implementation"), + } + for t in triggers + if isinstance(t, dict) and t.get("type") + ] + summaries.append( + { + "name": name, + "hash": cfg.get("hash") if isinstance(cfg, dict) else None, + "triggers": trigger_types, + "trigger_details": trigger_details, + } + ) + return summaries + + rows: List[Dict[str, Any]] = [] + + if not os.path.exists(self.cache_dir): + return rows + + with self._lock: + for entry in sorted(os.listdir(self.cache_dir)): + config_path = os.path.join(self.cache_dir, entry, "config.json") + if not os.path.exists(config_path): + continue + + with open(config_path, "r") as fp: + config = json.load(fp) + + for platform, dep_cfg in config.items(): + if deployment is not None and platform != deployment: + continue + if not isinstance(dep_cfg, dict): + continue + + for language, lang_cfg in dep_cfg.items(): + # Skip resource-level keys (storage/nosql live under the + # deployment too) - languages always map to a dict with + # a 'functions'/'code_package'/'containers' shape. + if not isinstance(lang_cfg, dict): + continue + if language in ("storage", "nosql"): + continue + + functions = lang_cfg.get("functions") or {} + if "containers" in lang_cfg and lang_cfg["containers"]: + packaging = "container" + elif "code_package" in lang_cfg and lang_cfg["code_package"]: + packaging = "code_package" + else: + packaging = "unknown" + + func_summaries = summarize_functions(functions) + all_triggers = sorted({t for f in func_summaries for t in f["triggers"]}) + # Collect every HTTP endpoint URL exposed across the + # functions of this (benchmark, platform, language) entry. + all_urls = sorted( + { + td["url"] + for f in func_summaries + for td in f["trigger_details"] + if td.get("url") + } + ) + + storage_cfg = dep_cfg.get("storage") or {} + nosql_cfg = dep_cfg.get("nosql") or {} + + rows.append( + { + "benchmark": entry, + "platform": platform, + "language": language, + "packaging": packaging, + "functions": [f["name"] for f in func_summaries], + "function_details": func_summaries, + "triggers": all_triggers, + "urls": all_urls, + "storage": sorted(storage_cfg.keys()), + "nosql": sorted(nosql_cfg.keys()), + } + ) + + return rows + + def get_allocated_resources( + self, deployment: Optional[str] = None + ) -> Dict[str, Dict[str, Any]]: + """Collect the allocated cloud resources recorded per platform. + + Reads the per-cloud cache files (`aws.json`, `local.json`, ...) and + extracts the resource block that SeBS persists for reuse: the + `resources_id` namespace, allocated storage buckets, and (for local + deployments) allocated container ports. + + Like `get_deployed_benchmarks`, this is a read-only view that does not + contact any cloud provider. + + Args: + deployment (Optional[str]): Restrict the result to a single platform. + When None, all platforms present in the cache are returned. + + Returns: + Dict[str, Dict[str, Any]]: Mapping of platform name to its resource + summary. Each summary contains 'resources_id' (Optional[str]), + 'region' (Optional[str]), 'storage_buckets' (Dict[str, Any]), + 'allocated_ports' (List[int]), 'resource_group' (Optional[str], + Azure), 'storage_accounts' (List[str], Azure), and 'nosql' + (Dict[str, Any]; Azure CosmosDB account or other NoSQL metadata). + """ + result: Dict[str, Dict[str, Any]] = {} + + clouds = [deployment] if deployment is not None else self.SUPPORTED_CLOUDS + + with self._lock: + for cloud in clouds: + cloud_cfg = self.cached_config.get(cloud) + if not isinstance(cloud_cfg, dict): + continue + + resources = cloud_cfg.get("resources") + if not isinstance(resources, dict): + resources = {} + + buckets = resources.get("storage_buckets") + if not isinstance(buckets, dict): + buckets = {} + + ports = resources.get("allocated_ports") + if not isinstance(ports, list): + ports = [] + + # Azure-specific resource classes: resource group, storage + # accounts, and the CosmosDB (NoSQL) account. + storage_accounts_cfg = resources.get("storage_accounts") + if isinstance(storage_accounts_cfg, list): + storage_accounts = [ + acct.get("account_name") or acct.get("name") + for acct in storage_accounts_cfg + if isinstance(acct, dict) + ] + storage_accounts = [name for name in storage_accounts if name] + else: + storage_accounts = [] + + nosql_cfg = resources.get("cosmosdb_account") + if not isinstance(nosql_cfg, dict): + nosql_cfg = {} + + # Return detached copies so this read-only view can never leak a + # live reference into `cached_config`; a caller mutating the + # result must not be able to corrupt in-memory cache state. + result[cloud] = { + "resources_id": resources.get("resources_id"), + "region": cloud_cfg.get("region"), + "storage_buckets": copy.deepcopy(buckets), + "allocated_ports": list(ports), + "resource_group": resources.get("resource_group"), + "storage_accounts": storage_accounts, + "nosql": copy.deepcopy(nosql_cfg), + } + + return result + def get_storage_config(self, deployment: str, benchmark: str) -> Optional[Dict[str, Any]]: """Access cached storage configuration of a benchmark. @@ -783,6 +1022,11 @@ def add_code_package( RuntimeError: If cached application already exists for the deployment. """ with self._lock: + # Fail before mutating the cache (creating directories, copying + # code) when a container deployment has no Docker client available. + if code_package.system_variant.is_container: + self._require_docker_client() + benchmark_dir = os.path.join(self.cache_dir, code_package.benchmark) os.makedirs(benchmark_dir, exist_ok=True) @@ -829,7 +1073,7 @@ def add_code_package( "functions": {}, } if code_package.system_variant.is_container: - image = self.docker_client.images.get(code_package.container_uri) + image = self._require_docker_client().images.get(code_package.container_uri) language_config["image-uri"] = code_package.container_uri language_config["image-id"] = image.id @@ -891,6 +1135,11 @@ def update_code_package( code_package (Benchmark): The benchmark code package to update. """ with self._lock: + # Fail before mutating the cache (deleting/copying code) when a + # container deployment has no Docker client available. + if code_package.system_variant.is_container: + self._require_docker_client() + benchmark_dir = os.path.join(self.cache_dir, code_package.benchmark) # Check if cache directory for this deployment exist @@ -952,7 +1201,7 @@ def update_code_package( cached_config["size"] = code_package.code_size if code_package.system_variant.is_container: - image = self.docker_client.images.get(code_package.container_uri) + image = self._require_docker_client().images.get(code_package.container_uri) cached_config["image-id"] = image.id cached_config["image-uri"] = code_package.container_uri diff --git a/sebs/cli.py b/sebs/cli.py index ec7bab57e..e8e20ff3f 100755 --- a/sebs/cli.py +++ b/sebs/cli.py @@ -20,6 +20,7 @@ import sebs from sebs import SeBS +from sebs.cache import Cache from sebs.sebs_types import Storage as StorageTypes from sebs.sebs_types import NoSQLStorage as NoSQLStorageTypes from sebs.regression import regression_suite @@ -844,6 +845,70 @@ def resources(): pass +@resources.command("inspect") +@click.option( + "--cache", + default=os.path.join(os.path.curdir, "cache"), + type=click.Path(file_okay=False, readable=True), + help="Location of the experiments cache to inspect.", +) +@click.option( + "--deployment", + default=None, + type=click.Choice(["azure", "aws", "gcp", "local", "openwhisk"]), + help="Restrict the view to a single platform.", +) +@click.option( + "--output", + "output_format", + default="tui", + type=click.Choice(["tui", "json"]), + help="Interactive Textual TUI (default) or machine-readable JSON.", +) +def resources_inspect(cache, deployment, output_format): + """Inspect the local SeBS cache and show what is deployed. + + This is a read-only view built directly from the on-disk cache. It does not + contact any cloud provider and does not require credentials or a running + Docker daemon. It reports which benchmarks are deployed (platform, language, + packaging, functions, triggers, URLs) and what resources are allocated per + platform (resources_id, storage buckets, NoSQL tables, resource groups, and + locally allocated ports). + + By default it launches an interactive Textual TUI that groups resources by + cloud system and resource class. Use ``--output json`` for machine-readable + output that is safe for non-TTY environments and scripting. + """ + # A missing cache directory is a valid, empty state - report it rather than + # failing or creating the directory (constructing Cache would create it). + if not os.path.isdir(cache): + if output_format == "json": + click.echo(json.dumps({"deployed_benchmarks": [], "allocated_resources": {}}, indent=2)) + else: + from rich.console import Console + + Console().print(f"[yellow]No cache found at {cache}.[/yellow]") + return + + cache_client = Cache(cache) + + benchmarks = cache_client.get_deployed_benchmarks(deployment) + allocated = cache_client.get_allocated_resources(deployment) + + if output_format == "json": + click.echo( + json.dumps( + {"deployed_benchmarks": benchmarks, "allocated_resources": allocated}, + indent=2, + ) + ) + return + + from sebs.tui import run_inspector + + run_inspector(benchmarks, allocated) + + @resources.command("list") @click.argument("resource", type=click.Choice(["buckets", "resource-groups"])) @common_params diff --git a/sebs/tui/__init__.py b/sebs/tui/__init__.py new file mode 100644 index 000000000..1e5ec0dae --- /dev/null +++ b/sebs/tui/__init__.py @@ -0,0 +1,9 @@ +# Copyright 2020-2025 ETH Zurich and the SeBS authors. All rights reserved. +"""Interactive terminal user interfaces for SeBS. + +This package hosts Textual-based TUIs used by the CLI. The first one is the +read-only cache inspector (``sebs resources inspect``), which visualises the +on-disk cache grouped by cloud system and resource class. +""" + +from .inspect import ResourceInspectorApp, run_inspector # noqa: F401 diff --git a/sebs/tui/inspect.py b/sebs/tui/inspect.py new file mode 100644 index 000000000..ebbf43263 --- /dev/null +++ b/sebs/tui/inspect.py @@ -0,0 +1,379 @@ +# Copyright 2020-2025 ETH Zurich and the SeBS authors. All rights reserved. +"""Interactive Textual TUI for read-only inspection of the SeBS cache. + +The inspector renders the on-disk cache as an interactive tree grouped by cloud +system and, within each cloud, by resource class (benchmarks, functions, +triggers and URLs, storage buckets, NoSQL tables, resource groups, and locally +allocated ports). Selecting a node in the tree updates a detail table on the +right, so complex experiments with many allocated resources stay readable +instead of collapsing into one busy printout. + +The heavy widget logic lives in :class:`ResourceInspectorApp`, but the grouping +itself is performed by the pure :func:`build_inventory` helper so it can be unit +tested without launching a terminal application. +""" + +from typing import Any, Dict, List, Optional + +from textual.app import App, ComposeResult +from textual.containers import Horizontal +from textual.widgets import DataTable, Footer, Header, Static, Tree +from textual.widgets.tree import TreeNode + + +# The stable order in which resource classes are shown under each cloud. +RESOURCE_CLASSES: List[str] = [ + "Benchmarks", + "Functions", + "Triggers & URLs", + "Storage buckets", + "NoSQL tables", + "Resource groups", + "Allocated ports", +] + + +def _clouds_in_use( + benchmarks: List[Dict[str, Any]], allocated: Dict[str, Dict[str, Any]] +) -> List[str]: + """Return the sorted set of clouds referenced by benchmarks or resources. + + Args: + benchmarks: Rows from :meth:`sebs.cache.Cache.get_deployed_benchmarks`. + allocated: Mapping from :meth:`sebs.cache.Cache.get_allocated_resources`. + + Returns: + Sorted list of platform names that appear in either input. + """ + clouds = {row["platform"] for row in benchmarks} + clouds.update(allocated.keys()) + return sorted(clouds) + + +def build_inventory( + benchmarks: List[Dict[str, Any]], allocated: Dict[str, Dict[str, Any]] +) -> Dict[str, Dict[str, Any]]: + """Group flat cache rows into a per-cloud, per-resource-class inventory. + + This is the pure data transform that backs the TUI. It reshapes the two flat + cache views into a nested structure keyed first by cloud and then by resource + class, attaching a small summary (resources id and region) per cloud. + + Args: + benchmarks: Rows from :meth:`sebs.cache.Cache.get_deployed_benchmarks`. + allocated: Mapping from :meth:`sebs.cache.Cache.get_allocated_resources`. + + Returns: + Mapping of cloud name to a dict with two keys: ``summary`` (a dict with + ``resources_id`` and ``region``) and ``classes`` (a mapping of resource + class name to a list of row dicts describing each entry). + """ + inventory: Dict[str, Dict[str, Any]] = {} + + for cloud in _clouds_in_use(benchmarks, allocated): + cloud_benchmarks = [row for row in benchmarks if row["platform"] == cloud] + resources = allocated.get(cloud, {}) + + classes: Dict[str, List[Dict[str, Any]]] = {name: [] for name in RESOURCE_CLASSES} + + for row in cloud_benchmarks: + classes["Benchmarks"].append( + { + "benchmark": row["benchmark"], + "language": row["language"], + "packaging": row["packaging"], + "functions": len(row.get("functions", [])), + "triggers": ", ".join(row.get("triggers", [])) or "-", + } + ) + for func in row.get("function_details", []): + classes["Functions"].append( + { + "name": func["name"], + "benchmark": row["benchmark"], + "language": row["language"], + "hash": func.get("hash") or "-", + "triggers": ", ".join(func.get("triggers", [])) or "-", + } + ) + for trigger in func.get("trigger_details", []): + classes["Triggers & URLs"].append( + { + "benchmark": row["benchmark"], + "function": func["name"], + "type": trigger.get("type") or "-", + "implementation": trigger.get("implementation") or "-", + "url": trigger.get("url") or "-", + } + ) + + # Storage buckets: allocated buckets plus any referenced per benchmark. + seen_buckets = set() + for name, value in (resources.get("storage_buckets") or {}).items(): + key = str(value) if value else name + if key in seen_buckets: + continue + seen_buckets.add(key) + classes["Storage buckets"].append({"name": key, "role": name}) + for account in resources.get("storage_accounts") or []: + if account in seen_buckets: + continue + seen_buckets.add(account) + classes["Storage buckets"].append({"name": account, "role": "storage_account"}) + for row in cloud_benchmarks: + for bucket in row.get("storage", []): + if bucket in seen_buckets: + continue + seen_buckets.add(bucket) + classes["Storage buckets"].append({"name": bucket, "role": "benchmark"}) + + # NoSQL tables: per-benchmark tables plus a cloud-level NoSQL account. + seen_tables = set() + for row in cloud_benchmarks: + for table in row.get("nosql", []): + if table in seen_tables: + continue + seen_tables.add(table) + classes["NoSQL tables"].append({"name": table, "benchmark": row["benchmark"]}) + nosql_account = resources.get("nosql") or {} + account_name = nosql_account.get("account_name") or nosql_account.get("name") + if account_name and account_name not in seen_tables: + seen_tables.add(account_name) + classes["NoSQL tables"].append({"name": account_name, "benchmark": "(account)"}) + + resource_group = resources.get("resource_group") + if resource_group: + classes["Resource groups"].append({"name": resource_group}) + + for port in resources.get("allocated_ports") or []: + classes["Allocated ports"].append({"port": str(port)}) + + inventory[cloud] = { + "summary": { + "resources_id": resources.get("resources_id"), + "region": resources.get("region"), + }, + "classes": classes, + } + + return inventory + + +# Column layout for the detail table, keyed by resource class name. +_CLASS_COLUMNS: Dict[str, List[str]] = { + "Benchmarks": ["benchmark", "language", "packaging", "functions", "triggers"], + "Functions": ["name", "benchmark", "language", "hash", "triggers"], + "Triggers & URLs": ["benchmark", "function", "type", "implementation", "url"], + "Storage buckets": ["name", "role"], + "NoSQL tables": ["name", "benchmark"], + "Resource groups": ["name"], + "Allocated ports": ["port"], +} + + +class ResourceInspectorApp(App): + """Textual application that visualises the SeBS cache interactively. + + The left pane is a tree grouping resources by cloud system and resource + class; the right pane is a detail table that reflects the highlighted node. + """ + + CSS = """ + Tree { + width: 40%; + border: round $primary; + } + #detail { + width: 60%; + } + #summary { + height: auto; + padding: 0 1; + color: $text-muted; + } + DataTable { + height: 1fr; + } + """ + + BINDINGS = [ + ("q", "quit", "Quit"), + ("e", "expand_all", "Expand all"), + ("c", "collapse_all", "Collapse all"), + ] + + def __init__( + self, + benchmarks: List[Dict[str, Any]], + allocated: Dict[str, Dict[str, Any]], + ) -> None: + """Store the cache views and precompute the grouped inventory. + + Args: + benchmarks: Rows from + :meth:`sebs.cache.Cache.get_deployed_benchmarks`. + allocated: Mapping from + :meth:`sebs.cache.Cache.get_allocated_resources`. + """ + super().__init__() + self._benchmarks = benchmarks + self._allocated = allocated + self._inventory = build_inventory(benchmarks, allocated) + + def compose(self) -> ComposeResult: + """Build the widget hierarchy for the application. + + Yields: + The header, the tree/detail split, and the footer widgets. + """ + yield Header(show_clock=False) + with Horizontal(): + yield Tree("SeBS cache", id="tree") + with Horizontal(id="detail"): + yield DataTable(id="table", zebra_stripes=True, cursor_type="row") + yield Static("", id="summary") + yield Footer() + + def on_mount(self) -> None: + """Populate the tree once the DOM is ready and focus it.""" + self.title = "SeBS Resource Inspector" + tree = self.query_one("#tree", Tree) + tree.root.data = {"kind": "root"} + tree.root.expand() + + if not self._inventory: + tree.root.add_leaf("No cached resources found", data={"kind": "empty"}) + self.query_one("#summary", Static).update( + "Cache is empty or contains no deployed benchmarks or allocated resources." + ) + return + + for cloud, entry in self._inventory.items(): + cloud_node = tree.root.add(cloud, data={"kind": "cloud", "cloud": cloud}) + cloud_node.expand() + for class_name in RESOURCE_CLASSES: + rows = entry["classes"].get(class_name, []) + label = f"{class_name} ({len(rows)})" + class_node = cloud_node.add( + label, + data={"kind": "class", "cloud": cloud, "class": class_name, "rows": rows}, + ) + for row in rows: + class_node.add_leaf( + self._leaf_label(class_name, row), + data={ + "kind": "item", + "cloud": cloud, + "class": class_name, + "rows": [row], + }, + ) + tree.focus() + + @staticmethod + def _leaf_label(class_name: str, row: Dict[str, Any]) -> str: + """Return a short label for a single resource entry. + + Args: + class_name: Resource class the row belongs to. + row: The row dict describing the entry. + + Returns: + A concise, human-readable label for the tree leaf. + """ + if class_name == "Benchmarks": + return f"{row['benchmark']} [{row['language']}]" + if class_name == "Functions": + return row["name"] + if class_name == "Triggers & URLs": + return f"{row['type']}: {row['url']}" + if class_name == "Allocated ports": + return row["port"] + return str(row.get("name", next(iter(row.values()), "-"))) + + def _render_rows(self, class_name: str, rows: List[Dict[str, Any]]) -> None: + """Render a set of resource rows into the detail table. + + Args: + class_name: Resource class whose column layout should be used. + rows: Row dicts to display. + """ + table = self.query_one("#table", DataTable) + table.clear(columns=True) + columns = _CLASS_COLUMNS.get(class_name, []) + if not columns: + return + table.add_columns(*[col.replace("_", " ").title() for col in columns]) + for row in rows: + table.add_row(*[str(row.get(col, "-")) for col in columns]) + + def _show_summary(self, cloud: Optional[str]) -> None: + """Update the summary line for the selected cloud. + + Args: + cloud: Cloud whose summary to show, or None to clear it. + """ + summary = self.query_one("#summary", Static) + if cloud is None or cloud not in self._inventory: + summary.update("") + return + info = self._inventory[cloud]["summary"] + rid = info.get("resources_id") or "-" + region = info.get("region") or "-" + summary.update(f"{cloud} resources_id: {rid} region: {region}") + + def on_tree_node_highlighted(self, event: Tree.NodeHighlighted) -> None: + """React to tree navigation by updating the detail table and summary. + + Args: + event: The highlight event carrying the focused tree node. + """ + self._update_from_node(event.node) + + def on_tree_node_selected(self, event: Tree.NodeSelected) -> None: + """React to explicit node selection identically to highlighting. + + Args: + event: The selection event carrying the chosen tree node. + """ + self._update_from_node(event.node) + + def _update_from_node(self, node: TreeNode) -> None: + """Refresh the detail panels from the given tree node's payload. + + Args: + node: The tree node whose attached data drives the detail view. + """ + data = node.data or {} + kind = data.get("kind") + cloud = data.get("cloud") + self._show_summary(cloud) + + if kind in ("class", "item"): + self._render_rows(data["class"], data.get("rows", [])) + else: + table = self.query_one("#table", DataTable) + table.clear(columns=True) + + def action_expand_all(self) -> None: + """Expand every node in the tree.""" + self.query_one("#tree", Tree).root.expand_all() + + def action_collapse_all(self) -> None: + """Collapse every node back to the cloud level.""" + tree = self.query_one("#tree", Tree) + for cloud_node in tree.root.children: + cloud_node.collapse_all() + cloud_node.expand() + + +def run_inspector(benchmarks: List[Dict[str, Any]], allocated: Dict[str, Dict[str, Any]]) -> None: + """Launch the interactive resource inspector. + + Args: + benchmarks: Rows from + :meth:`sebs.cache.Cache.get_deployed_benchmarks`. + allocated: Mapping from + :meth:`sebs.cache.Cache.get_allocated_resources`. + """ + ResourceInspectorApp(benchmarks, allocated).run() diff --git a/tests/test_cache_inspect.py b/tests/test_cache_inspect.py new file mode 100644 index 000000000..74096e824 --- /dev/null +++ b/tests/test_cache_inspect.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +# Copyright 2020-2025 ETH Zurich and the SeBS authors. All rights reserved. +"""Unit tests for the read-only cache inspection API. + +These tests build a synthetic cache directory on disk (per-cloud `.json` +files plus per-benchmark `config.json` files) and verify that +`Cache.get_deployed_benchmarks` and `Cache.get_allocated_resources` flatten the +model correctly. They run fully offline and require no Docker daemon. +""" + +import json +import os +import tempfile +import unittest + +from sebs.cache import Cache + + +def _write_json(path: str, data: dict) -> None: + """Write a dictionary as JSON to a path, creating parent directories. + + Args: + path: Destination file path. + data: JSON-serializable dictionary to write. + """ + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as fp: + json.dump(data, fp) + + +class CacheInspectTests(unittest.TestCase): + """Verify the read-only inspection helpers on a synthetic cache.""" + + def setUp(self) -> None: + """Create a temporary cache directory with representative content.""" + self._tmp = tempfile.TemporaryDirectory() + self.cache_dir = self._tmp.name + + # Per-cloud resource files. + _write_json( + os.path.join(self.cache_dir, "aws.json"), + { + "region": "us-east-1", + "resources": { + "resources_id": "abc123", + "storage_buckets": { + "benchmarks": "sebs-benchmarks-abc123", + "experiments": "sebs-experiments-abc123", + }, + }, + }, + ) + _write_json( + os.path.join(self.cache_dir, "azure.json"), + { + "region": "westeurope", + "resources": { + "resources_id": "az777", + "resource_group": "sebs_resource_group_az777", + "storage_accounts": [{"account_name": "sebsstorageaz777"}], + "cosmosdb_account": {"account_name": "sebs-cosmos-az777"}, + }, + }, + ) + _write_json( + os.path.join(self.cache_dir, "local.json"), + { + "resources": { + "resources_id": "local42", + "allocated_ports": [9000, 9001], + }, + }, + ) + + # Per-benchmark config with a deployed AWS python function. + _write_json( + os.path.join(self.cache_dir, "110.dynamic-html", "config.json"), + { + "aws": { + "python": { + "code_package": {"3.9": {"x64": {"location": "code"}}}, + "containers": {}, + "functions": { + "sebs-abc123-110.dynamic-html-python-3.9-x64": { + "name": "sebs-abc123-110.dynamic-html-python-3.9-x64", + "hash": "deadbeef", + "triggers": [ + {"type": "library"}, + { + "type": "http", + "url": "https://abc123.execute-api." + "us-east-1.amazonaws.com/inspect", + "implementation": "api_gateway", + }, + ], + } + }, + }, + "storage": {"sebs-benchmarks-abc123": {}}, + "nosql": {"sebs-table-abc123": {}}, + } + }, + ) + + def tearDown(self) -> None: + """Remove the temporary cache directory.""" + self._tmp.cleanup() + + def test_deployed_benchmarks(self) -> None: + """Deployed benchmark rows expose platform, packaging, and triggers.""" + cache = Cache(self.cache_dir) + rows = cache.get_deployed_benchmarks() + + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(row["benchmark"], "110.dynamic-html") + self.assertEqual(row["platform"], "aws") + self.assertEqual(row["language"], "python") + self.assertEqual(row["packaging"], "code_package") + self.assertEqual(len(row["functions"]), 1) + self.assertEqual(row["triggers"], ["http", "library"]) + self.assertEqual(row["storage"], ["sebs-benchmarks-abc123"]) + self.assertEqual(row["nosql"], ["sebs-table-abc123"]) + # The HTTP trigger URL is surfaced both per-function and aggregated. + self.assertEqual( + row["urls"], + ["https://abc123.execute-api.us-east-1.amazonaws.com/inspect"], + ) + func = row["function_details"][0] + http = [t for t in func["trigger_details"] if t["type"] == "http"][0] + self.assertEqual( + http["url"], + "https://abc123.execute-api.us-east-1.amazonaws.com/inspect", + ) + self.assertEqual(http["implementation"], "api_gateway") + + def test_deployed_benchmarks_filter(self) -> None: + """Filtering by a platform with no entries yields an empty list.""" + cache = Cache(self.cache_dir) + self.assertEqual(cache.get_deployed_benchmarks("gcp"), []) + + def test_allocated_resources(self) -> None: + """Allocated resources expose ids, buckets, region, and ports.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources() + + self.assertEqual(allocated["aws"]["resources_id"], "abc123") + self.assertEqual(allocated["aws"]["region"], "us-east-1") + self.assertEqual(len(allocated["aws"]["storage_buckets"]), 2) + self.assertEqual(allocated["aws"]["allocated_ports"], []) + + self.assertEqual(allocated["local"]["resources_id"], "local42") + self.assertEqual(allocated["local"]["allocated_ports"], [9000, 9001]) + + def test_allocated_resources_filter(self) -> None: + """Filtering allocated resources returns only the requested platform.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources("aws") + self.assertEqual(list(allocated.keys()), ["aws"]) + + def test_allocated_resources_ports_only(self) -> None: + """A local entry with only allocated ports is still reported.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources("local") + local = allocated["local"] + # local42 has no buckets, only ports - it must remain visible. + self.assertEqual(local["storage_buckets"], {}) + self.assertEqual(local["allocated_ports"], [9000, 9001]) + + def test_allocated_resources_azure_classes(self) -> None: + """Azure resource classes (group, accounts, NoSQL) are surfaced.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources("azure") + azure = allocated["azure"] + self.assertEqual(azure["resources_id"], "az777") + self.assertEqual(azure["region"], "westeurope") + self.assertEqual(azure["resource_group"], "sebs_resource_group_az777") + self.assertEqual(azure["storage_accounts"], ["sebsstorageaz777"]) + self.assertEqual(azure["nosql"], {"account_name": "sebs-cosmos-az777"}) + + def test_inspection_is_read_only(self) -> None: + """Reading and mutating the results never touches cache state.""" + cache = Cache(self.cache_dir) + self.assertFalse(cache.config_updated) + + allocated = cache.get_allocated_resources("aws") + cache.get_deployed_benchmarks("aws") + # Reads must not flag the cache as dirty (which would trigger a + # write-back on shutdown()). + self.assertFalse(cache.config_updated) + + # The returned structure must be detached from cached_config: mutating + # it must not leak back into the in-memory cache. + allocated["aws"]["storage_buckets"]["INJECTED"] = "should-not-leak" + allocated["aws"]["allocated_ports"].append(65535) + live = cache.cached_config["aws"]["resources"] + self.assertNotIn("INJECTED", live.get("storage_buckets", {})) + self.assertNotIn(65535, live.get("allocated_ports", [])) + self.assertFalse(cache.config_updated) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tui_inspect.py b/tests/test_tui_inspect.py new file mode 100644 index 000000000..8b1faa517 --- /dev/null +++ b/tests/test_tui_inspect.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +# Copyright 2020-2025 ETH Zurich and the SeBS authors. All rights reserved. +"""Unit tests for the interactive cache-inspection TUI. + +These tests exercise the pure grouping helper (:func:`build_inventory`) directly +and drive the Textual application headlessly through its test ``Pilot`` so the +widget wiring is covered without a real terminal. They run fully offline. +""" + +import unittest + +from sebs.tui.inspect import RESOURCE_CLASSES, ResourceInspectorApp, build_inventory + + +def _sample_inputs(): + """Return representative benchmark and allocated-resource fixtures. + + Returns: + Tuple of (benchmarks, allocated) matching the shape produced by + ``Cache.get_deployed_benchmarks`` and ``Cache.get_allocated_resources``. + """ + benchmarks = [ + { + "benchmark": "110.dynamic-html", + "platform": "aws", + "language": "python", + "packaging": "code_package", + "functions": ["fn-1"], + "function_details": [ + { + "name": "fn-1", + "hash": "deadbeef", + "triggers": ["http", "library"], + "trigger_details": [ + {"type": "library", "url": None, "implementation": None}, + { + "type": "http", + "url": "https://example.execute-api.amazonaws.com/x", + "implementation": "api_gateway", + }, + ], + } + ], + "triggers": ["http", "library"], + "urls": ["https://example.execute-api.amazonaws.com/x"], + "storage": ["sebs-benchmarks-abc"], + "nosql": ["sebs-table-abc"], + } + ] + allocated = { + "aws": { + "resources_id": "abc123", + "region": "us-east-1", + "storage_buckets": {"benchmarks": "sebs-benchmarks-abc"}, + "allocated_ports": [], + "resource_group": None, + "storage_accounts": [], + "nosql": {}, + }, + "local": { + "resources_id": "local42", + "region": None, + "storage_buckets": {}, + "allocated_ports": [9000, 9001], + "resource_group": None, + "storage_accounts": [], + "nosql": {}, + }, + } + return benchmarks, allocated + + +class BuildInventoryTests(unittest.TestCase): + """Verify the pure grouping transform behind the TUI.""" + + def test_groups_by_cloud_and_class(self) -> None: + """Every cloud is present and keyed by the known resource classes.""" + benchmarks, allocated = _sample_inputs() + inventory = build_inventory(benchmarks, allocated) + + self.assertEqual(sorted(inventory.keys()), ["aws", "local"]) + for cloud in inventory.values(): + self.assertEqual(list(cloud["classes"].keys()), RESOURCE_CLASSES) + + def test_aws_classes_populated(self) -> None: + """AWS groups its benchmark, function, trigger, storage, and NoSQL rows.""" + benchmarks, allocated = _sample_inputs() + classes = build_inventory(benchmarks, allocated)["aws"]["classes"] + + self.assertEqual(len(classes["Benchmarks"]), 1) + self.assertEqual(len(classes["Functions"]), 1) + triggers = classes["Triggers & URLs"] + self.assertEqual(len(triggers), 2) + http = [t for t in triggers if t["type"] == "http"][0] + self.assertEqual(http["url"], "https://example.execute-api.amazonaws.com/x") + self.assertEqual([b["name"] for b in classes["Storage buckets"]], ["sebs-benchmarks-abc"]) + self.assertEqual([t["name"] for t in classes["NoSQL tables"]], ["sebs-table-abc"]) + + def test_local_ports_grouped(self) -> None: + """Local allocated ports become individual rows under their class.""" + benchmarks, allocated = _sample_inputs() + classes = build_inventory(benchmarks, allocated)["local"]["classes"] + self.assertEqual([p["port"] for p in classes["Allocated ports"]], ["9000", "9001"]) + + def test_empty_inputs(self) -> None: + """Empty cache inputs yield an empty inventory.""" + self.assertEqual(build_inventory([], {}), {}) + + +class ResourceInspectorAppTests(unittest.IsolatedAsyncioTestCase): + """Drive the Textual application headlessly through its test Pilot.""" + + async def test_tree_populates_and_selection_fills_table(self) -> None: + """Mounting builds the cloud tree and selecting a class fills the table.""" + benchmarks, allocated = _sample_inputs() + app = ResourceInspectorApp(benchmarks, allocated) + async with app.run_test() as pilot: + from textual.widgets import DataTable, Tree + + tree = app.query_one("#tree", Tree) + cloud_labels = sorted(str(node.label) for node in tree.root.children) + self.assertEqual(cloud_labels, ["aws", "local"]) + + # Find the AWS "Benchmarks" class node and select it. + aws_node = [n for n in tree.root.children if str(n.label) == "aws"][0] + bench_node = [ + n for n in aws_node.children if str(n.label).startswith("Benchmarks") + ][0] + tree.select_node(bench_node) + await pilot.pause() + + table = app.query_one("#table", DataTable) + self.assertGreater(table.row_count, 0) + + async def test_empty_cache_shows_message(self) -> None: + """An empty cache renders a placeholder leaf instead of cloud nodes.""" + app = ResourceInspectorApp([], {}) + async with app.run_test() as pilot: + await pilot.pause() + from textual.widgets import Tree + + tree = app.query_one("#tree", Tree) + labels = [str(node.label) for node in tree.root.children] + self.assertEqual(len(labels), 1) + self.assertIn("No cached resources", labels[0]) + + +if __name__ == "__main__": + unittest.main()