style: repo-wide lint pass — make the lint gate green again

Push builds have been RED on the lint step since ~build 209 from accumulated
formatting drift. This is the mechanical cleanup: ruff format + ruff --fix
(UP038 isinstance unions, SIM105 contextlib.suppress, UP031 f-strings, SIM115
tempfile context manager), shfmt -i 2 -ci, nixpkgs-fmt/statix/deadnix (merged
attrsets, dropped unused lib args), yamllint, and shell quoting fixes in
tests/lasuite-docs/setup_custom_tests.sh. No behaviour changes intended;
lint: PASS, unit tests: 138 passed.
This commit is contained in:
2026-06-09 21:56:15 +00:00
parent e76d4005ab
commit 9a7772563a
115 changed files with 952 additions and 660 deletions

View File

@ -12,6 +12,7 @@ cc-ci host (`mode: host`); tests run on-host via cc-ci-run, so they connect to 1
from __future__ import annotations
import contextlib
import socket
import ssl
import struct
@ -29,8 +30,14 @@ MSG_USERSTATE = 9
MSG_SERVERCONFIG = 24
REJECT_TYPES = {
0: "None", 1: "WrongVersion", 2: "InvalidUsername", 3: "WrongUserPW",
4: "WrongServerPW", 5: "UsernameInUse", 6: "ServerFull", 7: "NoCertificate",
0: "None",
1: "WrongVersion",
2: "InvalidUsername",
3: "WrongUserPW",
4: "WrongServerPW",
5: "UsernameInUse",
6: "ServerFull",
7: "NoCertificate",
8: "AuthenticatorFail",
}
@ -81,7 +88,7 @@ def _dec_fields(data: bytes) -> dict:
off += 8
elif wire == 2:
length, off = _dec_varint(data, off)
raw = data[off:off + length]
raw = data[off : off + length]
off += length
try:
value = raw.decode("utf-8")
@ -120,9 +127,11 @@ def _recv(sock, timeout: float) -> tuple[int, bytes]:
def _build_version() -> bytes:
v = (1 << 16) | (5 << 8) | 0 # pretend client 1.5.0
return (_enc_field_varint(1, v)
+ _enc_field_string(2, "cc-ci mumble probe 1.0")
+ _enc_field_string(3, "Linux"))
return (
_enc_field_varint(1, v)
+ _enc_field_string(2, "cc-ci mumble probe 1.0")
+ _enc_field_string(3, "Linux")
)
def _build_authenticate(username: str, password: str = "") -> bytes:
@ -133,18 +142,29 @@ def _build_authenticate(username: str, password: str = "") -> bytes:
return payload
def handshake(host: str = "127.0.0.1", port: int = PORT, username: str = "cc-ci-probe",
password: str = "", timeout: float = 20.0) -> dict:
def handshake(
host: str = "127.0.0.1",
port: int = PORT,
username: str = "cc-ci-probe",
password: str = "",
timeout: float = 20.0,
) -> dict:
"""Full Mumble control-channel handshake. Returns a result dict:
tls_connect (bool), server_version (dict|None), auth_accepted (bool), channels (list[str]),
users (list[str]), server_sync (bool), welcome_text (str|None), server_config (dict),
error (str|None).
tls_connect (bool), server_version (dict|None), auth_accepted (bool), channels (list[str]),
users (list[str]), server_sync (bool), welcome_text (str|None), server_config (dict),
error (str|None).
"""
result = {
"tls_connect": False, "server_version": None, "auth_accepted": False,
"channels": [], "users": [], "server_sync": False, "welcome_text": None,
"server_config": {}, "error": None,
"tls_connect": False,
"server_version": None,
"auth_accepted": False,
"channels": [],
"users": [],
"server_sync": False,
"welcome_text": None,
"server_config": {},
"error": None,
}
raw = tls = None
try:
@ -181,19 +201,21 @@ def handshake(host: str = "127.0.0.1", port: int = PORT, username: str = "cc-ci-
break
try:
msg_type, payload = _recv(tls, timeout=remaining)
except (socket.timeout, ConnectionError):
except (TimeoutError, ConnectionError):
break
if msg_type == MSG_VERSION:
f = _dec_fields(payload)
v1 = f.get(1, 0)
result["server_version"] = {
"string": f"{(v1 >> 16) & 0xFF}.{(v1 >> 8) & 0xFF}.{v1 & 0xFF}",
"release": f.get(2, ""), "os": f.get(3, ""),
"release": f.get(2, ""),
"os": f.get(3, ""),
}
elif msg_type == MSG_REJECT:
f = _dec_fields(payload)
result["error"] = (f"Rejected: {REJECT_TYPES.get(f.get(1, 0), 'Unknown')} "
f"{f.get(2, '')}")
result["error"] = (
f"Rejected: {REJECT_TYPES.get(f.get(1, 0), 'Unknown')} " f"{f.get(2, '')}"
)
return result
elif msg_type == MSG_CHANNELSTATE:
f = _dec_fields(payload)
@ -209,9 +231,12 @@ def handshake(host: str = "127.0.0.1", port: int = PORT, username: str = "cc-ci-
# ServerConfig fields: 1 max_bandwidth, 2 welcome_text, 3 allow_html,
# 4 message_length, 5 image_message_length, 6 max_users
result["server_config"] = {
"max_bandwidth": f.get(1), "welcome_text": f.get(2),
"allow_html": f.get(3), "message_length": f.get(4),
"image_message_length": f.get(5), "max_users": f.get(6),
"max_bandwidth": f.get(1),
"welcome_text": f.get(2),
"allow_html": f.get(3),
"message_length": f.get(4),
"image_message_length": f.get(5),
"max_users": f.get(6),
}
elif msg_type == MSG_SERVERSYNC:
f = _dec_fields(payload)
@ -230,10 +255,8 @@ def handshake(host: str = "127.0.0.1", port: int = PORT, username: str = "cc-ci-
result["error"] = f"{type(e).__name__}: {e}"
finally:
if tls is not None:
try:
with contextlib.suppress(OSError):
tls.shutdown(socket.SHUT_RDWR)
except OSError:
pass
tls.close()
elif raw is not None:
raw.close()

View File

@ -25,7 +25,7 @@ def test_handshake_completes_with_channel_presence(live_app):
assert r["server_version"] is not None, "server did not send a Version message"
assert r["auth_accepted"], f"authentication not accepted — {r.get('error')}"
# Channel presence: the server must expose at least the root channel (beyond a bare TCP open).
assert len(r["channels"]) >= 1, (
f"server reported no channels (expected >=1 root channel) — {r!r}"
)
assert (
len(r["channels"]) >= 1
), f"server reported no channels (expected >=1 root channel) — {r!r}"
assert r["server_sync"], f"ServerSync handshake did not complete — {r.get('error')}"

View File

@ -32,6 +32,7 @@ def test_configured_max_users_surfaces_in_serverconfig(live_app):
)
# allow_html defaults true in the recipe; assert it is present/boolean to prove the field set
# is the real ServerConfig (not an empty/garbled decode).
assert cfg.get("allow_html") in (0, 1), (
f"ServerConfig.allow_html unexpected: {cfg.get('allow_html')!r}"
)
assert cfg.get("allow_html") in (
0,
1,
), f"ServerConfig.allow_html unexpected: {cfg.get('allow_html')!r}"