663 lines
25 KiB
Python
663 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""Developer smoke/integration tests for GREE Controller API 0.13.7.
|
|
|
|
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.13.7"
|
|
|
|
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",
|
|
"/api/integrations/home-assistant/snapshot",
|
|
]
|
|
|
|
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",
|
|
"/api/integrations/home-assistant/snapshot",
|
|
}
|
|
|
|
BOOTSTRAP_SETTINGS_SECTIONS = {
|
|
"application", "gree", "history", "influxdb", "notifications", "night", "home_assistant", "debug"
|
|
}
|
|
|
|
|
|
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 required 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_bootstrap_contract(client: ApiClient) -> str:
|
|
resp = client.get("/api/bootstrap")
|
|
assert_status(resp, 200)
|
|
data = assert_json_object(resp)
|
|
required = {
|
|
"devices", "zones", "groups", "schedules", "automations", "flows",
|
|
"access_tokens", "settings", "house", "outdoor_temperature",
|
|
"control_plan", "control_plan_revision", "system",
|
|
}
|
|
missing = sorted(required - set(data))
|
|
if missing:
|
|
raise TestFailure(f"bootstrap missing fields: {missing}")
|
|
settings = data.get("settings")
|
|
if not isinstance(settings, dict):
|
|
raise TestFailure("bootstrap.settings is not an object")
|
|
missing_sections = sorted(BOOTSTRAP_SETTINGS_SECTIONS - set(settings))
|
|
if missing_sections:
|
|
raise TestFailure(f"bootstrap.settings missing sections: {missing_sections}")
|
|
forbidden = {
|
|
"influxdb": {"password", "token"},
|
|
"home_assistant": {"token"},
|
|
"notifications": {"pushover_app_token", "pushover_user_key", "slack_webhook_url", "discord_webhook_url"},
|
|
}
|
|
for section, fields in forbidden.items():
|
|
leaked = sorted(fields & set(settings.get(section, {})))
|
|
if leaked:
|
|
raise TestFailure(f"bootstrap.settings.{section} leaks secret fields: {leaked}")
|
|
return f"settings_sections={len(BOOTSTRAP_SETTINGS_SECTIONS)}, control_plan_revision={data.get('control_plan_revision')}"
|
|
|
|
|
|
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 path == "/api/integrations/home-assistant/snapshot":
|
|
if not isinstance(data, dict):
|
|
raise TestFailure(f"snapshot is not an object: {type(data).__name__}")
|
|
required = {"devices", "control_plan", "control_plan_revision", "groups"}
|
|
missing = sorted(required - set(data))
|
|
if missing:
|
|
raise TestFailure(f"snapshot missing keys: {missing}")
|
|
if not isinstance(data["devices"], list) or not isinstance(data["groups"], list):
|
|
raise TestFailure("snapshot devices/groups must be arrays")
|
|
if not isinstance(data["control_plan"], dict):
|
|
raise TestFailure("snapshot control_plan must be an object")
|
|
if not isinstance(data["control_plan_revision"], int):
|
|
raise TestFailure("snapshot control_plan_revision must be an integer")
|
|
return f"devices={len(data['devices'])}, groups={len(data['groups'])}, revision={data['control_plan_revision']}"
|
|
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.13.7 contract", lambda: test_openapi(client, args.expected_version))
|
|
runner.run("bootstrap snapshot contract", lambda: test_bootstrap_contract(client))
|
|
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.13.7.",
|
|
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())
|