mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-29 19:01:41 +00:00
292 lines
10 KiB
Python
292 lines
10 KiB
Python
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.run_journal import InvalidRunTransitionError, SQLiteRunJournal
|
|
from app.run_sync.cloud_sync import (
|
|
CloudSyncConfiguration,
|
|
RunEventSyncHttpError,
|
|
)
|
|
from app.run_sync.command_sync import (
|
|
CommandControlWorker,
|
|
HttpCommandSyncTransport,
|
|
)
|
|
|
|
|
|
class FakeCommandTransport:
|
|
def __init__(self) -> None:
|
|
self.pending: list[dict[str, Any]] = []
|
|
self.confirmed: list[str] = []
|
|
self.ingested: list[Any] = []
|
|
self.pull_error: Exception | None = None
|
|
self.pull_count = 0
|
|
|
|
async def pull_pending(self, _configuration, *, limit):
|
|
self.pull_count += 1
|
|
if self.pull_error is not None:
|
|
raise self.pull_error
|
|
return self.pending[:limit]
|
|
|
|
async def confirm_receipt(self, _configuration, command):
|
|
self.confirmed.append(command.command_id)
|
|
return {
|
|
"result": "confirmed",
|
|
"receipt_state": "durably_received",
|
|
"may_execute": True,
|
|
}
|
|
|
|
async def ingest_events(self, _configuration, batch):
|
|
self.ingested.append(batch)
|
|
return {"expected_next_desktop_event_sequence": len(batch.events) + 1}
|
|
|
|
async def close(self):
|
|
return None
|
|
|
|
|
|
def _configuration() -> CloudSyncConfiguration:
|
|
return CloudSyncConfiguration(
|
|
endpoint_url="https://example.test/api/v1/sync/events:ingest",
|
|
authorization="Bearer token",
|
|
desktop_instance_id="device-1",
|
|
)
|
|
|
|
|
|
def _command() -> dict[str, Any]:
|
|
return {
|
|
"id": "command-1",
|
|
"session_id": "session-1",
|
|
"user_id": 7,
|
|
"project_id": "project-1",
|
|
"route_version": 1,
|
|
"type": "user_message",
|
|
"payload": {"content": "hello"},
|
|
"expires_at": "2030-01-01T00:00:00+00:00",
|
|
"receipt_grace_until": "2030-01-01T00:00:30+00:00",
|
|
"requires_online_receipt_confirmation": False,
|
|
"lease_token": "delivery-lease-1",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_confirms_receipt_and_drains_independent_lane(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
transport = FakeCommandTransport()
|
|
worker = CommandControlWorker(journal, transport)
|
|
worker.configure(_configuration())
|
|
record = await worker.persist_command(_command())
|
|
|
|
confirmed, may_execute = await worker.confirm_receipt(
|
|
record.command_id
|
|
)
|
|
assert may_execute is True
|
|
assert confirmed.receipt_status == "confirmed"
|
|
assert confirmed.delivery_lease_token == "delivery-lease-1"
|
|
assert transport.confirmed == ["command-1"]
|
|
|
|
assert await worker.drain_once() == 1
|
|
assert len(transport.ingested) == 1
|
|
assert transport.ingested[0].events[0].event_type == (
|
|
"receipt.durably_received"
|
|
)
|
|
assert transport.ingested[0].delivery_lease_token == (
|
|
"delivery-lease-1"
|
|
)
|
|
assert journal.claim_command_result_batches() == []
|
|
await worker.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_high_risk_command_does_not_execute_without_cloud_config(
|
|
tmp_path,
|
|
):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
transport = FakeCommandTransport()
|
|
worker = CommandControlWorker(journal, transport)
|
|
command = _command()
|
|
command["requires_online_receipt_confirmation"] = True
|
|
record = await worker.persist_command(command)
|
|
|
|
_record, may_execute = await worker.confirm_receipt(record.command_id)
|
|
assert may_execute is False
|
|
await worker.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminal_command_receipt_replay_never_executes_again(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
transport = FakeCommandTransport()
|
|
worker = CommandControlWorker(journal, transport)
|
|
worker.configure(_configuration())
|
|
record = await worker.persist_command(_command())
|
|
await worker.confirm_receipt(record.command_id)
|
|
journal.append_command_result(
|
|
record.command_id,
|
|
event_type="admission.accepted",
|
|
event_id="accepted",
|
|
)
|
|
journal.append_command_result(
|
|
record.command_id,
|
|
event_type="execution.completed",
|
|
event_id="completed",
|
|
payload={"result": {"ok": True}},
|
|
)
|
|
|
|
replayed, may_execute = await worker.confirm_receipt(record.command_id)
|
|
|
|
assert replayed.state == "completed"
|
|
assert may_execute is False
|
|
assert transport.confirmed == [record.command_id]
|
|
await worker.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inbound_pull_failure_does_not_starve_outbound_results(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
transport = FakeCommandTransport()
|
|
worker = CommandControlWorker(journal, transport)
|
|
worker.configure(_configuration())
|
|
await worker.persist_command(_command())
|
|
transport.pull_error = ValueError("malformed pending command")
|
|
|
|
assert await worker.drain_once() == 1
|
|
assert len(transport.ingested) == 1
|
|
assert journal.claim_command_result_batches() == []
|
|
await worker.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_device_owner_mismatch_backs_off_inbound_without_hot_loop(
|
|
tmp_path,
|
|
):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
transport = FakeCommandTransport()
|
|
transport.pull_error = RunEventSyncHttpError(
|
|
409,
|
|
{
|
|
"detail": {
|
|
"code": "device_owner_mismatch",
|
|
"message": "Desktop device belongs to another user",
|
|
}
|
|
},
|
|
)
|
|
worker = CommandControlWorker(
|
|
journal,
|
|
transport,
|
|
poll_interval_seconds=0.01,
|
|
max_retry_seconds=300,
|
|
)
|
|
worker.configure(_configuration())
|
|
|
|
assert await worker.drain_once() == 0
|
|
assert await worker.drain_once() == 0
|
|
assert transport.pull_count == 1
|
|
assert worker._next_inbound_attempt_at > 0
|
|
await worker.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_device_registration_error_retries_command_lane(tmp_path):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.url.path.endswith("/devices/register")
|
|
return httpx.Response(409, json={"detail": "device route conflict"})
|
|
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
transport = HttpCommandSyncTransport(
|
|
transport=httpx.MockTransport(handler)
|
|
)
|
|
worker = CommandControlWorker(journal, transport)
|
|
worker.configure(_configuration())
|
|
await worker.persist_command(_command())
|
|
|
|
assert await worker.drain_once() == 0
|
|
batch = journal.claim_command_result_batches(now=float("inf"))[0]
|
|
assert batch.attempt_count == 1
|
|
await worker.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transport_reregisters_when_authenticated_credential_changes():
|
|
registrations: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/devices/register"):
|
|
registrations.append(request.headers["authorization"])
|
|
return httpx.Response(200, json={})
|
|
assert request.url.path.endswith("/commands/pending")
|
|
return httpx.Response(200, json={"items": []})
|
|
|
|
transport = HttpCommandSyncTransport(
|
|
transport=httpx.MockTransport(handler)
|
|
)
|
|
first = _configuration()
|
|
second = CloudSyncConfiguration(
|
|
endpoint_url=first.endpoint_url,
|
|
authorization="Bearer another-account-token",
|
|
desktop_instance_id=first.desktop_instance_id,
|
|
)
|
|
|
|
assert await transport.pull_pending(first, limit=1) == []
|
|
assert await transport.pull_pending(first, limit=1) == []
|
|
assert await transport.pull_pending(second, limit=1) == []
|
|
|
|
assert registrations == ["Bearer token", "Bearer another-account-token"]
|
|
await transport.close()
|
|
|
|
|
|
def test_command_inbox_terminal_state_cannot_move_backwards(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
record = journal.persist_remote_command(
|
|
command_id="command-1",
|
|
session_id="session-1",
|
|
user_id=7,
|
|
project_id="project-1",
|
|
run_id=None,
|
|
route_version=1,
|
|
command_type="user_message",
|
|
payload={"content": "hello"},
|
|
expires_at=100,
|
|
receipt_grace_until=110,
|
|
requires_online_receipt_confirmation=False,
|
|
now=1,
|
|
)
|
|
journal.append_command_result(
|
|
record.command_id,
|
|
event_type="admission.accepted",
|
|
event_id="accepted",
|
|
occurred_at=2,
|
|
)
|
|
journal.append_command_result(
|
|
record.command_id,
|
|
event_type="execution.completed",
|
|
event_id="completed",
|
|
occurred_at=3,
|
|
)
|
|
|
|
with pytest.raises(InvalidRunTransitionError, match="completed"):
|
|
journal.append_command_result(
|
|
record.command_id,
|
|
event_type="admission.rejected",
|
|
event_id="late-rejected",
|
|
occurred_at=4,
|
|
)
|
|
|
|
assert (
|
|
journal.get_remote_command(record.command_id).state == "completed"
|
|
)
|