117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
|
|
def test_run_with_timeout_supports_sync_function():
|
|
from services.elephant_alpha_autonomous_engine import ElephantAlphaAutonomousEngine
|
|
|
|
result = asyncio.run(ElephantAlphaAutonomousEngine._run_with_timeout(lambda value: value + 1, 41))
|
|
|
|
assert result == 42
|
|
|
|
|
|
def test_execute_step_rejects_unknown_action():
|
|
from services.elephant_alpha_autonomous_engine import ElephantAlphaAutonomousEngine
|
|
|
|
engine = ElephantAlphaAutonomousEngine()
|
|
|
|
try:
|
|
asyncio.run(engine._execute_step({"agent": "mystery", "action": "do_anything"}))
|
|
except ValueError as exc:
|
|
assert "Unrecognized step" in str(exc)
|
|
else:
|
|
raise AssertionError("unknown action should fail")
|
|
|
|
|
|
def test_execute_step_routes_code_fix_to_autoheal(monkeypatch):
|
|
from services.elephant_alpha_autonomous_engine import ElephantAlphaAutonomousEngine
|
|
|
|
calls = []
|
|
engine = ElephantAlphaAutonomousEngine()
|
|
monkeypatch.setattr(
|
|
engine,
|
|
"_run_auto_heal",
|
|
lambda error_type, context: calls.append((error_type, context)) or {"ok": True},
|
|
)
|
|
|
|
asyncio.run(engine._execute_step({
|
|
"agent": "elephant_alpha",
|
|
"action": "code_fix",
|
|
"parameters": {"target_file": "services/example.py", "error_message": "Traceback"},
|
|
}))
|
|
|
|
assert calls == [("python_exception", {"target_file": "services/example.py", "error_message": "Traceback"})]
|
|
|
|
|
|
def test_execute_step_routes_price_adjustment_to_human_review(monkeypatch):
|
|
from services.elephant_alpha_autonomous_engine import ElephantAlphaAutonomousEngine
|
|
|
|
calls = []
|
|
engine = ElephantAlphaAutonomousEngine()
|
|
monkeypatch.setattr(
|
|
engine,
|
|
"_record_price_adjustment_review",
|
|
lambda step: calls.append(step) or {"status": "pending_review", "sku": "SKU-9"},
|
|
)
|
|
|
|
result = asyncio.run(engine._execute_step({
|
|
"agent": "nemotron",
|
|
"action": "execute_price_adjustment",
|
|
"parameters": {"sku": "SKU-9", "recommended_price": 1280},
|
|
}))
|
|
|
|
assert result == {"status": "pending_review", "sku": "SKU-9"}
|
|
assert calls[0]["parameters"]["recommended_price"] == 1280
|
|
|
|
|
|
def test_execute_step_skips_legacy_openclaw_resource_strategy():
|
|
from services.elephant_alpha_autonomous_engine import ElephantAlphaAutonomousEngine
|
|
|
|
engine = ElephantAlphaAutonomousEngine()
|
|
|
|
result = asyncio.run(engine._execute_step({
|
|
"agent": "openclaw",
|
|
"action": "generate_resource_optimization_strategy",
|
|
}))
|
|
|
|
assert result is None
|
|
|
|
|
|
def test_autoheal_derives_python_exception_from_traceback():
|
|
from services.auto_heal_service import AutoHealService
|
|
|
|
svc = AutoHealService()
|
|
|
|
assert svc._derive_error_type({"traceback_str": "Traceback (most recent call last):\nNameError"}) == "python_exception"
|
|
|
|
|
|
def test_execute_autonomous_decision_logs_short_circuit_telemetry_failure(monkeypatch, caplog):
|
|
from services.elephant_alpha_autonomous_engine import (
|
|
AutonomousTrigger,
|
|
ElephantAlphaAutonomousEngine,
|
|
)
|
|
import services.ai_call_logger as ai_call_logger
|
|
|
|
engine = ElephantAlphaAutonomousEngine()
|
|
|
|
async def _no_hermes_threats(top_n=5):
|
|
return None
|
|
|
|
def _broken_log_ai_call(*args, **kwargs):
|
|
raise RuntimeError("ai telemetry unavailable")
|
|
|
|
monkeypatch.setattr(engine, "_fetch_hermes_threats_summary", _no_hermes_threats)
|
|
monkeypatch.setattr(ai_call_logger, "log_ai_call", _broken_log_ai_call)
|
|
caplog.set_level(logging.WARNING, logger="services.elephant_alpha_autonomous_engine")
|
|
|
|
trigger = AutonomousTrigger(
|
|
trigger_type="price_drop_alert",
|
|
conditions={},
|
|
threshold=0.8,
|
|
enabled=True,
|
|
)
|
|
|
|
asyncio.run(engine._execute_autonomous_decision(trigger))
|
|
|
|
assert "EA short-circuit telemetry failed" in caplog.text
|