mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-21 06:29:24 +08:00
jubaoliang
7eeeef778c
Linux octop run now ensures bubblewrap in the background without blocking boot. Setup and invite default agents use the same home-scoped local backend as dashboard-created experts. Chat task progress sorts by original numeric ids instead of status, so completed steps stay in plan order.
69 lines
1.8 KiB
Python
69 lines
1.8 KiB
Python
"""Tests for non-blocking bubblewrap provisioning in ``octop run``."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from octop import launch
|
|
|
|
|
|
@pytest.mark.parametrize("platform", ["darwin", "win32"])
|
|
def test_schedule_skips_non_linux(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
platform: str,
|
|
) -> None:
|
|
monkeypatch.setattr(launch.sys, "platform", platform)
|
|
assert launch._schedule_linux_bubblewrap_ensure() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_runs_linux_ensure_in_background(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def ensure() -> None:
|
|
started.set()
|
|
release.wait(timeout=5)
|
|
|
|
monkeypatch.setattr(launch.sys, "platform", "linux")
|
|
monkeypatch.setattr(launch, "_ensure_linux_bubblewrap", ensure)
|
|
|
|
task = launch._schedule_linux_bubblewrap_ensure()
|
|
assert task is not None
|
|
try:
|
|
assert await asyncio.to_thread(started.wait, 1)
|
|
assert not task.done()
|
|
finally:
|
|
release.set()
|
|
await task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_background_task_is_noop_when_missing() -> None:
|
|
await launch._cancel_background_task(None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_background_task_drops_pending() -> None:
|
|
started = asyncio.Event()
|
|
|
|
async def hang() -> None:
|
|
started.set()
|
|
await asyncio.sleep(60)
|
|
|
|
task = asyncio.create_task(hang())
|
|
await started.wait()
|
|
await launch._cancel_background_task(task)
|
|
assert task.cancelled() or task.done()
|
|
|
|
|
|
def test_background_ensure_swallows_failures(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def fail() -> dict[str, str]:
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("octop.infra.utils.bwrap.ensure_bubblewrap", fail)
|
|
launch._ensure_linux_bubblewrap()
|