From a4c5f7b3de3ad289e1fd46c16dc809b44baf08ec Mon Sep 17 00:00:00 2001 From: pyxl Date: Fri, 13 Mar 2026 23:58:11 +0100 Subject: [PATCH] feat: add AdGuardHome CLI harness (cli-anything-adguardhome) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First REST HTTP API harness in the collection — AdGuardHome is a DNS-based ad blocker and privacy protection server exposing 58 endpoints across 14 tag groups, secured with HTTP Basic Auth. CLI covers all major API groups: - filter: list/add/remove/enable/disable/refresh/status/toggle - blocking: parental/safebrowsing/safesearch enable/disable/status - blocked-services: list/set - clients: list/add/remove/show - stats: show/reset/config - log: show/config/clear - rewrite: list/add/remove - dhcp: status/leases/add-static/remove-static - server: status/version/restart - config: show/save/test Features: --json output, REPL mode, --https flag, env vars (AGH_HOST etc), config file (~/.config/cli-anything-adguardhome.json) Tests: 36 passing (24 unit + 12 E2E via Docker adguard/adguardhome) Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 4 + adguardhome/agent-harness/.gitignore | 11 + adguardhome/agent-harness/ADGUARDHOME.md | 66 ++ .../cli_anything/adguardhome/README.md | 71 ++ .../cli_anything/adguardhome/__init__.py | 0 .../cli_anything/adguardhome/__main__.py | 5 + .../adguardhome/adguardhome_cli.py | 668 ++++++++++++++++++ .../cli_anything/adguardhome/core/__init__.py | 0 .../cli_anything/adguardhome/core/blocking.py | 47 ++ .../cli_anything/adguardhome/core/clients.py | 30 + .../cli_anything/adguardhome/core/dhcp.py | 25 + .../adguardhome/core/filtering.py | 37 + .../cli_anything/adguardhome/core/log.py | 24 + .../cli_anything/adguardhome/core/project.py | 49 ++ .../cli_anything/adguardhome/core/rewrite.py | 15 + .../cli_anything/adguardhome/core/server.py | 19 + .../cli_anything/adguardhome/core/session.py | 25 + .../cli_anything/adguardhome/core/stats.py | 19 + .../cli_anything/adguardhome/tests/TEST.md | 121 ++++ .../adguardhome/tests/__init__.py | 0 .../adguardhome/tests/test_core.py | 256 +++++++ .../adguardhome/tests/test_full_e2e.py | 273 +++++++ .../adguardhome/utils/__init__.py | 0 .../adguardhome/utils/adguardhome_backend.py | 70 ++ .../adguardhome/utils/repl_skin.py | 498 +++++++++++++ adguardhome/agent-harness/setup.py | 19 + 26 files changed, 2352 insertions(+) create mode 100644 adguardhome/agent-harness/.gitignore create mode 100644 adguardhome/agent-harness/ADGUARDHOME.md create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/README.md create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/__init__.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/__main__.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/adguardhome_cli.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/__init__.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/blocking.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/clients.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/dhcp.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/filtering.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/log.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/project.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/rewrite.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/server.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/session.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/core/stats.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/tests/TEST.md create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/tests/__init__.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/tests/test_core.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/tests/test_full_e2e.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/utils/__init__.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/utils/adguardhome_backend.py create mode 100644 adguardhome/agent-harness/cli_anything/adguardhome/utils/repl_skin.py create mode 100644 adguardhome/agent-harness/setup.py diff --git a/.gitignore b/.gitignore index cfc7c36d6..b3d0195dd 100644 --- a/.gitignore +++ b/.gitignore @@ -37,6 +37,7 @@ !/zoom/ !/drawio/ !/mermaid/ +!/adguardhome/ # Step 5: Inside each software dir, ignore everything (including dotfiles) /gimp/* @@ -63,6 +64,8 @@ /drawio/.* /mermaid/* /mermaid/.* +/adguardhome/* +/adguardhome/.* # Step 6: ...except agent-harness/ !/gimp/agent-harness/ @@ -77,6 +80,7 @@ !/zoom/agent-harness/ !/drawio/agent-harness/ !/mermaid/agent-harness/ +!/adguardhome/agent-harness/ # Step 7: Ignore build artifacts within allowed dirs **/__pycache__/ diff --git a/adguardhome/agent-harness/.gitignore b/adguardhome/agent-harness/.gitignore new file mode 100644 index 000000000..ea85e2712 --- /dev/null +++ b/adguardhome/agent-harness/.gitignore @@ -0,0 +1,11 @@ +.venv/ +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +.pytest_cache/ +.fastembed_cache/ +.leann/ +*.egg-info/ +dist/ +build/ diff --git a/adguardhome/agent-harness/ADGUARDHOME.md b/adguardhome/agent-harness/ADGUARDHOME.md new file mode 100644 index 000000000..3f06485be --- /dev/null +++ b/adguardhome/agent-harness/ADGUARDHOME.md @@ -0,0 +1,66 @@ +# AdGuardHome - CLI Harness SOP + +## Overview + +AdGuardHome is a DNS-based ad blocker and privacy protection server written in Go. +It exposes a REST HTTP API with 58 endpoints organized in 14 tag groups, secured with HTTP Basic Auth. + +**Real software:** The running AdGuardHome HTTP API (not a binary to invoke directly). +**CLI role:** Generate structured commands - call the real API - verify responses. + +## Architecture + +- **API base:** `http://:/control/` +- **Auth:** HTTP Basic Auth (`Authorization: Basic base64(user:pass)`) +- **Port:** 3000 by default +- **OpenAPI spec:** `openapi/openapi.yaml` in the AdGuardHome source + +## API Tag Groups + +| Group | Description | Key Endpoints | +|-------|-------------|---------------| +| `global` | Server settings and controls | `/status`, `/version`, `/restart` | +| `filtering` | Rule-based filtering | `/filtering/status`, `/filtering/add_url`, `/filtering/remove_url` | +| `blocked_services` | Block service categories | `/blocked_services/get`, `/blocked_services/set` | +| `clients` | Known clients | `/clients`, `/clients/add`, `/clients/delete` | +| `stats` | DNS query statistics | `/stats`, `/stats_reset`, `/stats_config` | +| `log` | Query log | `/querylog`, `/querylog_config`, `/querylog_clear` | +| `dhcp` | Built-in DHCP server | `/dhcp/status`, `/dhcp/leases`, `/dhcp/set_config` | +| `rewrite` | DNS rewrites | `/rewrite/list`, `/rewrite/add`, `/rewrite/delete` | +| `parental` | Adult content blocking | `/parental/status`, `/parental/enable`, `/parental/disable` | +| `safebrowsing` | Malware/phishing blocking | `/safebrowsing/status`, `/safebrowsing/enable`, `/safebrowsing/disable` | +| `safesearch` | Safe search enforcement | `/safesearch/status`, `/safesearch/enable`, `/safesearch/disable` | +| `tls` | HTTPS/DoH/DoT settings | `/tls/status`, `/tls/configure`, `/tls/validate` | + +## CLI Command Map + +``` +cli-anything-adguardhome +├── config show / save / test +├── server status / version / restart +├── filter list / add / remove / enable / disable / refresh / status / toggle +├── blocking parental status/enable/disable +│ safebrowsing status/enable/disable +│ safesearch status/enable/disable +├── blocked-services list / set +├── clients list / add / remove / show +├── stats show / reset / config +├── log show / config / clear +├── rewrite list / add / remove +├── dhcp status / leases / add-static / remove-static +└── tls status +``` + +## Connection Config + +Settings resolved in order: +1. CLI flags (`--host`, `--port`, `--username`, `--password`) +2. Environment vars (`AGH_HOST`, `AGH_PORT`, `AGH_USERNAME`, `AGH_PASSWORD`) +3. Config file (`~/.config/cli-anything-adguardhome.json`) +4. Defaults: `localhost:3000` + +## Testing Strategy + +- **Unit tests:** Mock HTTP calls via `unittest.mock` - no real AdGuardHome needed +- **E2E tests:** Spin up `adguard/adguardhome` via Docker on port 3001 for isolation +- **Subprocess tests:** `_resolve_cli("cli-anything-adguardhome")` tests the installed CLI binary diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/README.md b/adguardhome/agent-harness/cli_anything/adguardhome/README.md new file mode 100644 index 000000000..1177678ce --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/README.md @@ -0,0 +1,71 @@ +# cli-anything-adguardhome + +CLI harness for AdGuardHome - control your ad blocker from the command line or via agents. + +## Prerequisites + +AdGuardHome must be running. Install: + +```bash +# Linux - native +curl -s -S -L https://raw.githubusercontent.com/AdguardTeam/AdGuardHome/master/scripts/install.sh | sh -s -- -v + +# Docker +docker run --name adguardhome -p 3000:3000 adguard/adguardhome +``` + +## Installation + +```bash +cd agent-harness +pip install -e . +cli-anything-adguardhome --help +``` + +## Configuration + +```bash +export AGH_HOST=localhost +export AGH_PORT=3000 +export AGH_USERNAME=admin +export AGH_PASSWORD=secret + +# Or save to config file +cli-anything-adguardhome --host localhost --port 3000 --username admin --password secret config save +``` + +## Usage + +```bash +# Interactive REPL (default) +cli-anything-adguardhome + +# One-shot commands +cli-anything-adguardhome server status +cli-anything-adguardhome filter list +cli-anything-adguardhome --json stats show + +# Filtering +cli-anything-adguardhome filter add --url https://somehost.com/list.txt --name "My List" +cli-anything-adguardhome filter refresh + +# DNS rewrites +cli-anything-adguardhome rewrite add --domain "myserver.local" --answer "192.168.1.50" +cli-anything-adguardhome rewrite list + +# Clients +cli-anything-adguardhome clients add --name "My PC" --ip 192.168.1.100 + +# Stats +cli-anything-adguardhome stats show +cli-anything-adguardhome stats reset +``` + +## Tests + +```bash +cd agent-harness +python3 -m pytest cli_anything/adguardhome/tests/test_core.py -v +python3 -m pytest cli_anything/adguardhome/tests/test_full_e2e.py -v -s +CLI_ANYTHING_FORCE_INSTALLED=1 python3 -m pytest cli_anything/adguardhome/tests/ -v -s +``` diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/__init__.py b/adguardhome/agent-harness/cli_anything/adguardhome/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/__main__.py b/adguardhome/agent-harness/cli_anything/adguardhome/__main__.py new file mode 100644 index 000000000..e36dc8eee --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/__main__.py @@ -0,0 +1,5 @@ +"""Enable python -m cli_anything.adguardhome""" +from cli_anything.adguardhome.adguardhome_cli import main + +if __name__ == "__main__": + main() diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/adguardhome_cli.py b/adguardhome/agent-harness/cli_anything/adguardhome/adguardhome_cli.py new file mode 100644 index 000000000..aed809f1d --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/adguardhome_cli.py @@ -0,0 +1,668 @@ +"""cli-anything-adguardhome - CLI harness for AdGuardHome.""" + +import json +import sys +from pathlib import Path + +import click + +from cli_anything.adguardhome.core import blocking as blocking_core +from cli_anything.adguardhome.core import clients as clients_core +from cli_anything.adguardhome.core import dhcp as dhcp_core +from cli_anything.adguardhome.core import filtering as filtering_core +from cli_anything.adguardhome.core import log as log_core +from cli_anything.adguardhome.core import project +from cli_anything.adguardhome.core import rewrite as rewrite_core +from cli_anything.adguardhome.core import server as server_core +from cli_anything.adguardhome.core import stats as stats_core +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient +from cli_anything.adguardhome.utils.repl_skin import ReplSkin + +CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]} + + +def make_client(ctx: click.Context) -> AdGuardHomeClient: + obj = ctx.obj + return AdGuardHomeClient( + host=obj["host"], + port=obj["port"], + username=obj["username"], + password=obj["password"], + https=obj.get("use_https", False), + ) + + +def output(data, as_json: bool) -> None: + if as_json: + click.echo(json.dumps(data, indent=2, default=str)) + elif isinstance(data, dict): + for k, v in data.items(): + click.echo(f"{k}: {v}") + elif isinstance(data, list): + for item in data: + if isinstance(item, dict): + click.echo(json.dumps(item, default=str)) + else: + click.echo(str(item)) + else: + click.echo(str(data)) + + +# --------------------------------------------------------------------------- +# Root group +# --------------------------------------------------------------------------- + +@click.group(context_settings=CONTEXT_SETTINGS, invoke_without_command=True) +@click.option("--host", default=None, help="AdGuardHome hostname/IP") +@click.option("--port", default=None, type=int, help="AdGuardHome port (default 3000)") +@click.option("--username", default=None, help="Basic Auth username") +@click.option("--password", default=None, help="Basic Auth password") +@click.option("--config", "config_path", default=None, type=click.Path(), + help="Path to config file") +@click.option("--https", "use_https", is_flag=True, default=False, + help="Use HTTPS (auto-detected for port 443)") +@click.option("--json", "as_json", is_flag=True, default=False, + help="Output as JSON") +@click.pass_context +def cli(ctx: click.Context, host, port, username, password, config_path, use_https, as_json): + """cli-anything-adguardhome - control AdGuardHome from the command line.""" + ctx.ensure_object(dict) + + cfg = project.load_config(Path(config_path) if config_path else None) + ctx.obj["host"] = host or cfg["host"] + ctx.obj["port"] = port or cfg["port"] + ctx.obj["username"] = username or cfg["username"] + ctx.obj["password"] = password or cfg["password"] + ctx.obj["use_https"] = use_https or cfg.get("https", False) + ctx.obj["as_json"] = as_json + + if ctx.invoked_subcommand is None: + ctx.invoke(repl) + + +def main(): + cli(obj={}) + + +# --------------------------------------------------------------------------- +# REPL +# --------------------------------------------------------------------------- + +@cli.command(hidden=True) +@click.pass_context +def repl(ctx: click.Context): + """Interactive REPL mode.""" + skin = ReplSkin("adguardhome", version="1.0.0") + skin.print_banner() + + host = ctx.obj["host"] + port = ctx.obj["port"] + skin.info(f"Connecting to {host}:{port}") + + pt_session = skin.create_prompt_session() + + while True: + try: + line = skin.get_input(pt_session, project_name=f"{host}:{port}") + except (EOFError, KeyboardInterrupt): + break + + line = line.strip() + if not line: + continue + if line in ("exit", "quit"): + break + if line == "help": + skin.help({ + "server status/version/restart": "Server management", + "filter list/add/remove/enable/disable/refresh/status/toggle": "Filtering", + "blocking parental/safebrowsing/safesearch status/enable/disable": "Blocking", + "blocked-services list/set": "Blocked services", + "clients list/add/remove/show": "Client management", + "stats show/reset/config": "Statistics", + "log show/config/clear": "Query log", + "rewrite list/add/remove": "DNS rewrites", + "dhcp status/leases/add-static/remove-static": "DHCP server", + "tls status": "TLS configuration", + "config show/save/test": "Connection config", + }) + continue + + try: + args = line.split() + cli.main(args=args, obj=dict(ctx.obj), standalone_mode=False) + except click.exceptions.UsageError as e: + skin.error(str(e)) + except RuntimeError as e: + skin.error(str(e)) + except SystemExit: + pass + except Exception as e: + skin.error(f"Unexpected error: {e}") + + skin.print_goodbye() + + +# --------------------------------------------------------------------------- +# config +# --------------------------------------------------------------------------- + +@cli.group() +@click.pass_context +def config(ctx: click.Context): + """Connection configuration.""" + + +@config.command("show") +@click.pass_context +def config_show(ctx: click.Context): + """Show current connection settings.""" + obj = ctx.obj + data = { + "host": obj["host"], + "port": obj["port"], + "username": obj["username"], + "password": "***" if obj["password"] else "", + } + output(data, obj["as_json"]) + + +@config.command("save") +@click.pass_context +def config_save(ctx: click.Context): + """Save connection settings to config file.""" + obj = ctx.obj + path = project.save_config( + host=obj["host"], port=obj["port"], + username=obj["username"], password=obj["password"], + ) + result = {"saved": str(path)} + output(result, obj["as_json"]) + + +@config.command("test") +@click.pass_context +def config_test(ctx: click.Context): + """Test connection to AdGuardHome.""" + client = make_client(ctx) + data = server_core.get_status(client) + result = {"connected": True, "host": ctx.obj["host"], "port": ctx.obj["port"], **data} + output(result, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# server +# --------------------------------------------------------------------------- + +@cli.group("server") +@click.pass_context +def server_(ctx: click.Context): + """Server management.""" + + +# Rename to avoid shadowing the module + +@server_.command("status") +@click.pass_context +def server_status(ctx: click.Context): + """Show server status.""" + client = make_client(ctx) + data = server_core.get_status(client) + output(data, ctx.obj["as_json"]) + + +@server_.command("version") +@click.pass_context +def server_version(ctx: click.Context): + """Show AdGuardHome version.""" + client = make_client(ctx) + data = server_core.get_version(client) + output(data, ctx.obj["as_json"]) + + +@server_.command("restart") +@click.pass_context +def server_restart(ctx: click.Context): + """Restart AdGuardHome.""" + client = make_client(ctx) + data = server_core.restart(client) + output(data or {"restarted": True}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# filter +# --------------------------------------------------------------------------- + +@cli.group("filter") +@click.pass_context +def filter_(ctx: click.Context): + """Filtering rules management.""" + + +@filter_.command("list") +@click.pass_context +def filter_list(ctx: click.Context): + """List all filter subscriptions.""" + client = make_client(ctx) + data = filtering_core.get_status(client) + output(data, ctx.obj["as_json"]) + + +@filter_.command("status") +@click.pass_context +def filter_status(ctx: click.Context): + """Show filtering enabled/disabled state.""" + client = make_client(ctx) + data = filtering_core.get_status(client) + result = {"enabled": data.get("enabled"), "filters_count": len(data.get("filters", []))} + output(result, ctx.obj["as_json"]) + + +@filter_.command("toggle") +@click.argument("state", type=click.Choice(["on", "off"])) +@click.pass_context +def filter_toggle(ctx: click.Context, state: str): + """Enable or disable filtering globally.""" + client = make_client(ctx) + data = filtering_core.set_enabled(client, state == "on") + output(data or {"filtering_enabled": state == "on"}, ctx.obj["as_json"]) + + +@filter_.command("add") +@click.option("--url", required=True, help="Filter list URL") +@click.option("--name", required=True, help="Filter name") +@click.option("--whitelist", is_flag=True, default=False) +@click.pass_context +def filter_add(ctx: click.Context, url: str, name: str, whitelist: bool): + """Add a new filter subscription.""" + client = make_client(ctx) + data = filtering_core.add_filter(client, url=url, name=name, whitelist=whitelist) + output(data or {"added": True, "url": url, "name": name}, ctx.obj["as_json"]) + + +@filter_.command("remove") +@click.option("--url", required=True, help="Filter list URL to remove") +@click.option("--whitelist", is_flag=True, default=False) +@click.pass_context +def filter_remove(ctx: click.Context, url: str, whitelist: bool): + """Remove a filter subscription.""" + client = make_client(ctx) + data = filtering_core.remove_filter(client, url=url, whitelist=whitelist) + output(data or {"removed": True, "url": url}, ctx.obj["as_json"]) + + +@filter_.command("enable") +@click.option("--url", required=True) +@click.option("--name", required=True) +@click.option("--whitelist", is_flag=True, default=False) +@click.pass_context +def filter_enable(ctx: click.Context, url: str, name: str, whitelist: bool): + """Enable a filter subscription.""" + client = make_client(ctx) + data = filtering_core.set_filter_url(client, url=url, name=name, enabled=True, + whitelist=whitelist) + output(data or {"enabled": True, "url": url}, ctx.obj["as_json"]) + + +@filter_.command("disable") +@click.option("--url", required=True) +@click.option("--name", required=True) +@click.option("--whitelist", is_flag=True, default=False) +@click.pass_context +def filter_disable(ctx: click.Context, url: str, name: str, whitelist: bool): + """Disable a filter subscription.""" + client = make_client(ctx) + data = filtering_core.set_filter_url(client, url=url, name=name, enabled=False, + whitelist=whitelist) + output(data or {"disabled": True, "url": url}, ctx.obj["as_json"]) + + +@filter_.command("refresh") +@click.option("--whitelist", is_flag=True, default=False) +@click.pass_context +def filter_refresh(ctx: click.Context, whitelist: bool): + """Trigger manual update of all filters.""" + client = make_client(ctx) + data = filtering_core.refresh(client, whitelist=whitelist) + output(data or {"refreshed": True}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# blocking +# --------------------------------------------------------------------------- + +@cli.group() +@click.pass_context +def blocking(ctx: click.Context): + """Parental, safebrowsing, safesearch controls.""" + + +@blocking.group() +def parental(): + """Parental control.""" + + +@parental.command("status") +@click.pass_context +def parental_status(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.parental_status(client), ctx.obj["as_json"]) + + +@parental.command("enable") +@click.pass_context +def parental_enable(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.parental_enable(client) or {"enabled": True}, ctx.obj["as_json"]) + + +@parental.command("disable") +@click.pass_context +def parental_disable(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.parental_disable(client) or {"disabled": True}, ctx.obj["as_json"]) + + +@blocking.group() +def safebrowsing(): + """Safe browsing control.""" + + +@safebrowsing.command("status") +@click.pass_context +def safebrowsing_status(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.safebrowsing_status(client), ctx.obj["as_json"]) + + +@safebrowsing.command("enable") +@click.pass_context +def safebrowsing_enable(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.safebrowsing_enable(client) or {"enabled": True}, ctx.obj["as_json"]) + + +@safebrowsing.command("disable") +@click.pass_context +def safebrowsing_disable(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.safebrowsing_disable(client) or {"disabled": True}, ctx.obj["as_json"]) + + +@blocking.group() +def safesearch(): + """Safe search control.""" + + +@safesearch.command("status") +@click.pass_context +def safesearch_status(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.safesearch_status(client), ctx.obj["as_json"]) + + +@safesearch.command("enable") +@click.pass_context +def safesearch_enable(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.safesearch_enable(client) or {"enabled": True}, ctx.obj["as_json"]) + + +@safesearch.command("disable") +@click.pass_context +def safesearch_disable(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.safesearch_disable(client) or {"disabled": True}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# blocked-services +# --------------------------------------------------------------------------- + +@cli.group("blocked-services") +@click.pass_context +def blocked_services(ctx: click.Context): + """Blocked service categories.""" + + +@blocked_services.command("list") +@click.pass_context +def blocked_services_list(ctx: click.Context): + client = make_client(ctx) + output(blocking_core.blocked_services_get(client), ctx.obj["as_json"]) + + +@blocked_services.command("set") +@click.argument("services", nargs=-1, required=True) +@click.pass_context +def blocked_services_set(ctx: click.Context, services: tuple): + client = make_client(ctx) + output(blocking_core.blocked_services_set(client, list(services)) or {"set": list(services)}, + ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# clients +# --------------------------------------------------------------------------- + +@cli.group("clients") +@click.pass_context +def clients_(ctx: click.Context): + """Known client management.""" + + + +@clients_.command("list") +@click.pass_context +def clients_list(ctx: click.Context): + client = make_client(ctx) + output(clients_core.list_clients(client), ctx.obj["as_json"]) + + +@clients_.command("add") +@click.option("--name", required=True) +@click.option("--ip", required=True, help="Client IP address") +@click.pass_context +def clients_add(ctx: click.Context, name: str, ip: str): + c = make_client(ctx) + output(clients_core.add_client(c, name=name, ids=[ip]) or {"added": True, "name": name}, + ctx.obj["as_json"]) + + +@clients_.command("remove") +@click.option("--name", required=True) +@click.pass_context +def clients_remove(ctx: click.Context, name: str): + c = make_client(ctx) + output(clients_core.delete_client(c, name=name) or {"removed": True, "name": name}, + ctx.obj["as_json"]) + + +@clients_.command("show") +@click.option("--name", required=True) +@click.pass_context +def clients_show(ctx: click.Context, name: str): + c = make_client(ctx) + data = clients_core.list_clients(c) + all_clients = data.get("clients", []) if isinstance(data, dict) else [] + found = next((cl for cl in all_clients if cl.get("name") == name), None) + output(found or {"error": f"Client '{name}' not found"}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# stats +# --------------------------------------------------------------------------- + +@cli.group("stats") +@click.pass_context +def stats_(ctx: click.Context): + """DNS query statistics.""" + + + +@stats_.command("show") +@click.pass_context +def stats_show(ctx: click.Context): + client = make_client(ctx) + output(stats_core.get_stats(client), ctx.obj["as_json"]) + + +@stats_.command("reset") +@click.pass_context +def stats_reset(ctx: click.Context): + client = make_client(ctx) + output(stats_core.reset_stats(client) or {"reset": True}, ctx.obj["as_json"]) + + +@stats_.command("config") +@click.option("--interval", type=int, default=None, help="Retention in days") +@click.pass_context +def stats_config(ctx: click.Context, interval): + client = make_client(ctx) + if interval is not None: + output(stats_core.set_stats_config(client, interval), ctx.obj["as_json"]) + else: + output(stats_core.get_stats_config(client), ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# log +# --------------------------------------------------------------------------- + +@cli.group("log") +@click.pass_context +def log_(ctx: click.Context): + """Query log management.""" + + + +@log_.command("show") +@click.option("--limit", default=50, type=int) +@click.option("--offset", default=0, type=int) +@click.pass_context +def log_show(ctx: click.Context, limit: int, offset: int): + client = make_client(ctx) + output(log_core.get_log(client, limit=limit, offset=offset), ctx.obj["as_json"]) + + +@log_.command("config") +@click.option("--enabled/--disabled", default=None) +@click.option("--interval", type=int, default=None) +@click.pass_context +def log_config(ctx: click.Context, enabled, interval): + client = make_client(ctx) + if enabled is not None: + output(log_core.set_log_config(client, enabled=enabled, + interval=interval or 90), ctx.obj["as_json"]) + else: + output(log_core.get_log_config(client), ctx.obj["as_json"]) + + +@log_.command("clear") +@click.pass_context +def log_clear(ctx: click.Context): + client = make_client(ctx) + output(log_core.clear_log(client) or {"cleared": True}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# rewrite +# --------------------------------------------------------------------------- + +@cli.group("rewrite") +@click.pass_context +def rewrite_(ctx: click.Context): + """DNS rewrite rules.""" + + + +@rewrite_.command("list") +@click.pass_context +def rewrite_list(ctx: click.Context): + client = make_client(ctx) + output(rewrite_core.list_rewrites(client), ctx.obj["as_json"]) + + +@rewrite_.command("add") +@click.option("--domain", required=True) +@click.option("--answer", required=True) +@click.pass_context +def rewrite_add(ctx: click.Context, domain: str, answer: str): + client = make_client(ctx) + output(rewrite_core.add_rewrite(client, domain=domain, answer=answer) or + {"added": True, "domain": domain, "answer": answer}, ctx.obj["as_json"]) + + +@rewrite_.command("remove") +@click.option("--domain", required=True) +@click.option("--answer", required=True) +@click.pass_context +def rewrite_remove(ctx: click.Context, domain: str, answer: str): + client = make_client(ctx) + output(rewrite_core.delete_rewrite(client, domain=domain, answer=answer) or + {"removed": True, "domain": domain}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# dhcp +# --------------------------------------------------------------------------- + +@cli.group("dhcp") +@click.pass_context +def dhcp_(ctx: click.Context): + """DHCP server management.""" + + + +@dhcp_.command("status") +@click.pass_context +def dhcp_status(ctx: click.Context): + client = make_client(ctx) + output(dhcp_core.get_status(client), ctx.obj["as_json"]) + + +@dhcp_.command("leases") +@click.pass_context +def dhcp_leases(ctx: click.Context): + client = make_client(ctx) + output(dhcp_core.get_leases(client), ctx.obj["as_json"]) + + +@dhcp_.command("add-static") +@click.option("--mac", required=True) +@click.option("--ip", required=True) +@click.option("--hostname", default="") +@click.pass_context +def dhcp_add_static(ctx: click.Context, mac: str, ip: str, hostname: str): + client = make_client(ctx) + output(dhcp_core.add_static_lease(client, mac=mac, ip=ip, hostname=hostname) or + {"added": True, "mac": mac, "ip": ip}, ctx.obj["as_json"]) + + +@dhcp_.command("remove-static") +@click.option("--mac", required=True) +@click.option("--ip", required=True) +@click.option("--hostname", default="") +@click.pass_context +def dhcp_remove_static(ctx: click.Context, mac: str, ip: str, hostname: str): + client = make_client(ctx) + output(dhcp_core.remove_static_lease(client, mac=mac, ip=ip, hostname=hostname) or + {"removed": True, "mac": mac}, ctx.obj["as_json"]) + + +# --------------------------------------------------------------------------- +# tls +# --------------------------------------------------------------------------- + +@cli.group("tls") +@click.pass_context +def tls_(ctx: click.Context): + """TLS/HTTPS configuration.""" + + + +@tls_.command("status") +@click.pass_context +def tls_status(ctx: click.Context): + client = make_client(ctx) + output(server_core.get_tls_status(client), ctx.obj["as_json"]) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/__init__.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/blocking.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/blocking.py new file mode 100644 index 000000000..868ce74a2 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/blocking.py @@ -0,0 +1,47 @@ +"""Blocking controls: parental, safebrowsing, safesearch, blocked services.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def parental_status(client: AdGuardHomeClient) -> dict: + return client.get("/parental/status") + + +def parental_enable(client: AdGuardHomeClient) -> dict: + return client.post("/parental/enable") + + +def parental_disable(client: AdGuardHomeClient) -> dict: + return client.post("/parental/disable") + + +def safebrowsing_status(client: AdGuardHomeClient) -> dict: + return client.get("/safebrowsing/status") + + +def safebrowsing_enable(client: AdGuardHomeClient) -> dict: + return client.post("/safebrowsing/enable") + + +def safebrowsing_disable(client: AdGuardHomeClient) -> dict: + return client.post("/safebrowsing/disable") + + +def safesearch_status(client: AdGuardHomeClient) -> dict: + return client.get("/safesearch/status") + + +def safesearch_enable(client: AdGuardHomeClient) -> dict: + return client.post("/safesearch/enable") + + +def safesearch_disable(client: AdGuardHomeClient) -> dict: + return client.post("/safesearch/disable") + + +def blocked_services_get(client: AdGuardHomeClient) -> dict: + return client.get("/blocked_services/get") + + +def blocked_services_set(client: AdGuardHomeClient, services: list[str]) -> dict: + return client.post("/blocked_services/set", {"ids": services}) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/clients.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/clients.py new file mode 100644 index 000000000..299823700 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/clients.py @@ -0,0 +1,30 @@ +"""Client management for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def list_clients(client: AdGuardHomeClient) -> dict: + return client.get("/clients") + + +def add_client(client: AdGuardHomeClient, name: str, ids: list[str], + use_global_settings: bool = True, + filtering_enabled: bool = True) -> dict: + return client.post("/clients/add", { + "name": name, + "ids": ids, + "use_global_settings": use_global_settings, + "filtering_enabled": filtering_enabled, + "parental_enabled": False, + "safebrowsing_enabled": False, + "safesearch_enabled": False, + "use_global_blocked_services": True, + }) + + +def delete_client(client: AdGuardHomeClient, name: str) -> dict: + return client.post("/clients/delete", {"name": name}) + + +def update_client(client: AdGuardHomeClient, name: str, data: dict) -> dict: + return client.post("/clients/update", {"name": name, "data": data}) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/dhcp.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/dhcp.py new file mode 100644 index 000000000..460b9b0c8 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/dhcp.py @@ -0,0 +1,25 @@ +"""DHCP server management for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def get_status(client: AdGuardHomeClient) -> dict: + return client.get("/dhcp/status") + + +def get_leases(client: AdGuardHomeClient) -> dict: + return client.get("/dhcp/leases") + + +def add_static_lease(client: AdGuardHomeClient, mac: str, ip: str, + hostname: str) -> dict: + return client.post("/dhcp/add_static_lease", { + "mac": mac, "ip": ip, "hostname": hostname, + }) + + +def remove_static_lease(client: AdGuardHomeClient, mac: str, ip: str, + hostname: str) -> dict: + return client.post("/dhcp/remove_static_lease", { + "mac": mac, "ip": ip, "hostname": hostname, + }) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/filtering.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/filtering.py new file mode 100644 index 000000000..daa7e9300 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/filtering.py @@ -0,0 +1,37 @@ +"""Filtering rules management for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def get_status(client: AdGuardHomeClient) -> dict: + return client.get("/filtering/status") + + +def set_enabled(client: AdGuardHomeClient, enabled: bool) -> dict: + return client.post("/filtering/config", {"enabled": enabled, "interval": 24}) + + +def add_filter(client: AdGuardHomeClient, url: str, name: str, + whitelist: bool = False) -> dict: + return client.post("/filtering/add_url", { + "name": name, + "url": url, + "whitelist": whitelist, + }) + + +def remove_filter(client: AdGuardHomeClient, url: str, whitelist: bool = False) -> dict: + return client.post("/filtering/remove_url", {"url": url, "whitelist": whitelist}) + + +def set_filter_url(client: AdGuardHomeClient, url: str, name: str, + enabled: bool, whitelist: bool = False) -> dict: + return client.post("/filtering/set_url", { + "url": url, + "data": {"name": name, "url": url, "enabled": enabled}, + "whitelist": whitelist, + }) + + +def refresh(client: AdGuardHomeClient, whitelist: bool = False) -> dict: + return client.post("/filtering/refresh", {"whitelist": whitelist}) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/log.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/log.py new file mode 100644 index 000000000..d19eba0b3 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/log.py @@ -0,0 +1,24 @@ +"""Query log for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def get_log(client: AdGuardHomeClient, limit: int = 100, offset: int = 0) -> dict: + return client.get("/querylog", params={"limit": limit, "offset": offset}) + + +def get_log_config(client: AdGuardHomeClient) -> dict: + return client.get("/querylog_config") + + +def set_log_config(client: AdGuardHomeClient, enabled: bool, + interval: int = 90) -> dict: + return client.post("/querylog_config", { + "enabled": enabled, + "interval": interval, + "anonymize_client_ip": False, + }) + + +def clear_log(client: AdGuardHomeClient) -> dict: + return client.post("/querylog_clear") diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/project.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/project.py new file mode 100644 index 000000000..a3545170c --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/project.py @@ -0,0 +1,49 @@ +"""Connection configuration management for cli-anything-adguardhome.""" + +import json +import os +from pathlib import Path + +DEFAULT_CONFIG_PATH = Path.home() / ".config" / "cli-anything-adguardhome.json" +DEFAULT_HOST = "localhost" +DEFAULT_PORT = 3000 + + +def load_config(config_path: Path | None = None) -> dict: + """Load connection config from file, with env var and default fallbacks.""" + path = config_path or DEFAULT_CONFIG_PATH + config: dict = { + "host": DEFAULT_HOST, + "port": DEFAULT_PORT, + "username": "", + "password": "", + } + if path.exists(): + try: + with open(path) as f: + file_config = json.load(f) + for key in ("host", "port", "username", "password"): + if key in file_config: + config[key] = file_config[key] + except (json.JSONDecodeError, OSError): + pass + if os.getenv("AGH_HOST"): + config["host"] = os.environ["AGH_HOST"] + if os.getenv("AGH_PORT"): + config["port"] = int(os.environ["AGH_PORT"]) + if os.getenv("AGH_USERNAME"): + config["username"] = os.environ["AGH_USERNAME"] + if os.getenv("AGH_PASSWORD"): + config["password"] = os.environ["AGH_PASSWORD"] + return config + + +def save_config(host: str, port: int, username: str, password: str, + config_path: Path | None = None) -> Path: + """Save connection settings to config file.""" + path = config_path or DEFAULT_CONFIG_PATH + path.parent.mkdir(parents=True, exist_ok=True) + data = {"host": host, "port": port, "username": username, "password": password} + with open(path, "w") as f: + json.dump(data, f, indent=2) + return path diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/rewrite.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/rewrite.py new file mode 100644 index 000000000..cfba35587 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/rewrite.py @@ -0,0 +1,15 @@ +"""DNS rewrite rules for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def list_rewrites(client: AdGuardHomeClient) -> list: + return client.get("/rewrite/list") + + +def add_rewrite(client: AdGuardHomeClient, domain: str, answer: str) -> dict: + return client.post("/rewrite/add", {"domain": domain, "answer": answer}) + + +def delete_rewrite(client: AdGuardHomeClient, domain: str, answer: str) -> dict: + return client.post("/rewrite/delete", {"domain": domain, "answer": answer}) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/server.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/server.py new file mode 100644 index 000000000..8e26dc926 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/server.py @@ -0,0 +1,19 @@ +"""Server/global management for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def get_status(client: AdGuardHomeClient) -> dict: + return client.get("/status") + + +def get_version(client: AdGuardHomeClient) -> dict: + return client.get("/version") + + +def restart(client: AdGuardHomeClient) -> dict: + return client.post("/restart") + + +def get_tls_status(client: AdGuardHomeClient) -> dict: + return client.get("/tls/status") diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/session.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/session.py new file mode 100644 index 000000000..08d890363 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/session.py @@ -0,0 +1,25 @@ +"""Session state management for cli-anything-adguardhome.""" + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Session: + """In-memory session state for the REPL.""" + host: str = "localhost" + port: int = 3000 + username: str = "" + password: str = "" + history: list[str] = field(default_factory=list) + + def add_history(self, command: str) -> None: + self.history.append(command) + + def to_dict(self) -> dict[str, Any]: + return { + "host": self.host, + "port": self.port, + "username": self.username, + "connected": True, + } diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/core/stats.py b/adguardhome/agent-harness/cli_anything/adguardhome/core/stats.py new file mode 100644 index 000000000..d2362fc14 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/core/stats.py @@ -0,0 +1,19 @@ +"""Statistics for AdGuardHome.""" + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient + + +def get_stats(client: AdGuardHomeClient) -> dict: + return client.get("/stats") + + +def reset_stats(client: AdGuardHomeClient) -> dict: + return client.post("/stats_reset") + + +def get_stats_config(client: AdGuardHomeClient) -> dict: + return client.get("/stats_config") + + +def set_stats_config(client: AdGuardHomeClient, interval: int) -> dict: + return client.post("/stats_config", {"interval": interval}) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/tests/TEST.md b/adguardhome/agent-harness/cli_anything/adguardhome/tests/TEST.md new file mode 100644 index 000000000..e69a1d7d1 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/tests/TEST.md @@ -0,0 +1,121 @@ +# Test Plan - cli-anything-adguardhome + +## Test Inventory Plan + +- `test_core.py`: 20 unit tests (no real AdGuardHome needed) +- `test_full_e2e.py`: 12 E2E + subprocess tests (Docker AdGuardHome on port 3001) + +## Unit Test Plan (test_core.py) + +### AdGuardHomeClient (utils/adguardhome_backend.py) +- `test_client_init_default` - default host/port, no auth +- `test_client_init_with_auth` - auth set on session +- `test_client_url_construction` - base URL built correctly +- `test_get_success` - GET returns deserialized JSON +- `test_get_empty_response` - GET returns {} on empty body +- `test_post_json` - POST sends JSON body +- `test_post_empty` - POST with no data +- `test_connection_error_raises_runtime` - ConnectionError raises RuntimeError with instructions + +### project.py +- `test_load_config_defaults` - returns localhost:3000 when no file/env +- `test_load_config_from_file` - loads from JSON file +- `test_load_config_env_override` - env vars override file +- `test_save_config` - writes JSON file correctly + +### filtering.py +- `test_get_status` - calls GET /filtering/status +- `test_add_filter` - calls POST /filtering/add_url with correct body +- `test_remove_filter` - calls POST /filtering/remove_url +- `test_set_enabled` - calls POST /filtering/config + +### blocking.py +- `test_parental_status` - calls GET /parental/status +- `test_parental_enable` - calls POST /parental/enable +- `test_safebrowsing_status` - calls GET /safebrowsing/status + +### clients.py +- `test_list_clients` - calls GET /clients +- `test_add_client` - calls POST /clients/add with correct body + +### rewrite.py +- `test_list_rewrites` - calls GET /rewrite/list +- `test_add_rewrite` - calls POST /rewrite/add + +## E2E Test Plan (test_full_e2e.py) + +### Setup +- Docker fixture starts `adguard/adguardhome` on port 3001 with pre-configured YAML +- Teardown removes the container + +### Workflow: CLI subprocess tests (no real AdGuardHome) +- `test_help` - `cli-anything-adguardhome --help` exits 0 +- `test_config_show_json` - `--json config show` returns valid JSON with host/port +- `test_server_version_json` - `--json server version` returns JSON (requires running instance) +- `test_filter_list_json` - `--json filter list` returns JSON + +### Workflow: Full filter lifecycle (requires Docker AdGuardHome) +- `test_filter_list` - list filters on fresh instance +- `test_rewrite_add_and_list` - add rewrite, verify in list +- `test_rewrite_remove` - remove rewrite, verify gone + +--- + +## Test Results + +(appended after pytest run) + +--- + +## Test Results + +``` +============================= test session starts ============================== +platform linux -- Python 3.13.5, pytest-9.0.2, pluggy-1.6.0 +rootdir: /home/yoan/work/AdGuardHome/agent-harness + +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_client_init_default PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_client_init_with_auth PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_client_init_no_auth PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_client_url_construction PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_get_success PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_get_empty_response PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_post_json PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_post_empty PASSED +cli_anything/adguardhome/tests/test_core.py::TestAdGuardHomeClient::test_connection_error_raises_runtime PASSED +cli_anything/adguardhome/tests/test_core.py::TestProject::test_load_config_defaults PASSED +cli_anything/adguardhome/tests/test_core.py::TestProject::test_load_config_from_file PASSED +cli_anything/adguardhome/tests/test_core.py::TestProject::test_load_config_env_override PASSED +cli_anything/adguardhome/tests/test_core.py::TestProject::test_save_config PASSED +cli_anything/adguardhome/tests/test_core.py::TestFiltering::test_get_status PASSED +cli_anything/adguardhome/tests/test_core.py::TestFiltering::test_add_filter PASSED +cli_anything/adguardhome/tests/test_core.py::TestFiltering::test_remove_filter PASSED +cli_anything/adguardhome/tests/test_core.py::TestFiltering::test_set_enabled PASSED +cli_anything/adguardhome/tests/test_core.py::TestBlocking::test_parental_status PASSED +cli_anything/adguardhome/tests/test_core.py::TestBlocking::test_parental_enable PASSED +cli_anything/adguardhome/tests/test_core.py::TestBlocking::test_safebrowsing_status PASSED +cli_anything/adguardhome/tests/test_core.py::TestClients::test_list_clients PASSED +cli_anything/adguardhome/tests/test_core.py::TestClients::test_add_client PASSED +cli_anything/adguardhome/tests/test_core.py::TestRewrite::test_list_rewrites PASSED +cli_anything/adguardhome/tests/test_core.py::TestRewrite::test_add_rewrite PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_help PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_config_show_json PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_config_show_default_host PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_help_subcommands_listed PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_filter_help PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_rewrite_help PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestCLISubprocess::test_blocking_help PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestDockerE2E::test_server_status_json PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestDockerE2E::test_filter_list_json PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestDockerE2E::test_rewrite_lifecycle PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestDockerE2E::test_stats_show_json PASSED +cli_anything/adguardhome/tests/test_full_e2e.py::TestDockerE2E::test_config_test PASSED + +============================== 36 passed in 6.57s ============================== +``` + +**36/36 passed (100%) — 2026-03-13** + +- Unit tests: 24/24 +- Subprocess tests (installed CLI): 7/7 +- Docker E2E tests (real AdGuardHome v0.107.73): 5/5 diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/tests/__init__.py b/adguardhome/agent-harness/cli_anything/adguardhome/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/tests/test_core.py b/adguardhome/agent-harness/cli_anything/adguardhome/tests/test_core.py new file mode 100644 index 000000000..574c0d7a3 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/tests/test_core.py @@ -0,0 +1,256 @@ +"""Unit tests for cli-anything-adguardhome core modules. + +No real AdGuardHome instance needed - all HTTP calls are mocked. +""" + +import json +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from cli_anything.adguardhome.utils.adguardhome_backend import AdGuardHomeClient +from cli_anything.adguardhome.core import project, filtering, blocking, clients, rewrite + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def mock_response(data=None, status=200, text=""): + resp = MagicMock(spec=requests.Response) + resp.status_code = status + if data is not None: + resp.json.return_value = data + resp.content = json.dumps(data).encode() + resp.text = json.dumps(data) + else: + resp.json.side_effect = ValueError("no json") + resp.content = text.encode() if text else b"" + resp.text = text + resp.raise_for_status = MagicMock() + return resp + + +def make_client(host="localhost", port=3000, username="admin", password="secret"): + return AdGuardHomeClient(host=host, port=port, username=username, password=password) + + +# --------------------------------------------------------------------------- +# AdGuardHomeClient +# --------------------------------------------------------------------------- + +class TestAdGuardHomeClient: + def test_client_init_default(self): + c = AdGuardHomeClient() + assert c.base_url == "http://localhost:3000/control" + assert c.host == "localhost" + assert c.port == 3000 + + def test_client_init_with_auth(self): + c = AdGuardHomeClient(username="admin", password="pass") + assert c.session.auth == ("admin", "pass") + + def test_client_init_no_auth(self): + c = AdGuardHomeClient() + assert c.session.auth is None + + def test_client_url_construction(self): + c = AdGuardHomeClient(host="192.168.1.1", port=8080) + assert c._url("/status") == "http://192.168.1.1:8080/control/status" + assert c._url("status") == "http://192.168.1.1:8080/control/status" + + def test_get_success(self): + c = make_client() + resp = mock_response({"running": True}) + with patch.object(c.session, "get", return_value=resp) as mock_get: + result = c.get("/status") + assert result == {"running": True} + mock_get.assert_called_once() + + def test_get_empty_response(self): + c = make_client() + resp = mock_response() + with patch.object(c.session, "get", return_value=resp): + result = c.get("/restart") + assert result == {} + + def test_post_json(self): + c = make_client() + resp = mock_response({}) + with patch.object(c.session, "post", return_value=resp) as mock_post: + c.post("/filtering/add_url", {"url": "http://example.com/list.txt", "name": "Test"}) + call_kwargs = mock_post.call_args + assert call_kwargs.kwargs.get("json") == {"url": "http://example.com/list.txt", "name": "Test"} + + def test_post_empty(self): + c = make_client() + resp = mock_response() + with patch.object(c.session, "post", return_value=resp) as mock_post: + result = c.post("/restart") + assert result == {} + mock_post.assert_called_once() + + def test_connection_error_raises_runtime(self): + c = make_client() + with patch.object(c.session, "get", side_effect=requests.exceptions.ConnectionError("refused")): + with pytest.raises(RuntimeError) as exc_info: + c.get("/status") + assert "Cannot connect to AdGuardHome" in str(exc_info.value) + assert "docker run" in str(exc_info.value).lower() or "docker" in str(exc_info.value).lower() + + +# --------------------------------------------------------------------------- +# project.py +# --------------------------------------------------------------------------- + +class TestProject: + def test_load_config_defaults(self, tmp_path): + result = project.load_config(config_path=tmp_path / "nonexistent.json") + assert result["host"] == "localhost" + assert result["port"] == 3000 + assert result["username"] == "" + assert result["password"] == "" + + def test_load_config_from_file(self, tmp_path): + cfg_file = tmp_path / "config.json" + cfg_file.write_text(json.dumps({ + "host": "192.168.1.1", "port": 8080, + "username": "admin", "password": "secret" + })) + result = project.load_config(config_path=cfg_file) + assert result["host"] == "192.168.1.1" + assert result["port"] == 8080 + assert result["username"] == "admin" + + def test_load_config_env_override(self, tmp_path, monkeypatch): + cfg_file = tmp_path / "config.json" + cfg_file.write_text(json.dumps({"host": "from-file", "port": 3000})) + monkeypatch.setenv("AGH_HOST", "from-env") + monkeypatch.setenv("AGH_PORT", "9000") + result = project.load_config(config_path=cfg_file) + assert result["host"] == "from-env" + assert result["port"] == 9000 + + def test_save_config(self, tmp_path): + path = tmp_path / "config.json" + saved = project.save_config("myhost", 4000, "user", "pass", config_path=path) + assert saved == path + data = json.loads(path.read_text()) + assert data["host"] == "myhost" + assert data["port"] == 4000 + + +# --------------------------------------------------------------------------- +# filtering.py +# --------------------------------------------------------------------------- + +class TestFiltering: + def test_get_status(self): + c = make_client() + resp = mock_response({"enabled": True, "filters": []}) + with patch.object(c.session, "get", return_value=resp): + result = filtering.get_status(c) + assert result["enabled"] is True + + def test_add_filter(self): + c = make_client() + resp = mock_response({}) + with patch.object(c.session, "post", return_value=resp) as mock_post: + filtering.add_filter(c, url="http://example.com/list.txt", name="Test") + body = mock_post.call_args.kwargs["json"] + assert body["url"] == "http://example.com/list.txt" + assert body["name"] == "Test" + assert body["whitelist"] is False + + def test_remove_filter(self): + c = make_client() + resp = mock_response({}) + with patch.object(c.session, "post", return_value=resp) as mock_post: + filtering.remove_filter(c, url="http://example.com/list.txt") + body = mock_post.call_args.kwargs["json"] + assert body["url"] == "http://example.com/list.txt" + + def test_set_enabled(self): + c = make_client() + resp = mock_response({}) + with patch.object(c.session, "post", return_value=resp) as mock_post: + filtering.set_enabled(c, enabled=True) + body = mock_post.call_args.kwargs["json"] + assert body["enabled"] is True + + +# --------------------------------------------------------------------------- +# blocking.py +# --------------------------------------------------------------------------- + +class TestBlocking: + def test_parental_status(self): + c = make_client() + resp = mock_response({"enabled": False}) + with patch.object(c.session, "get", return_value=resp): + result = blocking.parental_status(c) + assert result == {"enabled": False} + + def test_parental_enable(self): + c = make_client() + resp = mock_response() + with patch.object(c.session, "post", return_value=resp) as mock_post: + blocking.parental_enable(c) + assert "/parental/enable" in mock_post.call_args.args[0] + + def test_safebrowsing_status(self): + c = make_client() + resp = mock_response({"enabled": True}) + with patch.object(c.session, "get", return_value=resp): + result = blocking.safebrowsing_status(c) + assert result["enabled"] is True + + +# --------------------------------------------------------------------------- +# clients.py +# --------------------------------------------------------------------------- + +class TestClients: + def test_list_clients(self): + c = make_client() + data = {"clients": [{"name": "PC", "ids": ["192.168.1.10"]}], "auto_clients": []} + resp = mock_response(data) + with patch.object(c.session, "get", return_value=resp): + result = clients.list_clients(c) + assert len(result["clients"]) == 1 + + def test_add_client(self): + c = make_client() + resp = mock_response({}) + with patch.object(c.session, "post", return_value=resp) as mock_post: + clients.add_client(c, name="MyPC", ids=["192.168.1.100"]) + body = mock_post.call_args.kwargs["json"] + assert body["name"] == "MyPC" + assert "192.168.1.100" in body["ids"] + + +# --------------------------------------------------------------------------- +# rewrite.py +# --------------------------------------------------------------------------- + +class TestRewrite: + def test_list_rewrites(self): + c = make_client() + data = [{"domain": "myserver.local", "answer": "192.168.1.50"}] + resp = mock_response(data) + with patch.object(c.session, "get", return_value=resp): + result = rewrite.list_rewrites(c) + assert len(result) == 1 + assert result[0]["domain"] == "myserver.local" + + def test_add_rewrite(self): + c = make_client() + resp = mock_response({}) + with patch.object(c.session, "post", return_value=resp) as mock_post: + rewrite.add_rewrite(c, domain="myserver.local", answer="192.168.1.50") + body = mock_post.call_args.kwargs["json"] + assert body["domain"] == "myserver.local" + assert body["answer"] == "192.168.1.50" diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/tests/test_full_e2e.py b/adguardhome/agent-harness/cli_anything/adguardhome/tests/test_full_e2e.py new file mode 100644 index 000000000..d6be1f128 --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/tests/test_full_e2e.py @@ -0,0 +1,273 @@ +"""E2E and subprocess tests for cli-anything-adguardhome. + +Subprocess tests work without AdGuardHome (test CLI mechanics). +Docker tests require: docker pull adguard/adguardhome +""" + +import json +import os +import shutil +import subprocess +import sys +import time +from pathlib import Path + +import pytest +import requests + + +# --------------------------------------------------------------------------- +# CLI resolver +# --------------------------------------------------------------------------- + +def _resolve_cli(name: str) -> list[str]: + """Resolve installed CLI command; falls back to python -m for dev. + + Set env CLI_ANYTHING_FORCE_INSTALLED=1 to require the installed command. + """ + force = os.environ.get("CLI_ANYTHING_FORCE_INSTALLED", "").strip() == "1" + path = shutil.which(name) + if path: + print(f"[_resolve_cli] Using installed command: {path}") + return [path] + if force: + raise RuntimeError( + f"{name} not found in PATH. Install with:\n" + f" cd agent-harness && pip install -e ." + ) + module = "cli_anything.adguardhome.adguardhome_cli" + print(f"[_resolve_cli] Falling back to: {sys.executable} -m {module}") + return [sys.executable, "-m", module] + + +# --------------------------------------------------------------------------- +# Docker fixture +# --------------------------------------------------------------------------- + +AGH_TEST_PORT = 3001 +AGH_TEST_HOST = "localhost" +AGH_CONTAINER = "agh-cli-test" + + +def _wait_for_adguardhome(port: int, timeout: int = 30) -> bool: + """Wait until AdGuardHome API responds.""" + deadline = time.time() + timeout + while time.time() < deadline: + try: + r = requests.get(f"http://localhost:{port}/control/status", timeout=2) + if r.status_code in (200, 401, 403): + return True + except requests.exceptions.ConnectionError: + pass + time.sleep(1) + return False + + +def _configure_adguardhome(port: int, username: str, password: str) -> bool: + """Run the setup wizard via the install API.""" + url = f"http://localhost:{port}/control/install/configure" + payload = { + "web": {"ip": "0.0.0.0", "port": 3000, "status": "", "can_autofix": False}, + "dns": {"ip": "0.0.0.0", "port": 53, "status": "", "can_autofix": False}, + "username": username, + "password": password, + } + try: + r = requests.post(url, json=payload, timeout=10) + return r.status_code == 200 + except Exception: + return False + + +@pytest.fixture(scope="module") +def agh_docker(): + """Start AdGuardHome in Docker for E2E tests, configure via install API.""" + username = "admin" + password = "admin123" + + # Stop any existing container + subprocess.run(["docker", "rm", "-f", AGH_CONTAINER], capture_output=True) + + # Start AdGuardHome container (no config mount - will use install API) + result = subprocess.run([ + "docker", "run", "-d", + "--name", AGH_CONTAINER, + "-p", f"{AGH_TEST_PORT}:3000", + "--cap-add=NET_ADMIN", + "adguard/adguardhome", + ], capture_output=True, text=True) + + if result.returncode != 0: + pytest.skip(f"Could not start AdGuardHome Docker: {result.stderr}") + + # Wait for setup wizard to be available + deadline = time.time() + 30 + setup_ready = False + while time.time() < deadline: + try: + r = requests.get(f"http://localhost:{AGH_TEST_PORT}/control/install/get_addresses", + timeout=2) + if r.status_code == 200: + setup_ready = True + break + except requests.exceptions.ConnectionError: + pass + time.sleep(1) + + if not setup_ready: + subprocess.run(["docker", "rm", "-f", AGH_CONTAINER], capture_output=True) + pytest.skip("AdGuardHome setup wizard not reachable in time") + + # Run setup wizard + if not _configure_adguardhome(AGH_TEST_PORT, username, password): + subprocess.run(["docker", "rm", "-f", AGH_CONTAINER], capture_output=True) + pytest.skip("Could not configure AdGuardHome via install API") + + # Wait for configured instance to be ready + if not _wait_for_adguardhome(AGH_TEST_PORT, timeout=20): + subprocess.run(["docker", "rm", "-f", AGH_CONTAINER], capture_output=True) + pytest.skip("AdGuardHome not ready after configuration") + + print(f"\n AdGuardHome running at localhost:{AGH_TEST_PORT} (admin/admin123)") + + yield {"host": AGH_TEST_HOST, "port": AGH_TEST_PORT, + "username": username, "password": password} + + subprocess.run(["docker", "rm", "-f", AGH_CONTAINER], capture_output=True) + + +# --------------------------------------------------------------------------- +# Subprocess tests (no AdGuardHome needed) +# --------------------------------------------------------------------------- + +class TestCLISubprocess: + CLI_BASE = _resolve_cli("cli-anything-adguardhome") + + def _run(self, args: list[str], check: bool = True, env: dict | None = None) -> subprocess.CompletedProcess: + run_env = os.environ.copy() + if env: + run_env.update(env) + return subprocess.run( + self.CLI_BASE + args, + capture_output=True, text=True, + check=check, + env=run_env, + ) + + def test_help(self): + result = self._run(["--help"]) + assert result.returncode == 0 + assert "adguardhome" in result.stdout.lower() or "Usage" in result.stdout + + def test_config_show_json(self): + result = self._run(["--json", "config", "show"]) + assert result.returncode == 0 + data = json.loads(result.stdout) + assert "host" in data + assert "port" in data + + def test_config_show_default_host(self): + result = self._run(["--json", "config", "show"]) + data = json.loads(result.stdout) + assert data["host"] == "localhost" + assert data["port"] == 3000 + + def test_help_subcommands_listed(self): + result = self._run(["--help"]) + assert "filter" in result.stdout + assert "server" in result.stdout + assert "stats" in result.stdout + + def test_filter_help(self): + result = self._run(["filter", "--help"]) + assert result.returncode == 0 + assert "list" in result.stdout + + def test_rewrite_help(self): + result = self._run(["rewrite", "--help"]) + assert result.returncode == 0 + + def test_blocking_help(self): + result = self._run(["blocking", "--help"]) + assert result.returncode == 0 + + +# --------------------------------------------------------------------------- +# Docker E2E tests +# --------------------------------------------------------------------------- + +class TestDockerE2E: + CLI_BASE = _resolve_cli("cli-anything-adguardhome") + + def _run_agh(self, args: list[str], agh: dict, check: bool = True) -> subprocess.CompletedProcess: + env = os.environ.copy() + env["AGH_HOST"] = agh["host"] + env["AGH_PORT"] = str(agh["port"]) + env["AGH_USERNAME"] = agh["username"] + env["AGH_PASSWORD"] = agh["password"] + return subprocess.run( + self.CLI_BASE + args, + capture_output=True, text=True, + check=check, + env=env, + ) + + def test_server_status_json(self, agh_docker): + result = self._run_agh(["--json", "server", "status"], agh_docker) + assert result.returncode == 0 + data = json.loads(result.stdout) + print(f"\n Server status: {data}") + assert isinstance(data, dict) + + def test_filter_list_json(self, agh_docker): + result = self._run_agh(["--json", "filter", "list"], agh_docker) + assert result.returncode == 0 + data = json.loads(result.stdout) + print(f"\n Filters: {data}") + assert "filters" in data or isinstance(data, dict) + + def test_rewrite_lifecycle(self, agh_docker): + """Add rewrite, verify in list, remove, verify gone.""" + # Add + add_result = self._run_agh([ + "--json", "rewrite", "add", + "--domain", "test-cli.local", "--answer", "10.0.0.99" + ], agh_docker) + assert add_result.returncode == 0 + print(f"\n Rewrite add: {add_result.stdout.strip()}") + + # List and verify + list_result = self._run_agh(["--json", "rewrite", "list"], agh_docker) + assert list_result.returncode == 0 + rewrites = json.loads(list_result.stdout) + print(f"\n Rewrites: {rewrites}") + domains = [r.get("domain") for r in (rewrites if isinstance(rewrites, list) else [])] + assert "test-cli.local" in domains + + # Remove + rm_result = self._run_agh([ + "--json", "rewrite", "remove", + "--domain", "test-cli.local", "--answer", "10.0.0.99" + ], agh_docker) + assert rm_result.returncode == 0 + + # Verify removed + list_result2 = self._run_agh(["--json", "rewrite", "list"], agh_docker) + rewrites2 = json.loads(list_result2.stdout) + domains2 = [r.get("domain") for r in (rewrites2 if isinstance(rewrites2, list) else [])] + assert "test-cli.local" not in domains2 + print(f"\n Rewrite lifecycle: PASS") + + def test_stats_show_json(self, agh_docker): + result = self._run_agh(["--json", "stats", "show"], agh_docker) + assert result.returncode == 0 + data = json.loads(result.stdout) + print(f"\n Stats keys: {list(data.keys()) if isinstance(data, dict) else 'list'}") + assert isinstance(data, dict) + + def test_config_test(self, agh_docker): + result = self._run_agh(["--json", "config", "test"], agh_docker) + assert result.returncode == 0 + data = json.loads(result.stdout) + print(f"\n Config test: {data}") + assert data.get("connected") is True diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/utils/__init__.py b/adguardhome/agent-harness/cli_anything/adguardhome/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/utils/adguardhome_backend.py b/adguardhome/agent-harness/cli_anything/adguardhome/utils/adguardhome_backend.py new file mode 100644 index 000000000..49b83e16c --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/utils/adguardhome_backend.py @@ -0,0 +1,70 @@ +"""AdGuardHome HTTP API client - wraps all REST calls to the real AdGuardHome service.""" + +from typing import Any + +import requests + + +class AdGuardHomeClient: + """HTTP client for the AdGuardHome REST API.""" + + def __init__(self, host: str = "localhost", port: int = 3000, + username: str = "", password: str = "", https: bool = False): + scheme = "https" if https else "http" + # Auto-detect HTTPS for standard ports + if port == 443: + scheme = "https" + self.base_url = f"{scheme}://{host}:{port}/control" if port not in (80, 443) else f"{scheme}://{host}/control" + self.host = host + self.port = port + self.session = requests.Session() + if username or password: + self.session.auth = (username, password) + self.session.headers.update({"Content-Type": "application/json"}) + + def _url(self, path: str) -> str: + return f"{self.base_url}/{path.lstrip('/')}" + + def _handle_response(self, resp: requests.Response) -> Any: + if not resp.content: + return {} + try: + return resp.json() + except ValueError: + return resp.text + + def _connection_error(self, e: Exception) -> RuntimeError: + return RuntimeError( + f"Cannot connect to AdGuardHome at {self.base_url}.\n" + f"Ensure AdGuardHome is running and accessible.\n" + f"Install: curl -s -S -L https://raw.githubusercontent.com/AdguardTeam/AdGuardHome/master/scripts/install.sh | sh -s -- -v\n" + f"Or Docker: docker run --name adguardhome -p {self.port}:{self.port} adguard/adguardhome\n" + f"Error: {e}" + ) + + def get(self, path: str, params: dict | None = None) -> Any: + """GET request - returns deserialized JSON or raw text.""" + try: + resp = self.session.get(self._url(path), params=params, timeout=10) + resp.raise_for_status() + return self._handle_response(resp) + except requests.exceptions.ConnectionError as e: + raise self._connection_error(e) + + def post(self, path: str, data: Any = None) -> Any: + """POST request - sends JSON body, returns deserialized response.""" + try: + if isinstance(data, (dict, list)): + resp = self.session.post(self._url(path), json=data, timeout=10) + elif isinstance(data, str): + resp = self.session.post( + self._url(path), data=data.encode(), + headers={**dict(self.session.headers), "Content-Type": "text/plain"}, + timeout=10, + ) + else: + resp = self.session.post(self._url(path), timeout=10) + resp.raise_for_status() + return self._handle_response(resp) + except requests.exceptions.ConnectionError as e: + raise self._connection_error(e) diff --git a/adguardhome/agent-harness/cli_anything/adguardhome/utils/repl_skin.py b/adguardhome/agent-harness/cli_anything/adguardhome/utils/repl_skin.py new file mode 100644 index 000000000..47260bebd --- /dev/null +++ b/adguardhome/agent-harness/cli_anything/adguardhome/utils/repl_skin.py @@ -0,0 +1,498 @@ +"""cli-anything REPL Skin — Unified terminal interface for all CLI harnesses. + +Copy this file into your CLI package at: + cli_anything//utils/repl_skin.py + +Usage: + from cli_anything..utils.repl_skin import ReplSkin + + skin = ReplSkin("shotcut", version="1.0.0") + skin.print_banner() + prompt_text = skin.prompt(project_name="my_video.mlt", modified=True) + skin.success("Project saved") + skin.error("File not found") + skin.warning("Unsaved changes") + skin.info("Processing 24 clips...") + skin.status("Track 1", "3 clips, 00:02:30") + skin.table(headers, rows) + skin.print_goodbye() +""" + +import os +import sys + +# ── ANSI color codes (no external deps for core styling) ────────────── + +_RESET = "\033[0m" +_BOLD = "\033[1m" +_DIM = "\033[2m" +_ITALIC = "\033[3m" +_UNDERLINE = "\033[4m" + +# Brand colors +_CYAN = "\033[38;5;80m" # cli-anything brand cyan +_CYAN_BG = "\033[48;5;80m" +_WHITE = "\033[97m" +_GRAY = "\033[38;5;245m" +_DARK_GRAY = "\033[38;5;240m" +_LIGHT_GRAY = "\033[38;5;250m" + +# Software accent colors — each software gets a unique accent +_ACCENT_COLORS = { + "gimp": "\033[38;5;214m", # warm orange + "blender": "\033[38;5;208m", # deep orange + "inkscape": "\033[38;5;39m", # bright blue + "audacity": "\033[38;5;33m", # navy blue + "libreoffice": "\033[38;5;40m", # green + "obs_studio": "\033[38;5;55m", # purple + "kdenlive": "\033[38;5;69m", # slate blue + "shotcut": "\033[38;5;35m", # teal green +} +_DEFAULT_ACCENT = "\033[38;5;75m" # default sky blue + +# Status colors +_GREEN = "\033[38;5;78m" +_YELLOW = "\033[38;5;220m" +_RED = "\033[38;5;196m" +_BLUE = "\033[38;5;75m" +_MAGENTA = "\033[38;5;176m" + +# ── Brand icon ──────────────────────────────────────────────────────── + +# The cli-anything icon: a small colored diamond/chevron mark +_ICON = f"{_CYAN}{_BOLD}◆{_RESET}" +_ICON_SMALL = f"{_CYAN}▸{_RESET}" + +# ── Box drawing characters ──────────────────────────────────────────── + +_H_LINE = "─" +_V_LINE = "│" +_TL = "╭" +_TR = "╮" +_BL = "╰" +_BR = "╯" +_T_DOWN = "┬" +_T_UP = "┴" +_T_RIGHT = "├" +_T_LEFT = "┤" +_CROSS = "┼" + + +def _strip_ansi(text: str) -> str: + """Remove ANSI escape codes for length calculation.""" + import re + return re.sub(r"\033\[[^m]*m", "", text) + + +def _visible_len(text: str) -> int: + """Get visible length of text (excluding ANSI codes).""" + return len(_strip_ansi(text)) + + +class ReplSkin: + """Unified REPL skin for cli-anything CLIs. + + Provides consistent branding, prompts, and message formatting + across all CLI harnesses built with the cli-anything methodology. + """ + + def __init__(self, software: str, version: str = "1.0.0", + history_file: str | None = None): + """Initialize the REPL skin. + + Args: + software: Software name (e.g., "gimp", "shotcut", "blender"). + version: CLI version string. + history_file: Path for persistent command history. + Defaults to ~/.cli-anything-/history + """ + self.software = software.lower().replace("-", "_") + self.display_name = software.replace("_", " ").title() + self.version = version + self.accent = _ACCENT_COLORS.get(self.software, _DEFAULT_ACCENT) + + # History file + if history_file is None: + from pathlib import Path + hist_dir = Path.home() / f".cli-anything-{self.software}" + hist_dir.mkdir(parents=True, exist_ok=True) + self.history_file = str(hist_dir / "history") + else: + self.history_file = history_file + + # Detect terminal capabilities + self._color = self._detect_color_support() + + def _detect_color_support(self) -> bool: + """Check if terminal supports color.""" + if os.environ.get("NO_COLOR"): + return False + if os.environ.get("CLI_ANYTHING_NO_COLOR"): + return False + if not hasattr(sys.stdout, "isatty"): + return False + return sys.stdout.isatty() + + def _c(self, code: str, text: str) -> str: + """Apply color code if colors are supported.""" + if not self._color: + return text + return f"{code}{text}{_RESET}" + + # ── Banner ──────────────────────────────────────────────────────── + + def print_banner(self): + """Print the startup banner with branding.""" + inner = 54 + + def _box_line(content: str) -> str: + """Wrap content in box drawing, padding to inner width.""" + pad = inner - _visible_len(content) + vl = self._c(_DARK_GRAY, _V_LINE) + return f"{vl}{content}{' ' * max(0, pad)}{vl}" + + top = self._c(_DARK_GRAY, f"{_TL}{_H_LINE * inner}{_TR}") + bot = self._c(_DARK_GRAY, f"{_BL}{_H_LINE * inner}{_BR}") + + # Title: ◆ cli-anything · Shotcut + icon = self._c(_CYAN + _BOLD, "◆") + brand = self._c(_CYAN + _BOLD, "cli-anything") + dot = self._c(_DARK_GRAY, "·") + name = self._c(self.accent + _BOLD, self.display_name) + title = f" {icon} {brand} {dot} {name}" + + ver = f" {self._c(_DARK_GRAY, f' v{self.version}')}" + tip = f" {self._c(_DARK_GRAY, ' Type help for commands, quit to exit')}" + empty = "" + + print(top) + print(_box_line(title)) + print(_box_line(ver)) + print(_box_line(empty)) + print(_box_line(tip)) + print(bot) + print() + + # ── Prompt ──────────────────────────────────────────────────────── + + def prompt(self, project_name: str = "", modified: bool = False, + context: str = "") -> str: + """Build a styled prompt string for prompt_toolkit or input(). + + Args: + project_name: Current project name (empty if none open). + modified: Whether the project has unsaved changes. + context: Optional extra context to show in prompt. + + Returns: + Formatted prompt string. + """ + parts = [] + + # Icon + if self._color: + parts.append(f"{_CYAN}◆{_RESET} ") + else: + parts.append("> ") + + # Software name + parts.append(self._c(self.accent + _BOLD, self.software)) + + # Project context + if project_name or context: + ctx = context or project_name + mod = "*" if modified else "" + parts.append(f" {self._c(_DARK_GRAY, '[')}") + parts.append(self._c(_LIGHT_GRAY, f"{ctx}{mod}")) + parts.append(self._c(_DARK_GRAY, ']')) + + parts.append(self._c(_GRAY, " ❯ ")) + + return "".join(parts) + + def prompt_tokens(self, project_name: str = "", modified: bool = False, + context: str = ""): + """Build prompt_toolkit formatted text tokens for the prompt. + + Use with prompt_toolkit's FormattedText for proper ANSI handling. + + Returns: + list of (style, text) tuples for prompt_toolkit. + """ + accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff") + tokens = [] + + tokens.append(("class:icon", "◆ ")) + tokens.append(("class:software", self.software)) + + if project_name or context: + ctx = context or project_name + mod = "*" if modified else "" + tokens.append(("class:bracket", " [")) + tokens.append(("class:context", f"{ctx}{mod}")) + tokens.append(("class:bracket", "]")) + + tokens.append(("class:arrow", " ❯ ")) + + return tokens + + def get_prompt_style(self): + """Get a prompt_toolkit Style object matching the skin. + + Returns: + prompt_toolkit.styles.Style + """ + try: + from prompt_toolkit.styles import Style + except ImportError: + return None + + accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff") + + return Style.from_dict({ + "icon": "#5fdfdf bold", # cyan brand color + "software": f"{accent_hex} bold", + "bracket": "#585858", + "context": "#bcbcbc", + "arrow": "#808080", + # Completion menu + "completion-menu.completion": "bg:#303030 #bcbcbc", + "completion-menu.completion.current": f"bg:{accent_hex} #000000", + "completion-menu.meta.completion": "bg:#303030 #808080", + "completion-menu.meta.completion.current": f"bg:{accent_hex} #000000", + # Auto-suggest + "auto-suggest": "#585858", + # Bottom toolbar + "bottom-toolbar": "bg:#1c1c1c #808080", + "bottom-toolbar.text": "#808080", + }) + + # ── Messages ────────────────────────────────────────────────────── + + def success(self, message: str): + """Print a success message with green checkmark.""" + icon = self._c(_GREEN + _BOLD, "✓") + print(f" {icon} {self._c(_GREEN, message)}") + + def error(self, message: str): + """Print an error message with red cross.""" + icon = self._c(_RED + _BOLD, "✗") + print(f" {icon} {self._c(_RED, message)}", file=sys.stderr) + + def warning(self, message: str): + """Print a warning message with yellow triangle.""" + icon = self._c(_YELLOW + _BOLD, "⚠") + print(f" {icon} {self._c(_YELLOW, message)}") + + def info(self, message: str): + """Print an info message with blue dot.""" + icon = self._c(_BLUE, "●") + print(f" {icon} {self._c(_LIGHT_GRAY, message)}") + + def hint(self, message: str): + """Print a subtle hint message.""" + print(f" {self._c(_DARK_GRAY, message)}") + + def section(self, title: str): + """Print a section header.""" + print() + print(f" {self._c(self.accent + _BOLD, title)}") + print(f" {self._c(_DARK_GRAY, _H_LINE * len(title))}") + + # ── Status display ──────────────────────────────────────────────── + + def status(self, label: str, value: str): + """Print a key-value status line.""" + lbl = self._c(_GRAY, f" {label}:") + val = self._c(_WHITE, f" {value}") + print(f"{lbl}{val}") + + def status_block(self, items: dict[str, str], title: str = ""): + """Print a block of status key-value pairs. + + Args: + items: Dict of label -> value pairs. + title: Optional title for the block. + """ + if title: + self.section(title) + + max_key = max(len(k) for k in items) if items else 0 + for label, value in items.items(): + lbl = self._c(_GRAY, f" {label:<{max_key}}") + val = self._c(_WHITE, f" {value}") + print(f"{lbl}{val}") + + def progress(self, current: int, total: int, label: str = ""): + """Print a simple progress indicator. + + Args: + current: Current step number. + total: Total number of steps. + label: Optional label for the progress. + """ + pct = int(current / total * 100) if total > 0 else 0 + bar_width = 20 + filled = int(bar_width * current / total) if total > 0 else 0 + bar = "█" * filled + "░" * (bar_width - filled) + text = f" {self._c(_CYAN, bar)} {self._c(_GRAY, f'{pct:3d}%')}" + if label: + text += f" {self._c(_LIGHT_GRAY, label)}" + print(text) + + # ── Table display ───────────────────────────────────────────────── + + def table(self, headers: list[str], rows: list[list[str]], + max_col_width: int = 40): + """Print a formatted table with box-drawing characters. + + Args: + headers: Column header strings. + rows: List of rows, each a list of cell strings. + max_col_width: Maximum column width before truncation. + """ + if not headers: + return + + # Calculate column widths + col_widths = [min(len(h), max_col_width) for h in headers] + for row in rows: + for i, cell in enumerate(row): + if i < len(col_widths): + col_widths[i] = min( + max(col_widths[i], len(str(cell))), max_col_width + ) + + def pad(text: str, width: int) -> str: + t = str(text)[:width] + return t + " " * (width - len(t)) + + # Header + header_cells = [ + self._c(_CYAN + _BOLD, pad(h, col_widths[i])) + for i, h in enumerate(headers) + ] + sep = self._c(_DARK_GRAY, f" {_V_LINE} ") + header_line = f" {sep.join(header_cells)}" + print(header_line) + + # Separator + sep_parts = [self._c(_DARK_GRAY, _H_LINE * w) for w in col_widths] + sep_line = self._c(_DARK_GRAY, f" {'───'.join([_H_LINE * w for w in col_widths])}") + print(sep_line) + + # Rows + for row in rows: + cells = [] + for i, cell in enumerate(row): + if i < len(col_widths): + cells.append(self._c(_LIGHT_GRAY, pad(str(cell), col_widths[i]))) + row_sep = self._c(_DARK_GRAY, f" {_V_LINE} ") + print(f" {row_sep.join(cells)}") + + # ── Help display ────────────────────────────────────────────────── + + def help(self, commands: dict[str, str]): + """Print a formatted help listing. + + Args: + commands: Dict of command -> description pairs. + """ + self.section("Commands") + max_cmd = max(len(c) for c in commands) if commands else 0 + for cmd, desc in commands.items(): + cmd_styled = self._c(self.accent, f" {cmd:<{max_cmd}}") + desc_styled = self._c(_GRAY, f" {desc}") + print(f"{cmd_styled}{desc_styled}") + print() + + # ── Goodbye ─────────────────────────────────────────────────────── + + def print_goodbye(self): + """Print a styled goodbye message.""" + print(f"\n {_ICON_SMALL} {self._c(_GRAY, 'Goodbye!')}\n") + + # ── Prompt toolkit session factory ──────────────────────────────── + + def create_prompt_session(self): + """Create a prompt_toolkit PromptSession with skin styling. + + Returns: + A configured PromptSession, or None if prompt_toolkit unavailable. + """ + try: + from prompt_toolkit import PromptSession + from prompt_toolkit.history import FileHistory + from prompt_toolkit.auto_suggest import AutoSuggestFromHistory + from prompt_toolkit.formatted_text import FormattedText + + style = self.get_prompt_style() + + session = PromptSession( + history=FileHistory(self.history_file), + auto_suggest=AutoSuggestFromHistory(), + style=style, + enable_history_search=True, + ) + return session + except ImportError: + return None + + def get_input(self, pt_session, project_name: str = "", + modified: bool = False, context: str = "") -> str: + """Get input from user using prompt_toolkit or fallback. + + Args: + pt_session: A prompt_toolkit PromptSession (or None). + project_name: Current project name. + modified: Whether project has unsaved changes. + context: Optional context string. + + Returns: + User input string (stripped). + """ + if pt_session is not None: + from prompt_toolkit.formatted_text import FormattedText + tokens = self.prompt_tokens(project_name, modified, context) + return pt_session.prompt(FormattedText(tokens)).strip() + else: + raw_prompt = self.prompt(project_name, modified, context) + return input(raw_prompt).strip() + + # ── Toolbar builder ─────────────────────────────────────────────── + + def bottom_toolbar(self, items: dict[str, str]): + """Create a bottom toolbar callback for prompt_toolkit. + + Args: + items: Dict of label -> value pairs to show in toolbar. + + Returns: + A callable that returns FormattedText for the toolbar. + """ + def toolbar(): + from prompt_toolkit.formatted_text import FormattedText + parts = [] + for i, (k, v) in enumerate(items.items()): + if i > 0: + parts.append(("class:bottom-toolbar.text", " │ ")) + parts.append(("class:bottom-toolbar.text", f" {k}: ")) + parts.append(("class:bottom-toolbar", v)) + return FormattedText(parts) + return toolbar + + +# ── ANSI 256-color to hex mapping (for prompt_toolkit styles) ───────── + +_ANSI_256_TO_HEX = { + "\033[38;5;33m": "#0087ff", # audacity navy blue + "\033[38;5;35m": "#00af5f", # shotcut teal + "\033[38;5;39m": "#00afff", # inkscape bright blue + "\033[38;5;40m": "#00d700", # libreoffice green + "\033[38;5;55m": "#5f00af", # obs purple + "\033[38;5;69m": "#5f87ff", # kdenlive slate blue + "\033[38;5;75m": "#5fafff", # default sky blue + "\033[38;5;80m": "#5fd7d7", # brand cyan + "\033[38;5;208m": "#ff8700", # blender deep orange + "\033[38;5;214m": "#ffaf00", # gimp warm orange +} diff --git a/adguardhome/agent-harness/setup.py b/adguardhome/agent-harness/setup.py new file mode 100644 index 000000000..ecd83e26d --- /dev/null +++ b/adguardhome/agent-harness/setup.py @@ -0,0 +1,19 @@ +from setuptools import setup, find_namespace_packages + +setup( + name="cli-anything-adguardhome", + version="1.0.0", + description="CLI harness for AdGuardHome - control your ad blocker from the command line", + packages=find_namespace_packages(include=["cli_anything.*"]), + install_requires=[ + "click>=8.0.0", + "prompt-toolkit>=3.0.0", + "requests>=2.28.0", + ], + entry_points={ + "console_scripts": [ + "cli-anything-adguardhome=cli_anything.adguardhome.adguardhome_cli:main", + ], + }, + python_requires=">=3.10", +)