wehub-resource-sync
938 lines
30 KiB
Python
938 lines
30 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import free_claude_code.messaging.session.persistence as persistence_module
|
|
from free_claude_code.config.admin.persistence import PreparedAdminUpdate
|
|
from free_claude_code.config.settings import Settings
|
|
from free_claude_code.messaging.command_context import StopOutcome
|
|
from free_claude_code.messaging.platforms.ports import (
|
|
InboundMessageHandler,
|
|
MessagingPlatformComponents,
|
|
MessagingStartupNotice,
|
|
)
|
|
from free_claude_code.messaging.session import SessionStore
|
|
from free_claude_code.messaging.workflow import MessagingWorkflow
|
|
from free_claude_code.providers.runtime import ProviderRuntime
|
|
from free_claude_code.runtime.application import ApplicationRuntime
|
|
from free_claude_code.runtime.provider_manager import ProviderRuntimeManager
|
|
|
|
|
|
class TrackingRuntime(ProviderRuntime):
|
|
def __init__(self, settings: Settings) -> None:
|
|
super().__init__(settings)
|
|
self.cleanup_calls = 0
|
|
|
|
async def cleanup(self) -> None:
|
|
self.cleanup_calls += 1
|
|
await super().cleanup()
|
|
|
|
|
|
class TrackingFactory:
|
|
def __init__(self) -> None:
|
|
self.runtimes: list[TrackingRuntime] = []
|
|
self.fail = False
|
|
self.events: list[str] = []
|
|
|
|
def __call__(self, settings: Settings) -> ProviderRuntime:
|
|
self.events.append(f"construct:{settings.model}")
|
|
if self.fail:
|
|
raise RuntimeError("candidate failed")
|
|
runtime = TrackingRuntime(settings)
|
|
self.runtimes.append(runtime)
|
|
return runtime
|
|
|
|
|
|
class TrackingTranscriber:
|
|
def __init__(self, events: list[str]) -> None:
|
|
self.events = events
|
|
self.close_calls = 0
|
|
|
|
async def transcribe(self, file_path: Path) -> str:
|
|
assert isinstance(file_path, Path)
|
|
return "transcribed"
|
|
|
|
async def close(self) -> None:
|
|
self.close_calls += 1
|
|
self.events.append("transcriber.close")
|
|
|
|
|
|
class FailingTranscriber(TrackingTranscriber):
|
|
async def close(self) -> None:
|
|
await super().close()
|
|
raise RuntimeError("transcriber close failed")
|
|
|
|
|
|
class CancelledTranscriber(TrackingTranscriber):
|
|
async def close(self) -> None:
|
|
await super().close()
|
|
raise asyncio.CancelledError
|
|
|
|
|
|
class CancellingOnceTranscriber(TrackingTranscriber):
|
|
async def close(self) -> None:
|
|
await super().close()
|
|
if self.close_calls == 1:
|
|
raise asyncio.CancelledError
|
|
|
|
|
|
class TrackingMessagingRuntime:
|
|
name = "tracking"
|
|
|
|
def __init__(
|
|
self,
|
|
events: list[str],
|
|
*,
|
|
fail_quiesce_once: bool = False,
|
|
fail_close_once: bool = False,
|
|
) -> None:
|
|
self.events = events
|
|
self.fail_quiesce_once = fail_quiesce_once
|
|
self.fail_close_once = fail_close_once
|
|
|
|
async def start(self) -> None:
|
|
self.events.append("messaging.start")
|
|
|
|
async def quiesce(self) -> None:
|
|
self.events.append("messaging.quiesce")
|
|
if self.fail_quiesce_once:
|
|
self.fail_quiesce_once = False
|
|
raise RuntimeError("quiesce failed")
|
|
|
|
async def close(self) -> None:
|
|
self.events.append("messaging.close")
|
|
if self.fail_close_once:
|
|
self.fail_close_once = False
|
|
raise RuntimeError("close failed")
|
|
|
|
def on_message(self, handler: InboundMessageHandler) -> None:
|
|
assert callable(handler)
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return True
|
|
|
|
|
|
class PersistentlyFailingMessagingRuntime(TrackingMessagingRuntime):
|
|
def __init__(self, events: list[str]) -> None:
|
|
super().__init__(events)
|
|
self.fail_quiesce = True
|
|
|
|
async def quiesce(self) -> None:
|
|
self.events.append("messaging.quiesce")
|
|
if self.fail_quiesce:
|
|
raise RuntimeError("quiesce failed")
|
|
|
|
|
|
def _settings(model: str, *, port: int = 8082) -> Settings:
|
|
return Settings().model_copy(update={"model": model, "port": port})
|
|
|
|
|
|
def _prepared(
|
|
settings: Settings,
|
|
tmp_path,
|
|
*,
|
|
pending_fields: tuple[str, ...] = (),
|
|
) -> PreparedAdminUpdate:
|
|
return PreparedAdminUpdate(
|
|
target_values={"MODEL": settings.model},
|
|
settings=settings,
|
|
errors=(),
|
|
pending_fields=pending_fields,
|
|
path=tmp_path / ".env",
|
|
)
|
|
|
|
|
|
def _applied_response(pending_fields: tuple[str, ...] = ()) -> dict[str, object]:
|
|
return {
|
|
"applied": True,
|
|
"valid": True,
|
|
"errors": [],
|
|
"env_preview": "MODEL=updated\n",
|
|
"path": ".env",
|
|
"pending_fields": list(pending_fields),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_all_maps_messaging_outcome_to_application_count() -> None:
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
workflow = MagicMock()
|
|
workflow.stop_all_tasks = AsyncMock(
|
|
return_value=StopOutcome(
|
|
cancelled_count=3,
|
|
status_feedback_scopes=frozenset(),
|
|
fallback_required=False,
|
|
)
|
|
)
|
|
runtime._messaging_workflow = workflow
|
|
|
|
result = await runtime.stop_all()
|
|
|
|
assert result is not None
|
|
assert result.cancelled_count == 3
|
|
workflow.stop_all_tasks.assert_awaited_once()
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_apply_constructs_before_commit_then_publishes(tmp_path) -> None:
|
|
factory = TrackingFactory()
|
|
manager = ProviderRuntimeManager(
|
|
_settings("nvidia_nim/old"),
|
|
runtime_factory=factory,
|
|
)
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
prepared = _prepared(_settings("nvidia_nim/new"), tmp_path)
|
|
factory.events.clear()
|
|
|
|
def commit(_prepared_update: PreparedAdminUpdate) -> dict[str, object]:
|
|
factory.events.append("commit")
|
|
assert manager.current_generation_id == 1
|
|
return _applied_response()
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.prepare_admin_update",
|
|
return_value=prepared,
|
|
),
|
|
patch(
|
|
"free_claude_code.runtime.application.commit_prepared_admin_update",
|
|
side_effect=commit,
|
|
),
|
|
):
|
|
result = await runtime.apply_admin_config({"MODEL": "nvidia_nim/new"})
|
|
|
|
assert factory.events == ["construct:nvidia_nim/new", "commit"]
|
|
assert manager.current_generation_id == 2
|
|
assert manager.current_settings().model == "nvidia_nim/new"
|
|
assert result["restart"] == {
|
|
"required": False,
|
|
"automatic": False,
|
|
"admin_url": None,
|
|
"fields": [],
|
|
}
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_candidate_failure_never_commits_and_preserves_current(tmp_path) -> None:
|
|
factory = TrackingFactory()
|
|
manager = ProviderRuntimeManager(
|
|
_settings("nvidia_nim/old"),
|
|
runtime_factory=factory,
|
|
)
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
prepared = _prepared(_settings("nvidia_nim/new"), tmp_path)
|
|
factory.fail = True
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.prepare_admin_update",
|
|
return_value=prepared,
|
|
),
|
|
patch(
|
|
"free_claude_code.runtime.application.commit_prepared_admin_update"
|
|
) as commit,
|
|
pytest.raises(RuntimeError, match="candidate failed"),
|
|
):
|
|
await runtime.apply_admin_config({"MODEL": "nvidia_nim/new"})
|
|
|
|
commit.assert_not_called()
|
|
assert manager.current_generation_id == 1
|
|
assert manager.current_settings().model == "nvidia_nim/old"
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_persistence_failure_closes_candidate_and_preserves_current(
|
|
tmp_path,
|
|
) -> None:
|
|
factory = TrackingFactory()
|
|
manager = ProviderRuntimeManager(
|
|
_settings("nvidia_nim/old"),
|
|
runtime_factory=factory,
|
|
)
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
prepared = _prepared(_settings("nvidia_nim/new"), tmp_path)
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.prepare_admin_update",
|
|
return_value=prepared,
|
|
),
|
|
patch(
|
|
"free_claude_code.runtime.application.commit_prepared_admin_update",
|
|
side_effect=OSError("disk full"),
|
|
),
|
|
pytest.raises(OSError, match="disk full"),
|
|
):
|
|
await runtime.apply_admin_config({"MODEL": "nvidia_nim/new"})
|
|
|
|
assert manager.current_generation_id == 1
|
|
assert factory.runtimes[0].cleanup_calls == 0
|
|
assert factory.runtimes[1].cleanup_calls == 1
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restart_required_apply_commits_without_hot_publication(tmp_path) -> None:
|
|
factory = TrackingFactory()
|
|
manager = ProviderRuntimeManager(
|
|
_settings("nvidia_nim/old"),
|
|
runtime_factory=factory,
|
|
)
|
|
restart = AsyncMock()
|
|
runtime = ApplicationRuntime(
|
|
manager,
|
|
transcriber=None,
|
|
restart_callback=restart,
|
|
)
|
|
prepared = _prepared(
|
|
_settings("nvidia_nim/old", port=9090),
|
|
tmp_path,
|
|
pending_fields=("PORT",),
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.prepare_admin_update",
|
|
return_value=prepared,
|
|
),
|
|
patch(
|
|
"free_claude_code.runtime.application.commit_prepared_admin_update",
|
|
return_value=_applied_response(("PORT",)),
|
|
) as commit,
|
|
):
|
|
result = await runtime.apply_admin_config({"PORT": "9090"})
|
|
|
|
commit.assert_called_once_with(prepared)
|
|
assert manager.current_generation_id == 1
|
|
assert len(factory.runtimes) == 1
|
|
assert result["restart"] == {
|
|
"required": True,
|
|
"automatic": True,
|
|
"admin_url": "http://127.0.0.1:9090/admin",
|
|
"fields": ["PORT"],
|
|
}
|
|
restart.assert_not_awaited()
|
|
await runtime.request_restart()
|
|
restart.assert_awaited_once()
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_drains_messaging_before_transcriber_and_is_idempotent() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = TrackingTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
runtime._messaging_runtime = TrackingMessagingRuntime(events)
|
|
workflow = MagicMock()
|
|
workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close"))
|
|
runtime._messaging_workflow = workflow
|
|
runtime._cli_manager = MagicMock()
|
|
|
|
assert runtime.is_closed is False
|
|
assert await runtime.close() is True
|
|
assert await runtime.close() is True
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"workflow.close",
|
|
"messaging.close",
|
|
"transcriber.close",
|
|
]
|
|
assert transcriber.close_calls == 1
|
|
assert runtime._transcriber is None
|
|
assert runtime._messaging_runtime is None
|
|
assert runtime._messaging_workflow is None
|
|
assert runtime.is_closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_retains_transcriber_ownership_when_close_fails() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = FailingTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
runtime._messaging_runtime = TrackingMessagingRuntime(events)
|
|
|
|
assert await runtime.close() is False
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.close",
|
|
"transcriber.close",
|
|
]
|
|
assert transcriber.close_calls == 1
|
|
assert runtime._transcriber is transcriber
|
|
assert runtime._closed is False
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_retries_runtime_before_closing_later_resources() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = TrackingTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
messaging = TrackingMessagingRuntime(events, fail_close_once=True)
|
|
runtime._messaging_runtime = messaging
|
|
|
|
assert await runtime.close() is False
|
|
|
|
assert events == ["messaging.quiesce", "messaging.close"]
|
|
assert runtime._messaging_runtime is messaging
|
|
assert runtime._transcriber is transcriber
|
|
assert runtime._closed is False
|
|
|
|
assert await runtime.close() is True
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.close",
|
|
"messaging.quiesce",
|
|
"messaging.close",
|
|
"transcriber.close",
|
|
]
|
|
assert runtime._messaging_runtime is None
|
|
assert runtime._transcriber is None
|
|
assert runtime._closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_retries_workflow_close_before_closing_delivery() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events)
|
|
workflow = MagicMock()
|
|
close_calls = 0
|
|
|
|
async def close_workflow() -> None:
|
|
nonlocal close_calls
|
|
close_calls += 1
|
|
if close_calls == 1:
|
|
raise RuntimeError("drain failed")
|
|
events.append("workflow.close")
|
|
|
|
workflow.close = AsyncMock(side_effect=close_workflow)
|
|
runtime._messaging_runtime = messaging
|
|
runtime._messaging_workflow = workflow
|
|
|
|
assert await runtime.close() is False
|
|
|
|
assert events == ["messaging.quiesce"]
|
|
assert runtime._messaging_runtime is messaging
|
|
assert runtime._messaging_workflow is workflow
|
|
assert runtime._closed is False
|
|
|
|
assert await runtime.close() is True
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"workflow.close",
|
|
"messaging.close",
|
|
]
|
|
assert runtime._closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_does_not_drain_workflow_until_ingress_is_quiescent() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events, fail_quiesce_once=True)
|
|
workflow = MagicMock()
|
|
workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close"))
|
|
runtime._messaging_runtime = messaging
|
|
runtime._messaging_workflow = workflow
|
|
|
|
assert await runtime.close() is False
|
|
|
|
workflow.close.assert_not_awaited()
|
|
assert runtime._messaging_runtime is messaging
|
|
assert runtime._messaging_workflow is workflow
|
|
|
|
assert await runtime.close() is True
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"workflow.close",
|
|
"messaging.close",
|
|
]
|
|
assert runtime._closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_retries_failed_persistence_before_closing_delivery() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events)
|
|
workflow = MagicMock()
|
|
close_calls = 0
|
|
|
|
async def close_workflow() -> None:
|
|
nonlocal close_calls
|
|
close_calls += 1
|
|
if close_calls == 1:
|
|
raise RuntimeError("flush failed")
|
|
events.append("workflow.close")
|
|
|
|
workflow.close = AsyncMock(side_effect=close_workflow)
|
|
runtime._messaging_runtime = messaging
|
|
runtime._messaging_workflow = workflow
|
|
|
|
await runtime.close()
|
|
|
|
assert runtime._messaging_workflow is workflow
|
|
assert runtime._messaging_runtime is messaging
|
|
assert "messaging.close" not in events
|
|
|
|
await runtime.close()
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"workflow.close",
|
|
"messaging.close",
|
|
]
|
|
assert runtime._closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_retries_real_workflow_persistence_without_losing_latest_state(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events)
|
|
cli_manager = MagicMock()
|
|
cli_manager.stop_all = AsyncMock()
|
|
outbound = MagicMock()
|
|
store_path = tmp_path / "sessions.json"
|
|
real_replace = persistence_module.os.replace
|
|
replace_calls = 0
|
|
|
|
def fail_first_replace(source: str, target: str) -> None:
|
|
nonlocal replace_calls
|
|
replace_calls += 1
|
|
if replace_calls == 1:
|
|
raise OSError("replace failed once")
|
|
real_replace(source, target)
|
|
|
|
with patch.object(persistence_module.threading, "Timer"):
|
|
store = SessionStore(storage_path=str(store_path))
|
|
store.record_message_id(
|
|
"telegram",
|
|
"chat_1",
|
|
"old",
|
|
"out",
|
|
"status",
|
|
)
|
|
store.flush_pending_save()
|
|
store.record_message_id(
|
|
"telegram",
|
|
"chat_1",
|
|
"latest",
|
|
"out",
|
|
"status",
|
|
)
|
|
workflow = MessagingWorkflow(outbound, cli_manager, store)
|
|
runtime._messaging_runtime = messaging
|
|
runtime._messaging_workflow = workflow
|
|
runtime._cli_manager = cli_manager
|
|
|
|
with patch.object(
|
|
persistence_module.os,
|
|
"replace",
|
|
side_effect=fail_first_replace,
|
|
):
|
|
assert await runtime.close() is False
|
|
|
|
assert runtime._messaging_runtime is messaging
|
|
assert runtime._messaging_workflow is workflow
|
|
assert runtime._cli_manager is cli_manager
|
|
assert runtime.is_closed is False
|
|
assert store.dirty is True
|
|
assert store.get_tracked_message_ids_for_chat("telegram", "chat_1") == [
|
|
"old",
|
|
"latest",
|
|
]
|
|
assert SessionStore(
|
|
storage_path=str(store_path)
|
|
).get_tracked_message_ids_for_chat("telegram", "chat_1") == ["old"]
|
|
|
|
assert await runtime.close() is True
|
|
|
|
assert runtime._messaging_runtime is None
|
|
assert runtime._messaging_workflow is None
|
|
assert runtime._cli_manager is None
|
|
assert runtime.is_closed is True
|
|
assert store.dirty is False
|
|
assert SessionStore(
|
|
storage_path=str(store_path)
|
|
).get_tracked_message_ids_for_chat("telegram", "chat_1") == [
|
|
"old",
|
|
"latest",
|
|
]
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"messaging.close",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_transcriber_close_retains_ownership() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = CancelledTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await runtime._cleanup_transcriber()
|
|
|
|
assert transcriber.close_calls == 1
|
|
assert runtime._transcriber is transcriber
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_application_close_remains_retryable() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = CancellingOnceTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await runtime.close()
|
|
|
|
assert runtime._closed is False
|
|
assert runtime._transcriber is transcriber
|
|
|
|
await runtime.close()
|
|
|
|
assert transcriber.close_calls == 2
|
|
assert runtime._transcriber is None
|
|
assert runtime._closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_failure_closes_owned_transcriber() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = TrackingTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
|
|
with (
|
|
patch.object(
|
|
manager,
|
|
"validate_configured_models",
|
|
AsyncMock(side_effect=RuntimeError("startup failed")),
|
|
),
|
|
pytest.raises(RuntimeError, match="startup failed"),
|
|
):
|
|
await runtime.start()
|
|
|
|
assert transcriber.close_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_cancellation_cleans_partial_messaging_and_reraises() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = TrackingTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
messaging = TrackingMessagingRuntime(events)
|
|
entered = asyncio.Event()
|
|
|
|
async def start_messaging() -> None:
|
|
runtime._messaging_runtime = messaging
|
|
entered.set()
|
|
await asyncio.Event().wait()
|
|
|
|
with patch.object(
|
|
runtime,
|
|
"_start_messaging_if_configured",
|
|
side_effect=start_messaging,
|
|
):
|
|
start_task = asyncio.create_task(runtime.start())
|
|
await entered.wait()
|
|
start_task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await start_task
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.close",
|
|
"transcriber.close",
|
|
]
|
|
assert runtime._closed is True
|
|
assert runtime._messaging_runtime is None
|
|
assert runtime._transcriber is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_start_retries_transient_partial_messaging_cleanup() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events, fail_quiesce_once=True)
|
|
workflow = MagicMock()
|
|
workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close"))
|
|
cli_manager = MagicMock()
|
|
components = MessagingPlatformComponents(
|
|
name="tracking",
|
|
runtime=messaging,
|
|
outbound=MagicMock(),
|
|
)
|
|
startup_failure = RuntimeError("partial messaging startup failed")
|
|
|
|
async def fail_after_publication(
|
|
published: MessagingPlatformComponents,
|
|
) -> None:
|
|
assert published is components
|
|
runtime._messaging_runtime = messaging
|
|
runtime._messaging_workflow = workflow
|
|
runtime._cli_manager = cli_manager
|
|
raise startup_failure
|
|
|
|
with (
|
|
patch.object(
|
|
runtime,
|
|
"_validate_configured_models_best_effort",
|
|
AsyncMock(),
|
|
),
|
|
patch.object(manager, "start_model_list_refresh"),
|
|
patch(
|
|
"free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components",
|
|
return_value=components,
|
|
),
|
|
patch.object(
|
|
runtime,
|
|
"_start_messaging_workflow",
|
|
side_effect=fail_after_publication,
|
|
),
|
|
pytest.raises(RuntimeError, match="cleanup incomplete") as raised,
|
|
):
|
|
await runtime.start()
|
|
|
|
assert raised.value.__cause__ is startup_failure
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"workflow.close",
|
|
"messaging.close",
|
|
]
|
|
workflow.close.assert_awaited_once()
|
|
assert runtime._messaging_runtime is None
|
|
assert runtime._messaging_workflow is None
|
|
assert runtime._cli_manager is None
|
|
assert runtime.is_closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_start_retains_persistently_unclean_partial_messaging_graph() -> (
|
|
None
|
|
):
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
transcriber = TrackingTranscriber(events)
|
|
runtime = ApplicationRuntime(manager, transcriber=transcriber)
|
|
messaging = PersistentlyFailingMessagingRuntime(events)
|
|
workflow = MagicMock()
|
|
workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close"))
|
|
cli_manager = MagicMock()
|
|
components = MessagingPlatformComponents(
|
|
name="tracking",
|
|
runtime=messaging,
|
|
outbound=MagicMock(),
|
|
)
|
|
startup_failure = RuntimeError("partial messaging startup failed")
|
|
|
|
async def fail_after_publication(
|
|
published: MessagingPlatformComponents,
|
|
) -> None:
|
|
assert published is components
|
|
runtime._messaging_runtime = messaging
|
|
runtime._messaging_workflow = workflow
|
|
runtime._cli_manager = cli_manager
|
|
raise startup_failure
|
|
|
|
with (
|
|
patch.object(
|
|
runtime,
|
|
"_validate_configured_models_best_effort",
|
|
AsyncMock(),
|
|
),
|
|
patch.object(manager, "start_model_list_refresh"),
|
|
patch(
|
|
"free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components",
|
|
return_value=components,
|
|
),
|
|
patch.object(
|
|
runtime,
|
|
"_start_messaging_workflow",
|
|
side_effect=fail_after_publication,
|
|
),
|
|
pytest.raises(RuntimeError, match="cleanup incomplete") as raised,
|
|
):
|
|
await runtime.start()
|
|
|
|
assert raised.value.__cause__ is startup_failure
|
|
assert events == ["messaging.quiesce", "messaging.quiesce"]
|
|
workflow.close.assert_not_awaited()
|
|
assert transcriber.close_calls == 0
|
|
assert runtime._messaging_runtime is messaging
|
|
assert runtime._messaging_workflow is workflow
|
|
assert runtime._cli_manager is cli_manager
|
|
assert runtime._transcriber is transcriber
|
|
assert runtime._provider_manager_closed is False
|
|
assert runtime.is_closed is False
|
|
|
|
messaging.fail_quiesce = False
|
|
assert await runtime.close() is True
|
|
|
|
assert events == [
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"messaging.quiesce",
|
|
"workflow.close",
|
|
"messaging.close",
|
|
"transcriber.close",
|
|
]
|
|
assert runtime.is_closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_messaging_start_failure_is_nonfatal_after_complete_cleanup() -> None:
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components",
|
|
side_effect=RuntimeError("messaging unavailable"),
|
|
),
|
|
patch.object(runtime, "_cleanup_messaging", AsyncMock(return_value=True)),
|
|
):
|
|
await runtime._start_messaging_if_configured()
|
|
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_messaging_start_failure_fails_closed_when_cleanup_is_incomplete() -> (
|
|
None
|
|
):
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components",
|
|
side_effect=RuntimeError("messaging unavailable"),
|
|
),
|
|
patch.object(runtime, "_cleanup_messaging", AsyncMock(return_value=False)),
|
|
pytest.raises(RuntimeError, match="cleanup incomplete") as exc_info,
|
|
):
|
|
await runtime._start_messaging_if_configured()
|
|
|
|
assert isinstance(exc_info.value.__cause__, RuntimeError)
|
|
await manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_composition_records_runtime_before_workspace_setup() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events)
|
|
components = MessagingPlatformComponents(
|
|
name="tracking",
|
|
runtime=messaging,
|
|
outbound=MagicMock(),
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.os.makedirs",
|
|
side_effect=OSError("workspace failed"),
|
|
),
|
|
pytest.raises(OSError, match="workspace failed"),
|
|
):
|
|
await runtime._start_messaging_workflow(components)
|
|
|
|
assert runtime._messaging_runtime is messaging
|
|
|
|
await runtime.close()
|
|
|
|
assert events == ["messaging.quiesce", "messaging.close"]
|
|
assert runtime._closed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_composition_publishes_startup_notice_after_runtime_and_repair() -> None:
|
|
events: list[str] = []
|
|
manager = ProviderRuntimeManager(_settings("nvidia_nim/model"))
|
|
runtime = ApplicationRuntime(manager, transcriber=None)
|
|
messaging = TrackingMessagingRuntime(events)
|
|
notice = MessagingStartupNotice(
|
|
chat_id="chat",
|
|
transport_label="test transport",
|
|
)
|
|
components = MessagingPlatformComponents(
|
|
name="tracking",
|
|
runtime=messaging,
|
|
outbound=MagicMock(),
|
|
startup_notice=notice,
|
|
)
|
|
workflow = MagicMock()
|
|
workflow.handle_message = AsyncMock()
|
|
workflow.restore.side_effect = lambda: events.append("workflow.restore")
|
|
workflow.repair_restored_statuses = AsyncMock(
|
|
side_effect=lambda: events.append("workflow.repair")
|
|
)
|
|
workflow.publish_startup_notice = AsyncMock(
|
|
side_effect=lambda published: events.append("workflow.notice")
|
|
)
|
|
workflow.close = AsyncMock()
|
|
cli_manager = MagicMock()
|
|
|
|
with (
|
|
patch(
|
|
"free_claude_code.runtime.application.cli_managed.ManagedClaudeSessionManager",
|
|
return_value=cli_manager,
|
|
) as manager_constructor,
|
|
patch("free_claude_code.runtime.application.messaging_session.SessionStore"),
|
|
patch(
|
|
"free_claude_code.runtime.application.messaging_workflow_module.MessagingWorkflow",
|
|
return_value=workflow,
|
|
),
|
|
):
|
|
await runtime._start_messaging_workflow(components)
|
|
|
|
assert events == [
|
|
"workflow.restore",
|
|
"messaging.start",
|
|
"workflow.repair",
|
|
"workflow.notice",
|
|
]
|
|
workflow.publish_startup_notice.assert_awaited_once_with(notice)
|
|
assert manager_constructor.call_args.kwargs["proxy_root_url"] == (
|
|
"http://127.0.0.1:8082"
|
|
)
|
|
assert "api_url" not in manager_constructor.call_args.kwargs
|
|
assert "plans_directory" not in manager_constructor.call_args.kwargs
|
|
|
|
assert await runtime.close() is True
|