yunxiafei/openclaw-mission-control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: rename compatibility check function and update version extraction logic #159

Browse Source
This commit is contained in:
Abhimanyu Saharan
2026-02-22 22:05:39 +05:30
parent 6f4ad41dba
commit 6b09f124e6
4 changed files with 115 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
backend/tests/test_gateway_version_compat.py
View File

@@ -16,147 +16,153 @@ from app.services.openclaw.gateway_rpc import GatewayConfig, OpenClawGatewayErro
from app.services.openclaw.session_service import GatewaySessionService
def test_extract_gateway_version_prefers_primary_path() -> None:
def test_extract_connect_server_version_uses_server_version_as_source_of_truth() -> None:
payload = {
"gateway": {"version": "2026.2.1"},
"protocolVersion": 3,
"meta": {"version": "2026.1.30"},
"version": "dev",
"runtime": {"version": "2026.1.0"},
"server": {"version": "2026.2.21-2"},
}
assert gateway_compat.extract_gateway_version(payload) == "2026.2.1"
assert gateway_compat.extract_connect_server_version(payload) == "2026.2.21-2"
def test_evaluate_gateway_version_detects_old_runtime() -> None:
def test_extract_connect_server_version_returns_none_when_server_version_missing() -> None:
payload = {
"version": "2026.2.21-2",
"runtime": {"version": "2026.2.21-2"},
}
assert gateway_compat.extract_connect_server_version(payload) is None
@pytest.mark.parametrize(
("current_version", "minimum_version", "expected_compatible"),
[
("2026.2.21", "2026.2.21", True),
("2026.02.20", "2026.2.20", True),
("2026.2.22", "2026.2.21", True),
("2026.2.21-2", "2026.2.21-1", True),
("2026.2.21-1", "2026.2.21-2", False),
("2026.2.20", "2026.2.21", False),
],
)
def test_evaluate_gateway_version_compares_calver(
*,
current_version: str,
minimum_version: str,
expected_compatible: bool,
) -> None:
result = gateway_compat.evaluate_gateway_version(
current_version="2025.12.1",
current_version=current_version,
minimum_version=minimum_version,
)
assert result.compatible is expected_compatible
assert result.current_version == current_version
assert result.minimum_version == minimum_version
@pytest.mark.parametrize("invalid_current", ["dev", "latest", "2026.13.1", "2026.2.0-beta"])
def test_evaluate_gateway_version_rejects_non_calver_current(invalid_current: str) -> None:
result = gateway_compat.evaluate_gateway_version(
current_version=invalid_current,
minimum_version="2026.1.30",
)
assert result.compatible is False
assert result.minimum_version == "2026.1.30"
assert "Minimum supported version is 2026.1.30" in (result.message or "")
assert result.current_version == invalid_current
assert "unsupported version format" in (result.message or "").lower()
def test_evaluate_gateway_version_rejects_non_calver_minimum_version() -> None:
result = gateway_compat.evaluate_gateway_version(
current_version="2026.2.21",
minimum_version="1.2.3",
)
assert result.compatible is False
assert result.minimum_version == "1.2.3"
assert "expected calver" in (result.message or "").lower()
@pytest.mark.asyncio
async def test_check_gateway_runtime_compatibility_prefers_schema_version(
async def test_check_gateway_version_compatibility_uses_connect_server_version_only(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
async def _fake_openclaw_call(method: str, params: object = None, *, config: object) -> object:
_ = (params, config)
calls.append(method)
if method == "config.schema":
return {"version": "2026.2.13"}
raise AssertionError(f"unexpected method: {method}")
async def _fake_connect_metadata(*, config: GatewayConfig) -> object | None:
_ = config
return None
return {
"version": "dev",
"runtime": {"version": "2026.1.0"},
"server": {"version": "2026.2.13"},
}
monkeypatch.setattr(gateway_compat, "openclaw_connect_metadata", _fake_connect_metadata)
monkeypatch.setattr(gateway_compat, "openclaw_call", _fake_openclaw_call)
result = await gateway_compat.check_gateway_runtime_compatibility(
result = await gateway_compat.check_gateway_version_compatibility(
GatewayConfig(url="ws://gateway.example/ws"),
minimum_version="2026.1.30",
)
assert calls == ["config.schema"]
assert result.compatible is True
assert result.current_version == "2026.2.13"
@pytest.mark.asyncio
async def test_check_gateway_runtime_compatibility_prefers_connect_metadata(
async def test_check_gateway_version_compatibility_fails_without_server_version(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
async def _fake_connect_metadata(*, config: GatewayConfig) -> object | None:
_ = config
return {"server": {"version": "2026.2.21-2"}}
async def _fake_openclaw_call(method: str, params: object = None, *, config: object) -> object:
_ = (params, config)
calls.append(method)
raise AssertionError(f"unexpected method: {method}")
return {"runtime": {"version": "2026.2.13"}}
monkeypatch.setattr(gateway_compat, "openclaw_connect_metadata", _fake_connect_metadata)
monkeypatch.setattr(gateway_compat, "openclaw_call", _fake_openclaw_call)
result = await gateway_compat.check_gateway_runtime_compatibility(
result = await gateway_compat.check_gateway_version_compatibility(
GatewayConfig(url="ws://gateway.example/ws"),
minimum_version="2026.1.30",
)
assert calls == []
assert result.compatible is True
assert result.current_version == "2026.2.21-2"
assert result.compatible is False
assert result.current_version is None
assert "unable to determine gateway version" in (result.message or "").lower()
@pytest.mark.asyncio
async def test_check_gateway_runtime_compatibility_falls_back_to_health(
async def test_check_gateway_version_compatibility_rejects_non_calver_server_version(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
async def _fake_openclaw_call(method: str, params: object = None, *, config: object) -> object:
_ = (params, config)
calls.append(method)
if method == "config.schema":
raise OpenClawGatewayError("unknown method")
if method == "status":
raise OpenClawGatewayError("unknown method")
return {"version": "2026.2.0"}
async def _fake_connect_metadata(*, config: GatewayConfig) -> object | None:
_ = config
return None
return {"server": {"version": "dev"}}
monkeypatch.setattr(gateway_compat, "openclaw_connect_metadata", _fake_connect_metadata)
monkeypatch.setattr(gateway_compat, "openclaw_call", _fake_openclaw_call)
result = await gateway_compat.check_gateway_runtime_compatibility(
result = await gateway_compat.check_gateway_version_compatibility(
GatewayConfig(url="ws://gateway.example/ws"),
minimum_version="2026.1.30",
)
assert calls == ["config.schema", "status", "health"]
assert result.compatible is True
assert result.current_version == "2026.2.0"
assert result.compatible is False
assert result.current_version == "dev"
assert "unsupported version format" in (result.message or "").lower()
@pytest.mark.asyncio
async def test_check_gateway_runtime_compatibility_uses_health_when_status_has_no_version(
async def test_check_gateway_version_compatibility_propagates_connect_errors(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
async def _fake_openclaw_call(method: str, params: object = None, *, config: object) -> object:
_ = (params, config)
calls.append(method)
if method == "config.schema":
return {"schema": {"title": "Gateway schema"}}
if method == "status":
return {"uptime": 1234}
return {"version": "2026.2.0"}
async def _fake_connect_metadata(*, config: GatewayConfig) -> object | None:
_ = config
return None
raise OpenClawGatewayError("connection refused")
monkeypatch.setattr(gateway_compat, "openclaw_connect_metadata", _fake_connect_metadata)
monkeypatch.setattr(gateway_compat, "openclaw_call", _fake_openclaw_call)
result = await gateway_compat.check_gateway_runtime_compatibility(
GatewayConfig(url="ws://gateway.example/ws"),
minimum_version="2026.1.30",
)
assert calls == ["config.schema", "status", "health"]
assert result.compatible is True
assert result.current_version == "2026.2.0"
with pytest.raises(OpenClawGatewayError, match="connection refused"):
await gateway_compat.check_gateway_version_compatibility(
GatewayConfig(url="ws://gateway.example/ws"),
minimum_version="2026.1.30",
)
@pytest.mark.asyncio
@@ -172,7 +178,7 @@ async def test_admin_service_rejects_incompatible_gateway(
message="Gateway version 2026.1.0 is not supported.",
)
monkeypatch.setattr(admin_service, "check_gateway_runtime_compatibility", _fake_check)
monkeypatch.setattr(admin_service, "check_gateway_version_compatibility", _fake_check)
service = GatewayAdminLifecycleService(session=object()) # type: ignore[arg-type]
with pytest.raises(HTTPException) as exc_info:
@@ -190,7 +196,7 @@ async def test_admin_service_maps_gateway_transport_errors(
_ = (config, minimum_version)
raise OpenClawGatewayError("connection refused")
monkeypatch.setattr(admin_service, "check_gateway_runtime_compatibility", _fake_check)
monkeypatch.setattr(admin_service, "check_gateway_version_compatibility", _fake_check)
service = GatewayAdminLifecycleService(session=object()) # type: ignore[arg-type]
with pytest.raises(HTTPException) as exc_info:
@@ -208,7 +214,7 @@ async def test_admin_service_maps_gateway_scope_errors_with_guidance(
_ = (config, minimum_version)
raise OpenClawGatewayError("missing scope: operator.read")
monkeypatch.setattr(admin_service, "check_gateway_runtime_compatibility", _fake_check)
monkeypatch.setattr(admin_service, "check_gateway_version_compatibility", _fake_check)
service = GatewayAdminLifecycleService(session=object()) # type: ignore[arg-type]
with pytest.raises(HTTPException) as exc_info:
@@ -231,7 +237,7 @@ async def test_gateway_status_reports_incompatible_version(
message="Gateway version 2026.1.0 is not supported.",
)
monkeypatch.setattr(session_service, "check_gateway_runtime_compatibility", _fake_check)
monkeypatch.setattr(session_service, "check_gateway_version_compatibility", _fake_check)
service = GatewaySessionService(session=object()) # type: ignore[arg-type]
response = await service.get_status(
@@ -252,7 +258,7 @@ async def test_gateway_status_surfaces_scope_error_guidance(
_ = (config, minimum_version)
raise OpenClawGatewayError("missing scope: operator.read")
monkeypatch.setattr(session_service, "check_gateway_runtime_compatibility", _fake_check)
monkeypatch.setattr(session_service, "check_gateway_version_compatibility", _fake_check)
service = GatewaySessionService(session=object()) # type: ignore[arg-type]
response = await service.get_status(
@@ -284,7 +290,7 @@ async def test_gateway_status_returns_sessions_when_version_compatible(
assert method == "sessions.list"
return {"sessions": [{"key": "agent:main"}]}
monkeypatch.setattr(session_service, "check_gateway_runtime_compatibility", _fake_check)
monkeypatch.setattr(session_service, "check_gateway_version_compatibility", _fake_check)
monkeypatch.setattr(session_service, "openclaw_call", _fake_openclaw_call)
service = GatewaySessionService(session=object()) # type: ignore[arg-type]