|
| 1 | +"""Real HA archive/download path with the real short-ASCII evidence collector.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from io import BytesIO |
| 6 | +import json |
| 7 | +from pathlib import Path |
| 8 | +import zipfile |
| 9 | + |
| 10 | +import pytest |
| 11 | +from pytest_homeassistant_custom_component.common import MockConfigEntry |
| 12 | + |
| 13 | +from custom_components.eybond_local.const import DOMAIN |
| 14 | +from custom_components.eybond_local.drivers.eybond_short_ascii import EybondShortAsciiDriver |
| 15 | +from custom_components.eybond_local.link_models import EybondLinkRoute |
| 16 | +from custom_components.eybond_local.models import CollectorInfo, ProbeTarget, RuntimeSnapshot |
| 17 | +from custom_components.eybond_local.payload.short_ascii import READ_COMMANDS, parse_q1, parse_rb |
| 18 | +from synthetic import SYNTHETIC_COLLECTOR_IP, SYNTHETIC_COLLECTOR_PN, SYNTHETIC_SERVER_IP |
| 19 | + |
| 20 | + |
| 21 | +@pytest.mark.parametrize("outcome", ["ok", "partial", "invalid", "offline"]) |
| 22 | +async def test_support_archive_preserves_read_evidence_through_ha_download( |
| 23 | + hass, hass_client_no_auth, fake_runtime, monkeypatch, outcome, |
| 24 | +) -> None: |
| 25 | + from conftest import FakeRuntimeManager |
| 26 | + |
| 27 | + body = b"230.0 04 03 115.0 013 60.0 13.2 35.0 1000010" + bytes(3) |
| 28 | + rb = bytearray(37) |
| 29 | + rb[:2] = (520).to_bytes(2, "big") |
| 30 | + rb[2] = 80 |
| 31 | + responses = { |
| 32 | + "MP": b"\x01" + bytes(range(36)) + b"\r", |
| 33 | + "Q1": b"\x01" + body + sum(body).to_bytes(2, "big") + b"\r", |
| 34 | + "MD": b"S2-8127-260101-V7.00 \x00\r", |
| 35 | + "F": b"#115.0 105 48.00 60.0\r", |
| 36 | + "RB": b"\x01" + rb + bytes([sum(rb) & 255]) + b"\r", |
| 37 | + } |
| 38 | + # Deliberately malformed wire evidence must survive export too. Do not |
| 39 | + # require successful semantic decoding just to retain a diagnostic sample. |
| 40 | + if outcome == "invalid": |
| 41 | + responses["RB"] = responses["RB"][:-2] + b"\xff\r" |
| 42 | + |
| 43 | + class Transport: |
| 44 | + connected = True |
| 45 | + capturing = False |
| 46 | + capture_commands = [] |
| 47 | + |
| 48 | + async def async_send_payload(self, payload, *, route, request_timeout=None): |
| 49 | + assert type(route) is EybondLinkRoute |
| 50 | + assert (route.devcode, route.collector_addr) == (767, 255) |
| 51 | + command = payload[:-2].decode("ascii") |
| 52 | + assert command in READ_COMMANDS and payload[-2:] == b"\x01\r" |
| 53 | + if self.capturing: |
| 54 | + self.capture_commands.append(command) |
| 55 | + if outcome == "offline" or (outcome == "partial" and command == "F"): |
| 56 | + raise ConnectionError("synthetic_connection_lost") |
| 57 | + if outcome == "partial" and command == "RB": |
| 58 | + raise TimeoutError("synthetic_read_timeout") |
| 59 | + return bytes(responses[command]) |
| 60 | + |
| 61 | + driver, transport = EybondShortAsciiDriver(), Transport() |
| 62 | + inverter = await driver.async_probe(transport, ProbeTarget(767, 255, 1)) |
| 63 | + assert inverter is not None |
| 64 | + |
| 65 | + async def refresh(self, *, poll_interval=None): |
| 66 | + return RuntimeSnapshot( |
| 67 | + connected=outcome != "offline", |
| 68 | + collector=CollectorInfo( |
| 69 | + remote_ip=SYNTHETIC_COLLECTOR_IP, collector_pn=SYNTHETIC_COLLECTOR_PN, |
| 70 | + ), |
| 71 | + inverter=inverter, |
| 72 | + values=parse_q1(responses["Q1"]) if outcome != "offline" else {}, |
| 73 | + ) |
| 74 | + |
| 75 | + async def capture(self): |
| 76 | + transport.capturing = True |
| 77 | + try: |
| 78 | + return await driver.async_capture_support_evidence(transport, inverter) |
| 79 | + finally: |
| 80 | + transport.capturing = False |
| 81 | + |
| 82 | + monkeypatch.setattr(FakeRuntimeManager, "async_refresh", refresh) |
| 83 | + monkeypatch.setattr(FakeRuntimeManager, "async_capture_support_evidence", capture) |
| 84 | + entry = MockConfigEntry( |
| 85 | + domain=DOMAIN, version=3, title="Short-ASCII support test", |
| 86 | + unique_id=f"collector:{SYNTHETIC_COLLECTOR_PN}", |
| 87 | + data={ |
| 88 | + "connection_type": "eybond", "connection_mode": "known_ip", |
| 89 | + "server_ip": SYNTHETIC_SERVER_IP, "collector_ip": SYNTHETIC_COLLECTOR_IP, |
| 90 | + "collector_pn": SYNTHETIC_COLLECTOR_PN, |
| 91 | + "collector_operation_mode": "home_assistant_only", |
| 92 | + "tcp_port": 8899, "udp_port": 58899, "driver_hint": "auto", |
| 93 | + "control_mode": "read_only", "connection_strategy": "callback_on_demand", |
| 94 | + "endpoint_control_policy": "external", "proxy_enabled": False, |
| 95 | + "detected_driver": driver.key, "detected_model": inverter.model_name, |
| 96 | + "detected_serial": "", "detection_confidence": "high", |
| 97 | + }, |
| 98 | + options={"poll_interval": 30, "poll_mode": "auto"}, |
| 99 | + ) |
| 100 | + entry.add_to_hass(hass) |
| 101 | + assert await hass.config_entries.async_setup(entry.entry_id) |
| 102 | + await hass.async_block_till_done() |
| 103 | + await entry.runtime_data.async_refresh() |
| 104 | + await hass.async_block_till_done() |
| 105 | + coordinator = entry.runtime_data |
| 106 | + path = Path(await coordinator.async_export_support_package()) |
| 107 | + assert path.is_relative_to(Path(hass.config.config_dir) / "eybond_local/support_packages") |
| 108 | + assert not (Path(hass.config.config_dir) / "www/eybond_local/support_packages").exists() |
| 109 | + url = coordinator.data.values["support_package_download_url"] |
| 110 | + assert "authSig=" in url |
| 111 | + # A frontend link opens without an Authorization header; its signed URL |
| 112 | + # must work on its own, while the same unsigned private route must not. |
| 113 | + client = await hass_client_no_auth() |
| 114 | + denied = await client.get(url.split("?", 1)[0]) |
| 115 | + assert denied.status == 401 |
| 116 | + response = await client.get(url) |
| 117 | + assert response.status == 200 |
| 118 | + downloaded = await response.read() |
| 119 | + assert downloaded == await hass.async_add_executor_job(path.read_bytes) |
| 120 | + |
| 121 | + with zipfile.ZipFile(BytesIO(downloaded)) as archive: |
| 122 | + assert archive.testzip() is None |
| 123 | + manifest = json.loads(archive.read("manifest.json")) |
| 124 | + bundle = json.loads(archive.read("support_bundle.json")) |
| 125 | + raw = json.loads(archive.read("raw_capture.json")) |
| 126 | + assert manifest["sharing_guidance"]["recommended_artifact"] == path.name |
| 127 | + assert SYNTHETIC_COLLECTOR_PN not in archive.read("support_bundle.json").decode() |
| 128 | + assert bundle["runtime"]["connected"] is (outcome != "offline") |
| 129 | + assert raw["capture_kind"] == "short_ascii_read_only" |
| 130 | + assert transport.capture_commands == list(READ_COMMANDS) |
| 131 | + failed = set(READ_COMMANDS) if outcome == "offline" else {"F", "RB"} if outcome == "partial" else set() |
| 132 | + assert set(raw["failures"]) == failed |
| 133 | + assert set(raw["responses_hex"]) == set(READ_COMMANDS) - failed |
| 134 | + for command, payload in raw["responses_hex"].items(): |
| 135 | + assert bytes.fromhex(payload) == responses[command], command |
| 136 | + if outcome == "partial": |
| 137 | + assert raw["failures"] == {"F": "ConnectionError", "RB": "TimeoutError"} |
| 138 | + if outcome == "ok": |
| 139 | + assert parse_rb(bytes.fromhex(raw["responses_hex"]["RB"]))["battery_soc"] == 80 |
| 140 | + assert await hass.config_entries.async_unload(entry.entry_id) |
| 141 | + await hass.async_block_till_done() |
0 commit comments