gru/gree-controller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.12.0-api

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-09-03 13:38:27 +02:00
parent abcd84f5ef
commit 351ea765fa
3 changed files with 636 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/FILE_MANIFEST.sha256
-13
View File
@@ -1,13 +0,0 @@
bb89bac237e750e9b1bf73761d7df97a6b81853091615878c03f13d7b6399aa7 ./README.md
5bc736c7bc76ca80aaa406bb171d2aa91baf4c3aa8695dce0e09b888b6ab3146 ./common.sh
6403786610ee6d2f628193c25aee0dd058d62e904aa1a31d5f62fdaae0e94b4f ./configure-gree-network.sh
1d6e14e26e49aa9d3527f30a23668bf8d9c48b67e6628ef686c3155c012155de ./dev.sh
14076104c042fba1284ebb07531a6c3ff972df1f9f5b18f70da18ab774efed27 ./generate_ha_migration.py
e4849261fd9ed1f01df96c0637c439c0c4eff8fa317b2918026167bba343af79 ./install-lxc.sh
bb7cd2c5b27c9dceec1d1d2846fbad9600df9c08e0ad07d13fcde97af533091c ./install.sh
e00d211e3885e30d7fed1e43b44e6fdad40a67019060156c0641816a93e3365f ./network-debug.sh
01952aa92b217f8eae2493b88870e2dec595100cd15c4d561ff11ae2b936c46f ./regenerate-sha.sh
81345b6a0b51736bdbc98fd23199b62e4c721b4e7437e02dab7ea79b97dff29a ./service.sh
7e8f558b284362f55f50bd7223a89754a2a4960e746a27035cc61180f4a2da6f ./smoke.sh
b50782b3742dfbf8a319c60571c968e93fdf8547db747c759edcffae68cb98bf ./update.sh
0790b7b388467d9e75b621188cca473e22d73e01bcfa74e92fa23f92b0695855 ./verify_flow_logic.py
scripts/api_dev_test.py
+611
View File
@@ -0,0 +1,611 @@
#!/usr/bin/env python3
"""Developer smoke/integration tests for GREE Controller API 0.12.0.
Default mode is read-only and safe to run against a real controller.
Use --settings-write to additionally round-trip all split settings resources and
exercise selected validation failures. This writes the same settings back and
therefore creates settings/events and can trigger normal settings side effects.
No third-party dependencies are required.
"""
from __future__ import annotations
import argparse
import copy
import json
import os
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Iterable, Optional
DEFAULT_BASE_URL = os.environ.get("GREE_API_URL", "http://127.0.0.1:8787")
DEFAULT_TOKEN = os.environ.get("GREE_API_TOKEN", "")
DEFAULT_EXPECTED_VERSION = "0.12.0"
SETTINGS_PATHS: dict[str, set[str]] = {
"/api/settings/application": {"simulator_enabled"},
"/api/settings/gree": {
"controller_id",
"poll_interval_seconds",
"zone_interval_seconds",
"discovery_timeout_ms",
"discovery_broadcast",
"suppress_device_beep",
"compressor_protection_enabled",
"compressor_protection_seconds",
},
"/api/settings/history": {"retention_days", "compaction_enabled", "event_retention_days"},
"/api/settings/influxdb": {
"enabled",
"version",
"url",
"database",
"username",
"password_configured",
"org",
"bucket",
"token_configured",
"history_threshold_days",
},
"/api/settings/notifications": {
"enabled",
"mode",
"provider",
"pushover_configured",
"slack_configured",
"discord_configured",
"cooldown_seconds",
"communication_failure_threshold",
"target_timeout_minutes",
"alert_types",
},
"/api/settings/night": {
"enabled", "start_time", "end_time", "max_fan_speed", "force_quiet", "use_native_sleep"
},
"/api/settings/home-assistant": {
"url",
"token_configured",
"default_entity_id",
"outdoor_entity_id",
"sensor_stale_after_seconds",
"allow_invalid_tls",
"sensor_aliases",
"flow_inputs",
"outdoor_assist_enabled",
},
"/api/settings/debug": {"overlay_enabled", "gree_frames"},
}
REMOVED_0120_PATHS = [
"/api/settings",
"/api/debug",
"/api/events/retention",
"/api/settings/export",
"/api/settings/import",
]
SAFE_GET_PATHS = [
"/api/bootstrap",
"/api/system/info",
"/api/devices",
"/api/zones",
"/api/groups",
"/api/schedules",
"/api/automations",
"/api/flows",
"/api/readings",
"/api/history",
"/api/control-plan",
"/api/events",
"/api/access-tokens",
"/api/configuration/export",
]
# These are administrator-authenticated restricted HA reads. They are safe but
# can legitimately fail if the server is configured to forbid this surface.
HA_SAFE_GET_PATHS = [
"/api/integrations/home-assistant/devices",
"/api/integrations/home-assistant/control-plan",
"/api/integrations/home-assistant/groups",
]
DETAIL_COLLECTIONS = [
("/api/devices", "/api/devices/{id}"),
("/api/zones", "/api/zones/{id}"),
("/api/groups", "/api/groups/{id}"),
("/api/schedules", "/api/schedules/{id}"),
("/api/automations", "/api/automations/{id}"),
("/api/flows", "/api/flows/{id}"),
]
EXPECTED_OPENAPI_PATHS = set(SETTINGS_PATHS) | {
"/api/configuration/export",
"/api/configuration/import",
}
class TestFailure(AssertionError):
pass
@dataclass
class HttpResponse:
status: int
headers: dict[str, str]
body: bytes
elapsed_ms: float
def text(self) -> str:
return self.body.decode("utf-8", errors="replace")
def json(self) -> Any:
try:
return json.loads(self.text())
except json.JSONDecodeError as exc:
raise TestFailure(f"response is not valid JSON: {exc}; body={self.text()[:300]!r}") from exc
class ApiClient:
def __init__(self, base_url: str, token: str, timeout: float, insecure: bool) -> None:
self.base_url = base_url.rstrip("/")
self.token = token.strip()
self.timeout = timeout
self.ssl_context: Optional[ssl.SSLContext] = None
if insecure:
self.ssl_context = ssl._create_unverified_context() # noqa: SLF001 - explicit developer option
def _url(self, path: str) -> str:
if not path.startswith("/"):
path = "/" + path
return self.base_url + path
def request(
self,
method: str,
path: str,
payload: Any = None,
*,
auth: bool = True,
headers: Optional[dict[str, str]] = None,
) -> HttpResponse:
req_headers = {"Accept": "application/json"}
if auth and self.token:
req_headers["Authorization"] = f"Bearer {self.token}"
if headers:
req_headers.update(headers)
data: Optional[bytes] = None
if payload is not None:
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req_headers["Content-Type"] = "application/json"
req = urllib.request.Request(
self._url(path),
data=data,
headers=req_headers,
method=method.upper(),
)
started = time.perf_counter()
try:
with urllib.request.urlopen(req, timeout=self.timeout, context=self.ssl_context) as resp:
body = resp.read()
status = resp.getcode()
response_headers = {k.lower(): v for k, v in resp.headers.items()}
except urllib.error.HTTPError as exc:
body = exc.read()
status = exc.code
response_headers = {k.lower(): v for k, v in exc.headers.items()}
except urllib.error.URLError as exc:
raise TestFailure(f"cannot connect to {self._url(path)}: {exc.reason}") from exc
elapsed_ms = (time.perf_counter() - started) * 1000.0
return HttpResponse(status, response_headers, body, elapsed_ms)
def get(self, path: str, *, auth: bool = True) -> HttpResponse:
return self.request("GET", path, auth=auth)
def put(self, path: str, payload: Any) -> HttpResponse:
return self.request("PUT", path, payload)
@dataclass
class Result:
name: str
status: str
detail: str = ""
elapsed_ms: float = 0.0
@dataclass
class Runner:
client: ApiClient
verbose: bool = False
results: list[Result] = field(default_factory=list)
def run(self, name: str, fn: Callable[[], Optional[str]]) -> None:
started = time.perf_counter()
try:
detail = fn() or ""
status = "PASS"
except SkipTest as exc:
status = "SKIP"
detail = str(exc)
except Exception as exc: # deliberate: one failed test must not stop the suite
status = "FAIL"
detail = str(exc)
elapsed_ms = (time.perf_counter() - started) * 1000.0
self.results.append(Result(name, status, detail, elapsed_ms))
marker = {"PASS": "+", "FAIL": "!", "SKIP": "-"}[status]
suffix = f" - {detail}" if detail and (self.verbose or status != "PASS") else ""
print(f"[{marker}] {status:<4} {name} ({elapsed_ms:.0f} ms){suffix}")
def summary(self) -> int:
counts = {name: sum(r.status == name for r in self.results) for name in ("PASS", "FAIL", "SKIP")}
print("\nSummary: " + ", ".join(f"{k}={v}" for k, v in counts.items()))
if counts["FAIL"]:
print("\nFailures:")
for result in self.results:
if result.status == "FAIL":
print(f" - {result.name}: {result.detail}")
return 1 if counts["FAIL"] else 0
class SkipTest(Exception):
pass
def assert_status(resp: HttpResponse, *expected: int) -> None:
if resp.status not in expected:
body = resp.text().strip().replace("\n", " ")[:500]
raise TestFailure(f"HTTP {resp.status}, expected {expected}; body={body!r}")
def assert_json_object(resp: HttpResponse) -> dict[str, Any]:
value = resp.json()
if not isinstance(value, dict):
raise TestFailure(f"expected JSON object, got {type(value).__name__}")
return value
def assert_json_collection(resp: HttpResponse) -> Any:
value = resp.json()
if not isinstance(value, (list, dict)):
raise TestFailure(f"expected JSON collection/object, got {type(value).__name__}")
return value
def first_item(value: Any) -> Optional[dict[str, Any]]:
if isinstance(value, list):
return value[0] if value and isinstance(value[0], dict) else None
if isinstance(value, dict):
# Handle APIs that wrap arrays in a named property.
for candidate in value.values():
if isinstance(candidate, list) and candidate and isinstance(candidate[0], dict):
return candidate[0]
return None
def convert_settings_view_to_update(path: str, view: dict[str, Any]) -> dict[str, Any]:
"""Build a PUT payload from a GET view without modifying stored secrets."""
payload = copy.deepcopy(view)
if path == "/api/settings/influxdb":
payload.pop("password_configured", None)
payload.pop("token_configured", None)
# Omitted Option fields preserve existing secrets.
elif path == "/api/settings/notifications":
payload.pop("pushover_configured", None)
payload.pop("slack_configured", None)
payload.pop("discord_configured", None)
elif path == "/api/settings/home-assistant":
payload.pop("token_configured", None)
return payload
def comparable_settings_view(path: str, view: dict[str, Any]) -> dict[str, Any]:
"""Normalize a settings view for equality checks after a no-op PUT."""
return copy.deepcopy(view)
def test_health(client: ApiClient, expected_version: str) -> str:
resp = client.get("/api/health", auth=False)
assert_status(resp, 200)
data = assert_json_object(resp)
if data.get("status") != "ok":
raise TestFailure(f"health.status={data.get('status')!r}, expected 'ok'")
if expected_version and data.get("version") != expected_version:
raise TestFailure(f"health.version={data.get('version')!r}, expected {expected_version!r}")
return f"version={data.get('version')}, control_ready={data.get('control_ready')}"
def test_openapi(client: ApiClient, expected_version: str) -> str:
resp = client.get("/api-docs/openapi.json", auth=False)
assert_status(resp, 200)
spec = assert_json_object(resp)
paths = spec.get("paths")
if not isinstance(paths, dict):
raise TestFailure("OpenAPI has no paths object")
missing = sorted(EXPECTED_OPENAPI_PATHS - set(paths))
if missing:
raise TestFailure(f"OpenAPI missing 0.12.0 paths: {missing}")
forbidden = sorted(set(REMOVED_0120_PATHS) & set(paths))
if forbidden:
raise TestFailure(f"OpenAPI still exposes removed paths: {forbidden}")
version = ((spec.get("info") or {}).get("version"))
if expected_version and version != expected_version:
raise TestFailure(f"OpenAPI info.version={version!r}, expected {expected_version!r}")
return f"paths={len(paths)}, version={version}"
def test_protected_auth(client: ApiClient) -> str:
if not client.token:
raise SkipTest("no token supplied; controller may be in trusted-LAN mode")
resp = client.get("/api/bootstrap", auth=False)
assert_status(resp, 401)
return "protected endpoint rejects missing token"
def test_safe_get(client: ApiClient, path: str) -> str:
resp = client.get(path)
assert_status(resp, 200)
content_type = resp.headers.get("content-type", "")
if "json" not in content_type.lower():
# Export is still expected to be JSON, but tolerate servers that return octet-stream attachment.
if path != "/api/configuration/export":
raise TestFailure(f"unexpected content-type {content_type!r}")
data = resp.json()
if path == "/api/configuration/export":
if not isinstance(data, dict):
raise TestFailure("configuration export is not a JSON object")
fmt = data.get("format_version")
if fmt != 3:
raise TestFailure(f"configuration export format_version={fmt!r}, expected 3")
return f"backup format_version={fmt}"
if not isinstance(data, (list, dict)):
raise TestFailure(f"unexpected JSON type {type(data).__name__}")
if isinstance(data, list):
return f"items={len(data)}"
return f"keys={len(data)}"
def test_ha_safe_get(client: ApiClient, path: str) -> str:
resp = client.get(path)
if not client.token:
# Restricted HA API always requires a token, even when the normal admin
# API runs in trusted-LAN mode.
assert_status(resp, 401)
return "restricted HA auth enforced (no token supplied)"
assert_status(resp, 200)
data = resp.json()
if not isinstance(data, (list, dict)):
raise TestFailure(f"unexpected JSON type {type(data).__name__}")
if isinstance(data, list):
return f"items={len(data)}"
return f"keys={len(data)}"
def test_settings_get(client: ApiClient, path: str, required: set[str]) -> str:
resp = client.get(path)
assert_status(resp, 200)
data = assert_json_object(resp)
missing = sorted(required - set(data))
if missing:
raise TestFailure(f"missing keys: {missing}")
# Secret values must never be returned by split GETs.
forbidden_secret_keys = {
"/api/settings/influxdb": {"password", "token"},
"/api/settings/notifications": {
"pushover_app_token", "pushover_user_key", "slack_webhook_url", "discord_webhook_url"
},
"/api/settings/home-assistant": {"token"},
}.get(path, set())
leaked = sorted(forbidden_secret_keys & set(data))
if leaked:
raise TestFailure(f"secret fields leaked in GET response: {leaked}")
return f"keys={len(data)}"
def test_removed_path(client: ApiClient, path: str) -> str:
# GET is enough to prove there is no compatibility alias. Some removed write-only
# paths may return 404 or 405 depending on router fallback/method handling.
resp = client.get(path)
if resp.status not in (404, 405):
raise TestFailure(f"removed endpoint still responds with HTTP {resp.status}")
return f"HTTP {resp.status}"
def test_detail_endpoint(client: ApiClient, collection_path: str, detail_template: str) -> str:
resp = client.get(collection_path)
assert_status(resp, 200)
collection = assert_json_collection(resp)
item = first_item(collection)
if item is None:
raise SkipTest("collection is empty")
item_id = item.get("id")
if not isinstance(item_id, str) or not item_id:
raise SkipTest("first item has no string id")
path = detail_template.replace("{id}", urllib.parse.quote(item_id, safe=""))
detail = client.get(path)
assert_status(detail, 200)
assert_json_object(detail)
return f"id={item_id}"
def test_flow_read_subresources(client: ApiClient) -> str:
resp = client.get("/api/flows")
assert_status(resp, 200)
item = first_item(assert_json_collection(resp))
if item is None or not isinstance(item.get("id"), str):
raise SkipTest("no Flow available")
flow_id = urllib.parse.quote(item["id"], safe="")
export_resp = client.get(f"/api/flows/{flow_id}/export")
assert_status(export_resp, 200)
export_resp.json()
logs_resp = client.get(f"/api/flows/{flow_id}/logs")
assert_status(logs_resp, 200)
logs_resp.json()
return f"id={item['id']}"
def test_settings_roundtrip(client: ApiClient, path: str) -> str:
before_resp = client.get(path)
assert_status(before_resp, 200)
before = assert_json_object(before_resp)
payload = convert_settings_view_to_update(path, before)
put_resp = client.put(path, payload)
assert_status(put_resp, 200)
put_view = assert_json_object(put_resp)
after_resp = client.get(path)
assert_status(after_resp, 200)
after = assert_json_object(after_resp)
if comparable_settings_view(path, put_view) != comparable_settings_view(path, after):
raise TestFailure(f"PUT response differs from following GET: put={put_view!r}, get={after!r}")
# No-op write should preserve the observable GET view, including *_configured flags.
if comparable_settings_view(path, before) != comparable_settings_view(path, after):
raise TestFailure(f"settings changed after no-op round-trip: before={before!r}, after={after!r}")
return "GET -> PUT(no-op) -> GET preserved view"
def expect_validation_error(client: ApiClient, path: str, mutate: Callable[[dict[str, Any]], None]) -> str:
before_resp = client.get(path)
assert_status(before_resp, 200)
before = assert_json_object(before_resp)
payload = convert_settings_view_to_update(path, before)
mutate(payload)
resp = client.put(path, payload)
assert_status(resp, 400)
error = assert_json_object(resp)
if not isinstance(error.get("error"), str) or not error["error"]:
raise TestFailure(f"400 response missing string error: {error!r}")
after_resp = client.get(path)
assert_status(after_resp, 200)
after = assert_json_object(after_resp)
if before != after:
raise TestFailure("settings changed despite rejected request")
return error["error"][:120]
def run_suite(args: argparse.Namespace) -> int:
client = ApiClient(args.base_url, args.token, args.timeout, args.insecure)
runner = Runner(client, verbose=args.verbose)
print(f"GREE Controller API test: {client.base_url}")
print(f"Expected version: {args.expected_version or '(any)'}")
print(f"Auth token: {'yes' if args.token else 'no'}")
print(f"Settings write tests: {'ENABLED' if args.settings_write else 'disabled'}\n")
runner.run("public health", lambda: test_health(client, args.expected_version))
runner.run("OpenAPI 0.12.0 contract", lambda: test_openapi(client, args.expected_version))
runner.run("protected API requires auth", lambda: test_protected_auth(client))
for path in SAFE_GET_PATHS:
runner.run(f"GET {path}", lambda path=path: test_safe_get(client, path))
for path in HA_SAFE_GET_PATHS:
runner.run(f"GET {path}", lambda path=path: test_ha_safe_get(client, path))
for path, required in SETTINGS_PATHS.items():
runner.run(f"split settings GET {path}", lambda path=path, required=required: test_settings_get(client, path, required))
for path in REMOVED_0120_PATHS:
runner.run(f"removed route {path}", lambda path=path: test_removed_path(client, path))
for collection, detail in DETAIL_COLLECTIONS:
runner.run(
f"detail read {detail}",
lambda collection=collection, detail=detail: test_detail_endpoint(client, collection, detail),
)
runner.run("Flow export/logs read", lambda: test_flow_read_subresources(client))
if args.settings_write:
print("\n--- settings write/validation tests ---")
for path in SETTINGS_PATHS:
runner.run(f"round-trip PUT {path}", lambda path=path: test_settings_roundtrip(client, path))
runner.run(
"validation: empty GREE controller_id -> 400",
lambda: expect_validation_error(client, "/api/settings/gree", lambda p: p.__setitem__("controller_id", "")),
)
runner.run(
"validation: invalid night start_time -> 400",
lambda: expect_validation_error(client, "/api/settings/night", lambda p: p.__setitem__("start_time", "99:99")),
)
runner.run(
"validation: invalid notification provider -> 400",
lambda: expect_validation_error(
client, "/api/settings/notifications", lambda p: p.__setitem__("provider", "invalid-provider")
),
)
runner.run(
"validation: invalid HA URL scheme -> 400",
lambda: expect_validation_error(
client, "/api/settings/home-assistant", lambda p: p.__setitem__("url", "ftp://invalid.local")
),
)
exit_code = runner.summary()
if args.report_json:
report_path = Path(args.report_json)
report_path.parent.mkdir(parents=True, exist_ok=True)
report = {
"base_url": client.base_url,
"expected_version": args.expected_version,
"settings_write": bool(args.settings_write),
"exit_code": exit_code,
"results": [r.__dict__ for r in runner.results],
}
report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
print(f"JSON report: {report_path}")
return exit_code
def parse_args(argv: Optional[Iterable[str]] = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Developer smoke/integration tests for GREE Controller API 0.12.0.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""Examples:
python3 scripts/api_dev_test.py
python3 scripts/api_dev_test.py --base-url http://192.168.1.50:8787
GREE_API_TOKEN=secret python3 scripts/api_dev_test.py --settings-write
python3 scripts/api_dev_test.py --settings-write --report-json /tmp/api-report.json
Default mode performs only reads. --settings-write re-saves the current settings
and intentionally generates normal settings events/side effects. It does NOT send
house/device/zone control commands.
""",
)
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help=f"API base URL (default: {DEFAULT_BASE_URL})")
parser.add_argument("--token", default=DEFAULT_TOKEN, help="administrator app token; or set GREE_API_TOKEN")
parser.add_argument(
"--expected-version",
default=DEFAULT_EXPECTED_VERSION,
help=f"expected health/OpenAPI version; use empty string to disable (default: {DEFAULT_EXPECTED_VERSION})",
)
parser.add_argument("--timeout", type=float, default=8.0, help="HTTP timeout in seconds (default: 8)")
parser.add_argument("--insecure", action="store_true", help="disable TLS certificate verification for HTTPS dev instances")
parser.add_argument(
"--settings-write",
action="store_true",
help="round-trip all 8 settings PUT endpoints and test selected 400 validation paths",
)
parser.add_argument("--report-json", help="write machine-readable test results to this JSON file")
parser.add_argument("-v", "--verbose", action="store_true", help="show PASS details")
return parser.parse_args(argv)
def main() -> int:
return run_suite(parse_args())
if __name__ == "__main__":
sys.exit(main())