fix(gateway): wrap cron helpers with staticmethod to prevent self-binding

Plain functions imported as class attributes in APIServerAdapter get
auto-bound as methods via Python's descriptor protocol.  Every
self._cron_*() call injected self as the first positional argument,
causing TypeError on all 8 cron API endpoints at runtime.

Wrap each import with staticmethod() so self._cron_*() calls dispatch
correctly without modifying any call sites.

Co-authored-by: teknium <teknium@nousresearch.com>
This commit is contained in:
MichaelWDanko
2026-04-05 12:28:09 -07:00
committed by Teknium
parent cc2b56b26a
commit c6793d6fc3
2 changed files with 78 additions and 0 deletions

View File

@@ -974,6 +974,18 @@ class APIServerAdapter(BasePlatformAdapter):
resume_job as _cron_resume,
trigger_job as _cron_trigger,
)
# Wrap as staticmethod to prevent descriptor binding — these are plain
# module functions, not instance methods. Without this, self._cron_*()
# injects ``self`` as the first positional argument and every call
# raises TypeError.
_cron_list = staticmethod(_cron_list)
_cron_get = staticmethod(_cron_get)
_cron_create = staticmethod(_cron_create)
_cron_update = staticmethod(_cron_update)
_cron_remove = staticmethod(_cron_remove)
_cron_pause = staticmethod(_cron_pause)
_cron_resume = staticmethod(_cron_resume)
_cron_trigger = staticmethod(_cron_trigger)
_CRON_AVAILABLE = True
except ImportError:
pass

View File

@@ -540,6 +540,72 @@ class TestCronUnavailable:
data = await resp.json()
assert "not available" in data["error"].lower()
@pytest.mark.asyncio
async def test_pause_handler_no_self_binding(self, adapter):
"""Pause must not inject ``self`` into the cron helper call."""
app = _create_app(adapter)
captured = {}
def _plain_pause(job_id):
captured["job_id"] = job_id
return SAMPLE_JOB
async with TestClient(TestServer(app)) as cli:
with patch.object(APIServerAdapter, "_CRON_AVAILABLE", True), patch.object(
APIServerAdapter, "_cron_pause", staticmethod(_plain_pause)
):
resp = await cli.post(f"/api/jobs/{VALID_JOB_ID}/pause")
assert resp.status == 200
data = await resp.json()
assert data["job"] == SAMPLE_JOB
assert captured["job_id"] == VALID_JOB_ID
@pytest.mark.asyncio
async def test_list_handler_no_self_binding(self, adapter):
"""List must preserve keyword arguments without injecting ``self``."""
app = _create_app(adapter)
captured = {}
def _plain_list(include_disabled=False):
captured["include_disabled"] = include_disabled
return [SAMPLE_JOB]
async with TestClient(TestServer(app)) as cli:
with patch.object(APIServerAdapter, "_CRON_AVAILABLE", True), patch.object(
APIServerAdapter, "_cron_list", staticmethod(_plain_list)
):
resp = await cli.get("/api/jobs?include_disabled=true")
assert resp.status == 200
data = await resp.json()
assert data["jobs"] == [SAMPLE_JOB]
assert captured["include_disabled"] is True
@pytest.mark.asyncio
async def test_update_handler_no_self_binding(self, adapter):
"""Update must pass positional arguments correctly without ``self``."""
app = _create_app(adapter)
captured = {}
updated_job = {**SAMPLE_JOB, "name": "updated-name"}
def _plain_update(job_id, updates):
captured["job_id"] = job_id
captured["updates"] = updates
return updated_job
async with TestClient(TestServer(app)) as cli:
with patch.object(APIServerAdapter, "_CRON_AVAILABLE", True), patch.object(
APIServerAdapter, "_cron_update", staticmethod(_plain_update)
):
resp = await cli.patch(
f"/api/jobs/{VALID_JOB_ID}",
json={"name": "updated-name"},
)
assert resp.status == 200
data = await resp.json()
assert data["job"] == updated_job
assert captured["job_id"] == VALID_JOB_ID
assert captured["updates"] == {"name": "updated-name"}
@pytest.mark.asyncio
async def test_cron_unavailable_create(self, adapter):
"""POST /api/jobs returns 501 when _CRON_AVAILABLE is False."""