test(e2e): the slot tests fill the free slots

Home Assistant holds a socket slot whenever it runs, so the tests that
needed every slot never ran on a device it uses. They now fill the slots
left and skip only when none is.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
cjkas 2026-10-27 22:24:48 +01:00
parent cfa298f053
commit c57b4bf432
2 changed files with 11 additions and 10 deletions

View file

@ -31,9 +31,9 @@ def watch_heap(host, token):
@pytest.fixture
def all_slots_free(host, token):
"""Skips unless every socket slot is free. The web UI holds one per open
tab, which would leave these tests a client short."""
def free_slots(host, token):
"""How many socket slots no other client holds. The web UI holds one per
open tab and Home Assistant one, so the tests fill what is left."""
clients = [ProtoClient(host, params={"apikey": token} if token else None) for _ in range(MAX_CLIENTS)]
try:
time.sleep(1)
@ -43,14 +43,15 @@ def all_slots_free(host, token):
c.close()
# The device frees a slot when it sees the disconnect.
time.sleep(0.5)
if taken:
pytest.skip("%u socket slot(s) in use, likely by the web UI: close its tabs" % taken)
if taken == MAX_CLIENTS:
pytest.skip("every socket slot is in use: close the web UI's tabs")
return MAX_CLIENTS - taken
# watch_heap comes before connect so pytest tears connect down first: the
# heap check after the test needs a free slot.
def test_every_client_reconnecting_at_once(watch_heap, all_slots_free, connect, schema):
clients = [connect() for _ in range(MAX_CLIENTS)]
def test_every_client_reconnecting_at_once(watch_heap, free_slots, connect, schema):
clients = [connect() for _ in range(free_slots)]
errors = []
def sync(c):
@ -64,8 +65,8 @@ def test_every_client_reconnecting_at_once(watch_heap, all_slots_free, connect,
assert not errors, errors
def test_one_client_too_many_is_closed(connect, schema, all_slots_free):
clients = [connect(schema=schema.hash) for _ in range(MAX_CLIENTS)]
def test_one_client_too_many_is_closed(connect, schema, free_slots):
clients = [connect(schema=schema.hash) for _ in range(free_slots)]
for c in clients:
run_sync(c)
extra = connect(schema=schema.hash)