ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[pre-commit.ci] auto fixes from pre-commit.com hooks

Browse source 
for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot] 2026-05-13 14:20:58 +00:00
commit a03a119469
9 changed files with 73 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

37
studio/backend/core/inference/tools.py
View file

@ -1386,10 +1386,23 @@ def _check_signal_escape_patterns(code: str):
# egress (``from requests import get; get("http://...")``).
_NETWORK_FROM_NAMES = frozenset(
{
"get", "post", "put", "delete", "patch", "head", "request",
"Session", "urlopen", "urlretrieve", "create_connection",
"getaddrinfo", "HTTPConnection", "HTTPSConnection",
"Client", "AsyncClient", "ClientSession",
"get",
"post",
"put",
"delete",
"patch",
"head",
"request",
"Session",
"urlopen",
"urlretrieve",
"create_connection",
"getaddrinfo",
"HTTPConnection",
"HTTPSConnection",
"Client",
"AsyncClient",
"ClientSession",
}
)
# Session/Client method names that perform HTTP egress. A literal
@ -1463,7 +1476,9 @@ def _check_signal_escape_patterns(code: str):
parts: list[str] = []
ok = True
for piece in value.values:
if isinstance(piece, ast.Constant) and isinstance(piece.value, str):
if isinstance(piece, ast.Constant) and isinstance(
piece.value, str
):
parts.append(piece.value)
elif (
isinstance(piece, ast.FormattedValue)
@ -1509,7 +1524,11 @@ def _check_signal_escape_patterns(code: str):
head = cur.id
# If head is a session-bound variable, synthesise a
# prefixable FQ (e.g. ``"requests.Session.get"``).
if head in self._session_vars and parts and parts[-1] in _SESSION_METHOD_NAMES:
if (
head in self._session_vars
and parts
and parts[-1] in _SESSION_METHOD_NAMES
):
return f"{self._session_vars[head]}.{parts[-1]}"
# Map the head through module-alias table.
resolved_head = self._module_aliases.get(head, head)
@ -1628,11 +1647,7 @@ def _check_signal_escape_patterns(code: str):
break
# Special case: requests.request(method, url, ...) -- the
# URL is the second positional arg, not the first.
if (
host_arg is None
and fq.endswith(".request")
and len(node.args) >= 2
):
if host_arg is None and fq.endswith(".request") and len(node.args) >= 2:
_, host_arg = self._resolve_url_arg(node.args[1])
if host_arg is None and node.args:
# The argument was opaque (variable, computed, etc.).

4
studio/backend/models/inference.py
View file

@ -451,7 +451,9 @@ class ChatMessage(BaseModel):
# See ``_pair_orphan_tool_ids`` in routes/inference.py.
import secrets as _secrets
self.tool_call_id = f"{TOOL_CALL_ID_SYNTH_PREFIX}{_secrets.token_hex(8)}"
self.tool_call_id = (
f"{TOOL_CALL_ID_SYNTH_PREFIX}{_secrets.token_hex(8)}"
)
if not self.content:
raise ValueError('role="tool" messages require non-empty "content".')
elif self.role == "assistant":

14
studio/backend/tests/test_health_unauth_contract.py
View file

@ -96,7 +96,7 @@ class TestUnauthHealth:
def test_invalid_bearer_drops_back_to_unauth(self, fastapi_client):
client, _ = fastapi_client
body = client.get(
"/api/health", headers={"Authorization": "Bearer not-real"}
"/api/health", headers = {"Authorization": "Bearer not-real"}
).json()
leaked = GATED_KEYS & set(body)
assert not leaked, f"invalid-bearer health leaked {sorted(leaked)}"
@ -112,7 +112,7 @@ class TestUnauthHealth:
"""
client, _ = fastapi_client
body = client.get(
"/api/health", headers={"Authorization": "Bearer x.y.z"}
"/api/health", headers = {"Authorization": "Bearer x.y.z"}
).json()
# Verify gated fields are NOT leaked when the token cannot be
# decoded (which would have been the case for the original bug).
@ -149,9 +149,9 @@ class TestAuthedHealth:
seed()
else:
pytest.skip("storage has no ensure_default_admin entrypoint")
assert storage.get_user_and_secret(storage.DEFAULT_ADMIN_USERNAME) is not None, (
"could not seed default admin"
)
assert (
storage.get_user_and_secret(storage.DEFAULT_ADMIN_USERNAME) is not None
), "could not seed default admin"
# Clear the must_change_password flag so /api/health's
# get_current_subject dependency accepts the token. Fresh installs
# block diagnostic access until the first-boot password change,
@ -165,9 +165,9 @@ class TestAuthedHealth:
conn.commit()
finally:
conn.close()
token = create_access_token(subject=storage.DEFAULT_ADMIN_USERNAME)
token = create_access_token(subject = storage.DEFAULT_ADMIN_USERNAME)
body = client.get(
"/api/health", headers={"Authorization": f"Bearer {token}"}
"/api/health", headers = {"Authorization": f"Bearer {token}"}
).json()
# Diagnostic keys are present.
for k in ("version", "device_type"):

26
studio/backend/tests/test_kill_process_tree_platform.py
View file

@ -41,17 +41,17 @@ def _spawn_sleep(seconds: int = 60):
if Path(sleep_bin).exists():
return subprocess.Popen(
[sleep_bin, str(seconds)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
stdout = subprocess.DEVNULL,
stderr = subprocess.DEVNULL,
start_new_session = True,
)
# Fallback: minimal Python sleeper. Adds ~25-40 MB per test which
# is fine for single-test runs but is the reason we prefer
# /bin/sleep when available (the suite spawns one per test).
return subprocess.Popen(
[sys.executable, "-c", "import time; time.sleep(%d)" % seconds],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdout = subprocess.DEVNULL,
stderr = subprocess.DEVNULL,
)
@ -73,7 +73,7 @@ class TestUnixPath:
@pytest.mark.skipif(
not (hasattr(os, "getpgid") and hasattr(os, "killpg")),
reason="No process-group APIs on this platform",
reason = "No process-group APIs on this platform",
)
def test_kill_terminates_subprocess(self, short_proc):
assert short_proc.poll() is None
@ -86,7 +86,7 @@ class TestUnixPath:
@pytest.mark.skipif(
not (hasattr(os, "getpgid") and hasattr(os, "killpg")),
reason="No process-group APIs on this platform",
reason = "No process-group APIs on this platform",
)
def test_no_raise_on_already_exited(self):
# poll() already returned non-None: helper must early-return.
@ -109,16 +109,16 @@ class TestWindowsFallback:
# Strip the Unix-only attributes so the helper takes the
# Windows branch.
if hasattr(os, "getpgid"):
monkeypatch.delattr(os, "getpgid", raising=False)
monkeypatch.delattr(os, "getpgid", raising = False)
if hasattr(os, "killpg"):
monkeypatch.delattr(os, "killpg", raising=False)
monkeypatch.delattr(os, "killpg", raising = False)
# Also lie about sys.platform so the taskkill fallback runs.
monkeypatch.setattr(tools_mod.sys, "platform", "win32")
# taskkill won't exist on Linux; capture its absence as a no-op
# via subprocess.run rather than failing the test.
import subprocess as _sp
with mock.patch.object(_sp, "run", return_value=None):
with mock.patch.object(_sp, "run", return_value = None):
tools_mod._kill_process_tree(short_proc)
deadline = time.time() + 5.0
while short_proc.poll() is None and time.time() < deadline:
@ -135,9 +135,9 @@ class TestWindowsFallback:
``hasattr(os, ...)`` first, so this test pins that contract.
"""
if hasattr(os, "getpgid"):
monkeypatch.delattr(os, "getpgid", raising=False)
monkeypatch.delattr(os, "getpgid", raising = False)
if hasattr(os, "killpg"):
monkeypatch.delattr(os, "killpg", raising=False)
monkeypatch.delattr(os, "killpg", raising = False)
monkeypatch.setattr(tools_mod.sys, "platform", "win32")
class FakeProc:
@ -155,6 +155,6 @@ class TestWindowsFallback:
fp = FakeProc()
import subprocess as _sp
with mock.patch.object(_sp, "run", return_value=None):
with mock.patch.object(_sp, "run", return_value = None):
tools_mod._kill_process_tree(fp) # must not raise
assert fp.killed

10
studio/backend/tests/test_middleware.py
View file

@ -264,9 +264,7 @@ class TestHealthAuthGate:
missing = self.UNAUTH_KEYS - set(body)
assert not missing, f"unauth /api/health missing {sorted(missing)}"
for forbidden in self.GATED_KEYS:
assert forbidden not in body, (
f"unauth /api/health leaked {forbidden!r}"
)
assert forbidden not in body, f"unauth /api/health leaked {forbidden!r}"
def test_invalid_bearer_returns_launcher_payload(self, health_app):
# Regression: calling the async dep without await made any
@ -283,9 +281,9 @@ class TestHealthAuthGate:
missing = self.UNAUTH_KEYS - set(body)
assert not missing, f"invalid-bearer /api/health missing {sorted(missing)}"
for forbidden in self.GATED_KEYS:
assert forbidden not in body, (
f"invalid-bearer /api/health leaked {forbidden!r}"
)
assert (
forbidden not in body
), f"invalid-bearer /api/health leaked {forbidden!r}"
def test_valid_bearer_returns_full_payload(self, health_app):
from auth import storage

17
studio/backend/tests/test_refresh_token_consume.py
View file

@ -45,14 +45,14 @@ def isolated_storage(tmp_path, monkeypatch):
yield storage
def _make_token(storage, *, is_desktop=False):
def _make_token(storage, *, is_desktop = False):
"""Insert a fresh, far-future refresh token and return its raw form."""
import secrets as _secrets
from datetime import datetime, timedelta, timezone
raw = _secrets.token_urlsafe(32)
expires = (datetime.now(timezone.utc) + timedelta(days=14)).isoformat()
storage.save_refresh_token(raw, "unsloth", expires, is_desktop=is_desktop)
expires = (datetime.now(timezone.utc) + timedelta(days = 14)).isoformat()
storage.save_refresh_token(raw, "unsloth", expires, is_desktop = is_desktop)
return raw
@ -77,7 +77,7 @@ class TestSingleUseRotation:
def test_desktop_flag_round_trips(self, isolated_storage):
storage = isolated_storage
token = _make_token(storage, is_desktop=True)
token = _make_token(storage, is_desktop = True)
assert storage.consume_refresh_token(token) == ("unsloth", True)
@ -117,14 +117,14 @@ class TestReturningFallback:
r = storage.consume_refresh_token(token)
(winners if r else losers).append(r)
threads = [threading.Thread(target=attempt) for _ in range(8)]
threads = [threading.Thread(target = attempt) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(winners) == 1, (
f"expected exactly 1 winner; winners={winners} losers={losers}"
)
assert (
len(winners) == 1
), f"expected exactly 1 winner; winners={winners} losers={losers}"
assert winners[0] == ("unsloth", False)
assert all(r is None for r in losers)
@ -136,4 +136,3 @@ class TestReturningSupportedProbe:
storage._RETURNING_SUPPORTED = None
result = storage._supports_returning()
assert isinstance(result, bool)

17
studio/backend/tests/test_sandbox_tools.py
View file

@ -124,22 +124,20 @@ class TestUntrustedHostBlock:
# URL is staged into a local. Before this hardening pass the
# variable-URL path was treated as opaque and slipped through.
_blocked(
'import requests\n'
'url = "https://example.com/"\n'
'requests.get(url)',
"import requests\n" 'url = "https://example.com/"\n' "requests.get(url)",
expect_phrase = "Blocked: host not in sandbox allowlist",
)
_blocked(
'import requests\n'
"import requests\n"
'url = "http://169.254.169.254/latest/meta-data/"\n'
'requests.get(url)',
"requests.get(url)",
expect_phrase = "Blocked: cloud-metadata host",
)
def test_constant_fstring_url_resolved(self):
# f-string URLs that fold to a constant should still be checked.
_blocked(
'import requests\n'
"import requests\n"
'host = "169.254.169.254"\n'
'requests.get(f"http://{host}/latest/")',
expect_phrase = "Blocked: cloud-metadata host",
@ -151,8 +149,7 @@ class TestUntrustedHostBlock:
# bash blocklist + cloud-metadata IP block at OS layer cover the
# rest, but the AST can no longer say "looks fine to me".
_blocked(
'import os, requests\n'
'requests.get(os.environ["WEBHOOK"])',
"import os, requests\n" 'requests.get(os.environ["WEBHOOK"])',
expect_phrase = "network call target is computed at runtime",
)
@ -294,13 +291,13 @@ class TestImportAliasResolution:
def test_from_import_trusted_passes(self):
_ok(
'from urllib.request import urlopen\n'
"from urllib.request import urlopen\n"
'urlopen("https://en.wikipedia.org/wiki/Foo")'
)
def test_nested_module_alias_blocked(self):
_blocked(
'import urllib.request as ur\n'
"import urllib.request as ur\n"
'ur.urlopen("http://169.254.169.254/latest/")',
expect_phrase = "Blocked: cloud-metadata host",
)

4
studio/backend/tests/test_tool_id_pairing.py
View file

@ -128,12 +128,12 @@ class TestValidatorSynth:
def test_validator_emits_synth_prefix(self):
from models.inference import ChatMessage
m = ChatMessage(role="tool", content="result")
m = ChatMessage(role = "tool", content = "result")
assert m.tool_call_id is not None
assert m.tool_call_id.startswith(TOOL_CALL_ID_SYNTH_PREFIX)
def test_validator_keeps_explicit_id(self):
from models.inference import ChatMessage
m = ChatMessage(role="tool", tool_call_id="call_real_xyz", content="ok")
m = ChatMessage(role = "tool", tool_call_id = "call_real_xyz", content = "ok")
assert m.tool_call_id == "call_real_xyz"

4
tests/studio/playwright_extra_ui.py
View file

@ -539,7 +539,9 @@ with sync_playwright() as p:
timeout = 30_000,
)
except PlaywrightTimeoutError:
info("/studio hydration didn't complete in 30s; continuing with whatever rendered")
info(
"/studio hydration didn't complete in 30s; continuing with whatever rendered"
)
page.wait_for_timeout(500)
shoot("08-studio")
if chat_only: