mirror of
https://github.com/hansjone/oclaw.git
synced 2026-10-09 03:30:48 +08:00
oliver
420abac9f1
- Rename platform/ to svc/ to avoid shadowing stdlib platform. - Replace from oclaw.* with from svc/runtime/interfaces; update -m CLI paths. - tests/conftest: prepend repo root to sys.path (no parent-folder package name). - CI: paths and offline_eval script under repo root. - Ops scripts: PYTHONPATH must be repo root for python -m runtime.* (fixes gateway/WhatsApp sidecar startup). - Fix default oclaw.json path in tabular/file attachment limits; stabilize attachment test config. Co-authored-by: Cursor <cursoragent@cursor.com>
534 lines
22 KiB
Python
534 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import hashlib
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from interfaces.http.fastapi_app import create_app
|
|
from svc.persistence.sqlite_store import SqliteStore
|
|
|
|
|
|
class AdminChatStreamAsyncTaskTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|
self.db = Path(self._tmp.name) / "ops.sqlite"
|
|
self.oclaw_cfg = Path(self._tmp.name) / "oclaw.json"
|
|
self.oclaw_cfg.write_text(
|
|
json.dumps(
|
|
{
|
|
"plugins": {
|
|
"entries": {
|
|
"memory-wiki": {
|
|
"auto": {
|
|
"attachments": {
|
|
"tabular": {
|
|
"max_rows_read": 5000,
|
|
"max_columns": 200,
|
|
"max_cell_chars": 500,
|
|
"max_excel_sheets": 50,
|
|
"sql_timeout_ms": 8000,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
os.environ["OPS_ASSISTANT_DB_PATH"] = str(self.db)
|
|
os.environ["OPS_ASSISTANT_PASSWORD"] = "test-admin-pass"
|
|
os.environ["AIA_OCLAW_CONFIG_PATH"] = str(self.oclaw_cfg)
|
|
self.store = SqliteStore(str(self.db))
|
|
tenant = self.store.create_tenant("Team")
|
|
self.tenant_id = str(tenant["id"])
|
|
user = self.store.create_user_account(
|
|
tenant_id=self.tenant_id,
|
|
username="administrator",
|
|
display_name="Admin",
|
|
role="owner",
|
|
password_hash=hashlib.sha256("test-admin-pass".encode("utf-8")).hexdigest(),
|
|
is_active=True,
|
|
)
|
|
self.user_id = str(user["id"])
|
|
self.session_id = self.store.create_session_for_user(title="chat", tenant_id=self.tenant_id, user_id=self.user_id).id
|
|
self.client = TestClient(create_app())
|
|
self.client.post("/admin/api/auth/bootstrap", json={})
|
|
|
|
def tearDown(self) -> None:
|
|
os.environ.pop("AIA_OCLAW_CONFIG_PATH", None)
|
|
self._tmp.cleanup()
|
|
|
|
def _login(self) -> str:
|
|
resp = self.client.post(
|
|
"/admin/api/auth/login",
|
|
json={
|
|
"tenant_id": self.tenant_id,
|
|
"username": "administrator",
|
|
"password": "test-admin-pass",
|
|
"purpose": "chat",
|
|
},
|
|
)
|
|
body = resp.json()
|
|
self.assertTrue(body.get("ok"), body)
|
|
return str(body.get("token") or "")
|
|
|
|
def test_stream_done_contains_async_task_id(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "text/event-stream",
|
|
"content-type": "application/json",
|
|
}
|
|
with self.client.stream(
|
|
"POST",
|
|
f"/admin/api/chat/sessions/{self.session_id}/messages/stream",
|
|
headers=headers,
|
|
json={"text": "请总结并发送到项目群"},
|
|
) as resp:
|
|
self.assertEqual(resp.status_code, 200)
|
|
done = None
|
|
for line in resp.iter_lines():
|
|
if not line:
|
|
continue
|
|
s = line.decode("utf-8") if isinstance(line, bytes) else str(line)
|
|
if not s.startswith("data:"):
|
|
continue
|
|
ev = json.loads(s[5:].strip())
|
|
if ev.get("type") == "done":
|
|
done = ev
|
|
break
|
|
self.assertIsNotNone(done)
|
|
self.assertEqual(str(done.get("mode") or ""), "async_task")
|
|
self.assertTrue(str(done.get("task_id") or "").strip())
|
|
|
|
def test_non_stream_send_contains_async_task_id(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
resp = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/messages",
|
|
headers=headers,
|
|
json={"text": "请总结并发送到项目群"},
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertTrue(body.get("ok"), body)
|
|
self.assertEqual(str(body.get("mode") or ""), "async_task")
|
|
self.assertTrue(str(body.get("task_id") or "").strip())
|
|
|
|
def test_async_task_first_turn_renames_new_chat_title(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
created = self.client.post("/admin/api/chat/sessions", headers=headers, json={}).json()
|
|
sid = str(((created.get("session") or {}).get("id")) or "")
|
|
self.assertTrue(sid)
|
|
send = self.client.post(
|
|
f"/admin/api/chat/sessions/{sid}/messages",
|
|
headers=headers,
|
|
json={"text": "请总结并发送到项目群"},
|
|
)
|
|
self.assertEqual(send.status_code, 200)
|
|
self.assertTrue(send.json().get("ok"), send.json())
|
|
sess = self.store.get_session(sid)
|
|
self.assertIsNotNone(sess)
|
|
title = str(getattr(sess, "title", "") or "")
|
|
self.assertNotIn(title, {"新会话", "New Chat"})
|
|
|
|
def test_async_task_payload_contains_selected_specialist(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
resp = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/messages",
|
|
headers=headers,
|
|
json={"text": "请总结并发送到项目群", "chat_mode": "specialist", "specialist": "generalist"},
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertTrue(body.get("ok"), body)
|
|
self.assertEqual(str(body.get("mode") or ""), "async_task")
|
|
task_id = str(body.get("task_id") or "")
|
|
self.assertTrue(task_id.strip())
|
|
task = self.store.oclaw_task_get(task_id=task_id, tenant_id=self.tenant_id)
|
|
self.assertIsNotNone(task)
|
|
payload = json.loads(str(task.payload or "{}"))
|
|
self.assertEqual(str(payload.get("selected_specialist") or ""), "generalist")
|
|
|
|
def test_user_mode_plan_version_sets_v2_feature_flag_in_store(self) -> None:
|
|
"""POST /user-mode mirrors plan_agent_version to AIA_EXPERT_PLAN_AGENT_V2_ENABLED (v2→1, v1→0)."""
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
r2 = self.client.post(
|
|
"/admin/api/chat/user-mode",
|
|
headers=headers,
|
|
json={
|
|
"interaction_mode": "expert",
|
|
"specialist": "generalist",
|
|
"confirm_strategy": "strict",
|
|
"plan_agent_version": "v2",
|
|
},
|
|
)
|
|
self.assertEqual(r2.status_code, 200)
|
|
body2 = r2.json()
|
|
self.assertTrue(body2.get("ok"), body2)
|
|
self.assertTrue(body2.get("plan_agent_v2_globally_enabled"), body2)
|
|
self.assertEqual(str(self.store.get_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED") or "").strip(), "1")
|
|
|
|
r1 = self.client.post(
|
|
"/admin/api/chat/user-mode",
|
|
headers=headers,
|
|
json={
|
|
"interaction_mode": "expert",
|
|
"specialist": "generalist",
|
|
"confirm_strategy": "strict",
|
|
"plan_agent_version": "v1",
|
|
},
|
|
)
|
|
self.assertEqual(r1.status_code, 200)
|
|
body1 = r1.json()
|
|
self.assertTrue(body1.get("ok"), body1)
|
|
self.assertFalse(body1.get("plan_agent_v2_globally_enabled"), body1)
|
|
self.assertEqual(str(self.store.get_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED") or "").strip(), "0")
|
|
|
|
def test_session_mode_setting_roundtrip(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
_ = self.client.post(
|
|
"/admin/api/chat/user-mode",
|
|
headers=headers,
|
|
json={
|
|
"interaction_mode": "expert",
|
|
"specialist": "generalist",
|
|
"confirm_strategy": "auto",
|
|
"plan_agent_version": "v1",
|
|
},
|
|
)
|
|
resp1 = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/mode",
|
|
headers=headers,
|
|
json={"interaction_mode": "expert", "specialist": "ops", "memory_mode": "store_only"},
|
|
)
|
|
self.assertEqual(resp1.status_code, 200)
|
|
body1 = resp1.json()
|
|
self.assertTrue(body1.get("ok"), body1)
|
|
self.assertEqual(str(body1.get("interaction_mode") or ""), "expert")
|
|
# Session POST ignores interaction/specialist; user menu still has generalist.
|
|
self.assertEqual(str(body1.get("specialist") or ""), "generalist")
|
|
self.assertEqual(str(body1.get("memory_mode") or ""), "store_only")
|
|
self.assertEqual(str(body1.get("confirm_strategy") or ""), "auto")
|
|
|
|
resp2 = self.client.get(
|
|
f"/admin/api/chat/sessions/{self.session_id}/mode",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(resp2.status_code, 200)
|
|
body2 = resp2.json()
|
|
self.assertTrue(body2.get("ok"), body2)
|
|
self.assertEqual(str(body2.get("interaction_mode") or ""), "expert")
|
|
self.assertEqual(str(body2.get("specialist") or ""), "generalist")
|
|
self.assertEqual(str(body2.get("memory_mode") or ""), "store_only")
|
|
self.assertEqual(str(body2.get("confirm_strategy") or ""), "auto")
|
|
|
|
def test_messages_use_session_mode_when_payload_omits_mode(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
_ = self.client.post(
|
|
"/admin/api/chat/user-mode",
|
|
headers=headers,
|
|
json={
|
|
"interaction_mode": "expert",
|
|
"specialist": "ops",
|
|
"confirm_strategy": "strict",
|
|
"plan_agent_version": "v1",
|
|
},
|
|
)
|
|
_ = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/mode",
|
|
headers=headers,
|
|
json={"memory_mode": "store_only"},
|
|
)
|
|
resp = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/messages",
|
|
headers=headers,
|
|
json={"text": "请总结并发送到项目群"},
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertTrue(body.get("ok"), body)
|
|
task_id = str(body.get("task_id") or "")
|
|
self.assertTrue(task_id.strip())
|
|
task = self.store.oclaw_task_get(task_id=task_id, tenant_id=self.tenant_id)
|
|
self.assertIsNotNone(task)
|
|
payload = json.loads(str(task.payload or "{}"))
|
|
self.assertEqual(str(payload.get("interaction_mode") or ""), "expert")
|
|
self.assertEqual(str(payload.get("requested_specialist") or ""), "ops")
|
|
self.assertEqual(str(payload.get("memory_mode") or ""), "store_only")
|
|
|
|
def test_new_session_inherits_user_mode_preference(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
pref = self.client.post(
|
|
"/admin/api/chat/user-mode",
|
|
headers=headers,
|
|
json={
|
|
"interaction_mode": "expert",
|
|
"specialist": "ops",
|
|
"confirm_strategy": "strict",
|
|
"plan_agent_version": "v1",
|
|
},
|
|
)
|
|
self.assertEqual(pref.status_code, 200)
|
|
|
|
create_resp = self.client.post(
|
|
"/admin/api/chat/sessions",
|
|
headers=headers,
|
|
json={"title": "new one"},
|
|
)
|
|
self.assertEqual(create_resp.status_code, 200)
|
|
created = create_resp.json()
|
|
self.assertTrue(created.get("ok"), created)
|
|
new_session_id = str(((created.get("session") or {}).get("id")) or "")
|
|
self.assertTrue(new_session_id)
|
|
|
|
mode_resp = self.client.get(
|
|
f"/admin/api/chat/sessions/{new_session_id}/mode",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(mode_resp.status_code, 200)
|
|
mode_body = mode_resp.json()
|
|
self.assertTrue(mode_body.get("ok"), mode_body)
|
|
self.assertEqual(str(mode_body.get("interaction_mode") or ""), "expert")
|
|
self.assertEqual(str(mode_body.get("specialist") or ""), "ops")
|
|
self.assertEqual(str(mode_body.get("memory_mode") or ""), "default")
|
|
self.assertEqual(str(mode_body.get("execution_mode") or ""), "agent")
|
|
self.assertEqual(str(mode_body.get("confirm_strategy") or ""), "strict")
|
|
gm = mode_body.get("global_menu") if isinstance(mode_body.get("global_menu"), dict) else {}
|
|
self.assertEqual(str(gm.get("interaction_mode") or ""), "expert")
|
|
self.assertEqual(str(gm.get("specialist") or ""), "ops")
|
|
|
|
def test_new_session_default_mode_is_expert_generalist(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
create_resp = self.client.post(
|
|
"/admin/api/chat/sessions",
|
|
headers=headers,
|
|
json={"title": "default mode chat"},
|
|
)
|
|
self.assertEqual(create_resp.status_code, 200)
|
|
created = create_resp.json()
|
|
self.assertTrue(created.get("ok"), created)
|
|
new_session_id = str(((created.get("session") or {}).get("id")) or "")
|
|
self.assertTrue(new_session_id)
|
|
|
|
mode_resp = self.client.get(
|
|
f"/admin/api/chat/sessions/{new_session_id}/mode",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(mode_resp.status_code, 200)
|
|
mode_body = mode_resp.json()
|
|
self.assertTrue(mode_body.get("ok"), mode_body)
|
|
self.assertEqual(str(mode_body.get("interaction_mode") or ""), "expert")
|
|
self.assertEqual(str(mode_body.get("specialist") or ""), "generalist")
|
|
self.assertEqual(str(mode_body.get("memory_mode") or ""), "default")
|
|
self.assertEqual(str(mode_body.get("confirm_strategy") or ""), "strict")
|
|
|
|
def test_admin_dynamic_expert_stats_endpoint(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
_ = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/messages",
|
|
headers=headers,
|
|
json={
|
|
"text": "请总结并发送到项目群",
|
|
"interaction_mode": "comprehensive",
|
|
"specialist": "generalist",
|
|
},
|
|
)
|
|
resp = self.client.get(
|
|
"/admin/api/chat/admin/dynamic-expert-stats?limit=200",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertTrue(body.get("ok"), body)
|
|
self.assertIn("dynamic_used_count", body)
|
|
self.assertIn("fallback_generalist_count", body)
|
|
self.assertIn("dynamic_used_rate", body)
|
|
self.assertIn("dispatch_reasons", body)
|
|
self.assertIn("dispatch_reason_labels", body)
|
|
|
|
def test_admin_dynamic_expert_stats_labels_override_setting(self) -> None:
|
|
token = self._login()
|
|
self.store.set_setting(
|
|
"AIA_DISPATCH_REASON_LABELS_JSON",
|
|
json.dumps(
|
|
{
|
|
"manager_no_specialist_fallback": {
|
|
"zh": "自定义回退文案",
|
|
"en": "Custom fallback label",
|
|
}
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
_ = self.client.post(
|
|
f"/admin/api/chat/sessions/{self.session_id}/messages",
|
|
headers=headers,
|
|
json={"text": "请总结并发送到项目群", "interaction_mode": "comprehensive", "specialist": "generalist"},
|
|
)
|
|
resp = self.client.get(
|
|
"/admin/api/chat/admin/dynamic-expert-stats?limit=200",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertTrue(body.get("ok"), body)
|
|
labels = body.get("dispatch_reason_labels") or {}
|
|
if "manager_no_specialist_fallback" in labels:
|
|
self.assertEqual(str(labels.get("manager_no_specialist_fallback") or ""), "自定义回退文案")
|
|
|
|
def test_admin_dispatch_reason_labels_settings_api_roundtrip(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
set_resp = self.client.post(
|
|
"/admin/api/chat/settings/dispatch-reason-labels",
|
|
headers=headers,
|
|
json={
|
|
"overrides": {
|
|
"manager_select_failed": {"zh": "自定义总控失败", "en": "Custom manager failed"}
|
|
}
|
|
},
|
|
)
|
|
self.assertEqual(set_resp.status_code, 200)
|
|
set_body = set_resp.json()
|
|
self.assertTrue(set_body.get("ok"), set_body)
|
|
effective = set_body.get("effective") or {}
|
|
row = effective.get("manager_select_failed") or {}
|
|
self.assertEqual(str(row.get("zh") or ""), "自定义总控失败")
|
|
|
|
get_resp = self.client.get(
|
|
"/admin/api/chat/settings/dispatch-reason-labels",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(get_resp.status_code, 200)
|
|
get_body = get_resp.json()
|
|
self.assertTrue(get_body.get("ok"), get_body)
|
|
overrides = get_body.get("overrides") or {}
|
|
self.assertIn("manager_select_failed", overrides)
|
|
|
|
def test_admin_attachment_limits_settings_api_roundtrip(self) -> None:
|
|
token = self._login()
|
|
headers = {
|
|
"authorization": f"Bearer {token}",
|
|
"accept": "application/json",
|
|
"content-type": "application/json",
|
|
}
|
|
get_resp = self.client.get(
|
|
"/admin/api/chat/settings/attachment-limits",
|
|
headers={"authorization": f"Bearer {token}", "accept": "application/json"},
|
|
)
|
|
self.assertEqual(get_resp.status_code, 200)
|
|
get_body = get_resp.json()
|
|
self.assertTrue(get_body.get("ok"), get_body)
|
|
self.assertEqual(int((get_body.get("limits") or {}).get("max_rows_read") or 0), 5000)
|
|
self.assertEqual(int((get_body.get("limits") or {}).get("sql_timeout_ms") or 0), 8000)
|
|
# newly added replay caps should exist with sane defaults
|
|
self.assertGreaterEqual(int((get_body.get("limits") or {}).get("image_result_replay_cap_chars") or 0), 600)
|
|
self.assertGreaterEqual(int((get_body.get("limits") or {}).get("video_result_replay_cap_chars") or 0), 600)
|
|
self.assertGreaterEqual(int((get_body.get("limits") or {}).get("video_transcript_chunk_size") or 0), 1)
|
|
self.assertGreaterEqual(int((get_body.get("limits") or {}).get("video_transcript_chunk_overlap") or 0), 0)
|
|
self.assertGreaterEqual(int((get_body.get("limits") or {}).get("archive_max_depth") or 0), 1)
|
|
self.assertGreaterEqual(int((get_body.get("limits") or {}).get("archive_max_file_count") or 0), 1)
|
|
|
|
set_resp = self.client.post(
|
|
"/admin/api/chat/settings/attachment-limits",
|
|
headers=headers,
|
|
json={
|
|
"limits": {
|
|
"max_rows_read": 123,
|
|
"max_columns": 45,
|
|
"max_cell_chars": 67,
|
|
"max_excel_sheets": 8,
|
|
"large_table_preview_rows": 33,
|
|
"sql_timeout_ms": 1234,
|
|
"video_transcript_chunk_size": 1700,
|
|
"video_transcript_chunk_overlap": 240,
|
|
"archive_max_depth": 3,
|
|
"archive_max_file_count": 555,
|
|
"archive_max_entry_bytes": 123456,
|
|
"archive_max_total_uncompressed_bytes": 654321,
|
|
}
|
|
},
|
|
)
|
|
self.assertEqual(set_resp.status_code, 200)
|
|
set_body = set_resp.json()
|
|
self.assertTrue(set_body.get("ok"), set_body)
|
|
limits = set_body.get("limits") or {}
|
|
self.assertEqual(int(limits.get("max_rows_read") or 0), 123)
|
|
self.assertEqual(int(limits.get("max_columns") or 0), 45)
|
|
self.assertEqual(int(limits.get("max_cell_chars") or 0), 67)
|
|
self.assertEqual(int(limits.get("max_excel_sheets") or 0), 8)
|
|
self.assertEqual(int(limits.get("large_table_preview_rows") or 0), 33)
|
|
self.assertEqual(int(limits.get("sql_timeout_ms") or 0), 1234)
|
|
self.assertEqual(int(limits.get("video_transcript_chunk_size") or 0), 1700)
|
|
self.assertEqual(int(limits.get("video_transcript_chunk_overlap") or 0), 240)
|
|
self.assertEqual(int(limits.get("archive_max_depth") or 0), 3)
|
|
self.assertEqual(int(limits.get("archive_max_file_count") or 0), 555)
|
|
self.assertEqual(int(limits.get("archive_max_entry_bytes") or 0), 123456)
|
|
self.assertEqual(int(limits.get("archive_max_total_uncompressed_bytes") or 0), 654321)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|