ESPSomfy-RTS/test/e2e/conftest.py
cjkas 0b54264ee3 feat: command ops for socket protocol 2
shadeCommand, groupCommand, tiltCommand, repeatCommand, setPositions,
setSensor, setMyPosition, unpairShade, setRollingCode (types 16-24) are
read on async_tcp by ProtoOps and handed to loop() through the
LoopQueue, which replies once the request has run: ok, not found or
radio unavailable. A bad payload, a paused radio or a full queue is
answered straight away. sendRemoteCommand (25) touches nothing of the
shade model, so it goes to the radio from async_tcp like its route.

ShadeRequests::run() now says how a request went, not only whether
the radio took it. An op reads only the fields that were sent, so
setPositions can leave the tilt as it is. The shadeCommand event is
renamed shadeCommanded, since the op has the route's name.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-24 13:42:41 +02:00

97 lines
3.2 KiB
Python

"""Options and fixtures for the e2e suite. See README.md."""
import os
import sys
import time
import pytest
sys.path.insert(0, os.path.dirname(__file__))
from proto_client import ProtoClient, Schema, http_json, run_sync # noqa: E402
def pytest_addoption(parser):
parser.addoption("--device", required=True, help="the device's address, e.g. 192.168.1.120")
parser.addoption("--password", default=None, help="the device's password, when it is Secured with one")
parser.addoption("--username", default="", help="the username that goes with --password")
parser.addoption("--pin", default=None, help="the device's PIN, when it is Secured with one")
parser.addoption("--record", action="store_true", help="save what the device sends to test/fixtures/proto2")
parser.addoption("--radio", action="store_true", help="also run the tests that send commands on air")
parser.addoption("--shade", type=int, default=None, help="with --radio: the id of the shade the commands may move")
@pytest.fixture(scope="session")
def host(request):
return request.config.getoption("--device")
@pytest.fixture(scope="session")
def login_context(host):
return http_json(host, "/loginContext")
@pytest.fixture(scope="session")
def secured(login_context):
return login_context.get("type", 0) != 0
@pytest.fixture(scope="session")
def token(request, host, secured):
"""The token for this PC, or None when the device isn't Secured."""
if not secured:
return None
password = request.config.getoption("--password")
pin = request.config.getoption("--pin")
if password is None and pin is None:
pytest.skip("the device is Secured: pass --password or --pin")
body = {"username": request.config.getoption("--username"), "password": password or "", "pin": pin or ""}
return http_json(host, "/login", method="PUT", body=body)["apiKey"]
@pytest.fixture
def connect(host, token):
"""Opens protocol 2 clients and closes them after the test."""
clients = []
def open_client(**params):
if token and "apikey" not in params:
params["apikey"] = token
c = ProtoClient(host, params=params or None)
clients.append(c)
return c
yield open_client
for c in clients:
c.close()
# The device frees a slot when it sees the disconnect; the next test may
# need every slot.
time.sleep(0.5)
@pytest.fixture(scope="session")
def schema(host, token):
"""The device's schema, from one full sync."""
s = Schema()
c = ProtoClient(host, params={"apikey": token} if token else None)
try:
assert c.status == 101, "the device has no /ws: is it running protocol 2 firmware?"
run_sync(c, s)
finally:
c.close()
assert s.complete() and s.hash
return s
@pytest.fixture(scope="session")
def recording(request):
return request.config.getoption("--record")
@pytest.fixture(scope="session")
def radio_shade(request):
"""The shade --radio tests may move, or a skip."""
if not request.config.getoption("--radio"):
pytest.skip("sends on air: pass --radio --shade <id>")
shade = request.config.getoption("--shade")
if shade is None:
pytest.skip("--radio needs --shade <id>, a shade that may move")
return shade