mirror of
https://github.com/HKUDS/CLI-Anything.git
synced 2026-08-28 23:27:04 +08:00
feat: add AdGuardHome CLI harness (cli-anything-adguardhome)
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 <noreply@anthropic.com>
This commit is contained in:
pyxl
@@ -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__/
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
.venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
*.egg-info/
|
||||
.pytest_cache/
|
||||
.fastembed_cache/
|
||||
.leann/
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
@@ -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://<host>:<port>/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
|
||||
@@ -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
|
||||
```
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Enable python -m cli_anything.adguardhome"""
|
||||
from cli_anything.adguardhome.adguardhome_cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -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"])
|
||||
@@ -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})
|
||||
@@ -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})
|
||||
@@ -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,
|
||||
})
|
||||
@@ -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})
|
||||
@@ -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")
|
||||
@@ -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
|
||||
@@ -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})
|
||||
@@ -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")
|
||||
@@ -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,
|
||||
}
|
||||
@@ -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})
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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/<software>/utils/repl_skin.py
|
||||
|
||||
Usage:
|
||||
from cli_anything.<software>.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-<software>/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
|
||||
}
|
||||
@@ -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",
|
||||
)
|
||||
Reference in New Issue
Block a user