Shaun ZhouandClaude Opus 5
9ead8c0a6e
服务器上的 Label Studio 关闭了旧式 token（jwt_auth_jwtsettings.legacy_api_tokens_enabled=0）， 用 authtoken_token 里的 key 直接请求返回 401。个人访问令牌是 JWT refresh token， 要先 POST /api/token/refresh/ 换成 access token 再以 Bearer 发送。 脚本按令牌形态自动区分：三段点分的按 JWT 换取，否则沿用旧式 Token。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""批量给 Label Studio 推预测：URL 还原与分页拉取。"""
|
|
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("requests")
|
|
|
|
sys.path.insert(0, "scripts")
|
|
import push_predictions as pp # noqa: E402
|
|
|
|
|
|
def test_local_path_from_local_files_url(tmp_path):
|
|
url = "/data/local-files/?d=boxes_shap_label_studio/images/blackdot_0489.png"
|
|
assert pp.local_path(url, tmp_path) == (
|
|
tmp_path / "boxes_shap_label_studio/images/blackdot_0489.png"
|
|
)
|
|
|
|
|
|
def test_local_path_unquotes_escaped_names(tmp_path):
|
|
"""文件名里有中文或空格时 URL 是转义过的，得还原回来。"""
|
|
url = "/data/local-files/?d=proj/images/%E5%8F%8D%E9%9D%A2%20a.png"
|
|
assert pp.local_path(url, tmp_path) == tmp_path / "proj/images/反面 a.png"
|
|
|
|
|
|
def test_local_path_rejects_non_local_urls(tmp_path):
|
|
with pytest.raises(ValueError, match="本地存储"):
|
|
pp.local_path("https://example.com/a.png", tmp_path)
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload, status=200):
|
|
self._payload = payload
|
|
self.status_code = status
|
|
|
|
@property
|
|
def text(self):
|
|
return str(self._payload)
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise AssertionError(f"HTTP {self.status_code}")
|
|
|
|
|
|
class _FakeSession:
|
|
"""按页返回任务，记录请求参数。"""
|
|
|
|
def __init__(self, pages):
|
|
self.pages = pages
|
|
self.headers = {}
|
|
self.calls = []
|
|
|
|
def get(self, url, params=None, timeout=None):
|
|
self.calls.append(params)
|
|
page = params["page"]
|
|
return _FakeResponse(self.pages[page - 1] if page <= len(self.pages) else {"tasks": []})
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
monkeypatch.setattr(pp, "LabelStudio", pp.LabelStudio)
|
|
instance = pp.LabelStudio.__new__(pp.LabelStudio)
|
|
instance.url = "http://ls"
|
|
return instance
|
|
|
|
|
|
def test_tasks_pagination_stops_at_total(client):
|
|
"""分页要拉完再停，不能只拿第一页。"""
|
|
client.session = _FakeSession([
|
|
{"tasks": [{"id": 1}, {"id": 2}], "total": 3},
|
|
{"tasks": [{"id": 3}], "total": 3},
|
|
])
|
|
assert [t["id"] for t in client.tasks(39, page_size=2)] == [1, 2, 3]
|
|
|
|
|
|
def test_tasks_handles_drf_style_pagination(client):
|
|
"""有的版本返回 results/count 而不是 tasks/total。"""
|
|
client.session = _FakeSession([
|
|
{"results": [{"id": 1}], "count": 2},
|
|
{"results": [{"id": 2}], "count": 2},
|
|
])
|
|
assert [t["id"] for t in client.tasks(39, page_size=1)] == [1, 2]
|
|
|
|
|
|
def test_tasks_stops_on_empty_page(client):
|
|
"""没有 total 字段时，靠空页收尾，不能无限翻。"""
|
|
client.session = _FakeSession([{"tasks": [{"id": 1}]}, {"tasks": []}])
|
|
assert [t["id"] for t in client.tasks(39)] == [1]
|
|
assert len(client.session.calls) == 2
|
|
|
|
|
|
# ---------------- 两种令牌 ----------------
|
|
|
|
|
|
class _TokenSession:
|
|
def __init__(self, access="ACCESS", status=200):
|
|
self.headers = {}
|
|
self.access, self.status, self.posted = access, status, []
|
|
|
|
def post(self, url, json=None, timeout=None):
|
|
self.posted.append((url, json))
|
|
return _FakeResponse({"access": self.access}, self.status)
|
|
|
|
|
|
def test_legacy_token_is_sent_as_token(monkeypatch):
|
|
client = pp.LabelStudio.__new__(pp.LabelStudio)
|
|
client.url = "http://ls"
|
|
client.session = _TokenSession()
|
|
assert client._authorization("c657261234567890") == "Token c657261234567890"
|
|
assert client.session.posted == [] # 旧式令牌不需要换取
|
|
|
|
|
|
def test_jwt_personal_token_is_exchanged_for_a_bearer(monkeypatch):
|
|
"""新版默认禁用旧式 token，个人访问令牌要先换成 access token。"""
|
|
client = pp.LabelStudio.__new__(pp.LabelStudio)
|
|
client.url = "http://ls"
|
|
client.session = _TokenSession(access="AAA")
|
|
assert client._authorization("aaa.bbb.ccc") == "Bearer AAA"
|
|
assert client.session.posted[0][0] == "http://ls/api/token/refresh/"
|
|
assert client.session.posted[0][1] == {"refresh": "aaa.bbb.ccc"}
|
|
|
|
|
|
def test_failed_exchange_reports_clearly():
|
|
client = pp.LabelStudio.__new__(pp.LabelStudio)
|
|
client.url = "http://ls"
|
|
client.session = _TokenSession(status=401)
|
|
with pytest.raises(RuntimeError, match="access token 失败"):
|
|
client._authorization("aaa.bbb.ccc")
|