129e4ea1fc
New router intelligence modules (26 files): alert_ingest/store, audit_store, architecture_pressure, backlog_generator/store, cost_analyzer, data_governance, dependency_scanner, drift_analyzer, incident_* (5 files), llm_enrichment, platform_priority_digest, provider_budget, release_check_runner, risk_* (6 files), signature_state_store, sofiia_auto_router, tool_governance New services: - sofiia-console: Dockerfile, adapters/, monitor/nodes/ops/voice modules, launchd, react static - memory-service: integration_endpoints, integrations, voice_endpoints, static UI - aurora-service: full app suite (analysis, job_store, orchestrator, reporting, schemas, subagents) - sofiia-supervisor: new supervisor service - aistalk-bridge-lite: Telegram bridge lite - calendar-service: CalDAV calendar service with reminders - mlx-stt-service / mlx-tts-service: Apple Silicon speech services - binance-bot-monitor: market monitor service - node-worker: STT/TTS memory providers New tools (9): agent_email, browser_tool, contract_tool, observability_tool, oncall_tool, pr_reviewer_tool, repo_tool, safe_code_executor, secure_vault New crews: agromatrix_crew (10 modules: depth_classifier, doc_facts, doc_focus, farm_state, light_reply, llm_factory, memory_manager, proactivity, reflection_engine, session_context, style_adapter, telemetry) Tests: 85+ test files for all new modules Made-with: Cursor
245 lines
11 KiB
Python
245 lines
11 KiB
Python
"""
|
||
tests/test_stepan_v43_farmos.py
|
||
|
||
Unit tests for v4.3 FarmOS minimal tool pipeline.
|
||
|
||
Scope:
|
||
- agromatrix_tools.tool_farmos_read._farmos_ping_impl (fail-closed logic)
|
||
- crews.agromatrix_crew.operator_commands.handle_farmos_status
|
||
- crews.agromatrix_crew.operator_commands.parse_operator_command (farmos registerd)
|
||
|
||
Без залежностей від crewai, httpx, memory-service.
|
||
Тільки monkeypatch env + mock requests.
|
||
"""
|
||
from __future__ import annotations
|
||
import os
|
||
import sys
|
||
import types
|
||
import pytest
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
# ── Шляхи ────────────────────────────────────────────────────────────────────
|
||
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
_PKG = os.path.join(_ROOT, "packages", "agromatrix-tools")
|
||
_CREWS = os.path.join(_ROOT, "crews")
|
||
for _p in (_PKG, _CREWS):
|
||
if _p not in sys.path:
|
||
sys.path.insert(0, _p)
|
||
|
||
# ── Мок недоступних модулів до будь-якого імпорту agromatrix_tools ────────────
|
||
# nats, crewai — потрібні тільки на сервері, тут мокуємо.
|
||
def _stub_module(name: str) -> types.ModuleType:
|
||
m = types.ModuleType(name)
|
||
sys.modules[name] = m
|
||
return m
|
||
|
||
for _mod in (
|
||
"nats", "nats.aio", "nats.aio.client",
|
||
"crewai", "crewai.tools",
|
||
):
|
||
if _mod not in sys.modules:
|
||
_stub_module(_mod)
|
||
|
||
# crewai.tools.tool — повертає identity decorator
|
||
sys.modules["crewai.tools"].tool = lambda name: (lambda f: f) # type: ignore[attr-defined]
|
||
|
||
# nats.aio.client.Client stub
|
||
sys.modules["nats.aio.client"].Client = MagicMock # type: ignore[attr-defined]
|
||
|
||
# audit_tool_call stub (використовує nats)
|
||
import importlib, agromatrix_tools # noqa: E402
|
||
# Форсуємо stub для audit всередині пакету
|
||
_audit_stub = types.ModuleType("agromatrix_tools.audit")
|
||
_audit_stub.audit_tool_call = lambda *a, **kw: None # type: ignore[attr-defined]
|
||
sys.modules["agromatrix_tools.audit"] = _audit_stub
|
||
|
||
# Тепер безпечно імпортуємо
|
||
import importlib as _il # noqa: F811
|
||
if "agromatrix_tools.tool_farmos_read" in sys.modules:
|
||
_il.reload(sys.modules["agromatrix_tools.tool_farmos_read"])
|
||
|
||
from agromatrix_tools.tool_farmos_read import _farmos_ping_impl # noqa: E402
|
||
from crews.agromatrix_crew.operator_commands import ( # noqa: E402
|
||
handle_farmos_status,
|
||
parse_operator_command,
|
||
OPERATOR_COMMANDS,
|
||
)
|
||
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def clear_farmos_env(monkeypatch):
|
||
"""Прибираємо farmos env перед кожним тестом."""
|
||
for k in ("FARMOS_BASE_URL", "FARMOS_TOKEN", "FARMOS_USER", "FARMOS_PASS", "FARMOS_PASSWORD"):
|
||
monkeypatch.delenv(k, raising=False)
|
||
yield
|
||
|
||
|
||
# ─── _farmos_ping_impl ────────────────────────────────────────────────────────
|
||
|
||
class TestFarmosPingImpl:
|
||
def test_no_base_url(self):
|
||
result = _farmos_ping_impl()
|
||
assert "FARMOS_BASE_URL" in result
|
||
assert "відсутній" in result.lower() or "не налаштований" in result.lower()
|
||
|
||
def test_base_url_no_auth(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
result = _farmos_ping_impl()
|
||
assert "токен" in result.lower() or "логін" in result.lower() or "auth" in result.lower()
|
||
|
||
def test_base_url_with_token_ok(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "testtoken123")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = _farmos_ping_impl()
|
||
assert "доступний" in result.lower()
|
||
assert "FarmOS доступний." == result
|
||
|
||
def test_base_url_with_token_401(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "badtoken")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 401
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = _farmos_ping_impl()
|
||
assert "401" in result
|
||
assert "авторизац" in result.lower()
|
||
|
||
def test_base_url_with_token_403(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "limitedtoken")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 403
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = _farmos_ping_impl()
|
||
assert "403" in result
|
||
assert "авторизац" in result.lower()
|
||
|
||
def test_base_url_with_token_503(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "tok")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 503
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = _farmos_ping_impl()
|
||
assert "503" in result
|
||
|
||
def test_timeout(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "tok")
|
||
import requests as _r
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get",
|
||
side_effect=_r.exceptions.Timeout()):
|
||
result = _farmos_ping_impl()
|
||
assert "timeout" in result.lower()
|
||
|
||
def test_connection_error(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "tok")
|
||
import requests as _r
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get",
|
||
side_effect=_r.exceptions.ConnectionError("refused")):
|
||
result = _farmos_ping_impl()
|
||
assert "недоступний" in result.lower()
|
||
|
||
def test_user_pass_auth_ok(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_USER", "admin")
|
||
monkeypatch.setenv("FARMOS_PASS", "secret")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = _farmos_ping_impl()
|
||
assert "доступний" in result.lower()
|
||
|
||
def test_never_reveals_url(self, monkeypatch):
|
||
"""Відповідь не містить BASE_URL або токен."""
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://secret-farmos.internal")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "supersecrettoken")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = _farmos_ping_impl()
|
||
assert "secret-farmos" not in result
|
||
assert "supersecrettoken" not in result
|
||
|
||
def test_never_raises(self, monkeypatch):
|
||
"""При будь-якій помилці — повертає рядок, не кидає."""
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "tok")
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get",
|
||
side_effect=RuntimeError("unexpected")):
|
||
result = _farmos_ping_impl()
|
||
assert isinstance(result, str)
|
||
assert len(result) > 0
|
||
|
||
|
||
# ─── handle_farmos_status ─────────────────────────────────────────────────────
|
||
|
||
class TestHandleFarmosStatus:
|
||
def test_no_args_returns_status(self):
|
||
result = handle_farmos_status([])
|
||
assert isinstance(result, dict)
|
||
assert "summary" in result
|
||
assert "FARMOS_BASE_URL" in result["summary"] or "FarmOS" in result["summary"]
|
||
|
||
def test_status_arg_same_as_no_args(self):
|
||
r1 = handle_farmos_status([])
|
||
r2 = handle_farmos_status(["status"])
|
||
assert r1["summary"] == r2["summary"]
|
||
|
||
def test_unknown_subcommand(self):
|
||
result = handle_farmos_status(["foo"])
|
||
assert "підтримується" in result["summary"].lower() or "farmos" in result["summary"].lower()
|
||
|
||
def test_returns_dict_with_required_keys(self):
|
||
result = handle_farmos_status([])
|
||
for key in ("status", "summary", "artifacts", "tool_calls", "next_actions"):
|
||
assert key in result
|
||
|
||
def test_with_ok_farmos(self, monkeypatch):
|
||
monkeypatch.setenv("FARMOS_BASE_URL", "http://farmos.example.com")
|
||
monkeypatch.setenv("FARMOS_TOKEN", "tok")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
with patch("agromatrix_tools.tool_farmos_read.requests.get", return_value=mock_resp):
|
||
result = handle_farmos_status([])
|
||
assert "доступний" in result["summary"].lower()
|
||
|
||
def test_internal_error_fallback(self, monkeypatch):
|
||
"""handle_farmos_status не падає при внутрішній помилці importside."""
|
||
# Патчимо саме там де lazy import шукає
|
||
with patch("agromatrix_tools.tool_farmos_read._farmos_ping_impl",
|
||
side_effect=RuntimeError("crash")):
|
||
result = handle_farmos_status([])
|
||
assert "помилка" in result["summary"].lower() or "недоступний" in result["summary"].lower()
|
||
|
||
|
||
# ─── OPERATOR_COMMANDS registry ───────────────────────────────────────────────
|
||
|
||
class TestFarmosCommandRegistry:
|
||
def test_farmos_in_operator_commands(self):
|
||
assert "farmos" in OPERATOR_COMMANDS
|
||
|
||
def test_parse_farmos_no_args(self):
|
||
parsed = parse_operator_command("/farmos")
|
||
assert parsed is not None
|
||
assert parsed["cmd"] == "farmos"
|
||
assert parsed["args"] == []
|
||
|
||
def test_parse_farmos_status(self):
|
||
parsed = parse_operator_command("/farmos status")
|
||
assert parsed is not None
|
||
assert parsed["cmd"] == "farmos"
|
||
assert "status" in parsed["args"]
|
||
|
||
def test_parse_farmos_foo(self):
|
||
parsed = parse_operator_command("/farmos foo")
|
||
assert parsed is not None
|
||
assert parsed["cmd"] == "farmos"
|
||
assert "foo" in parsed["args"]
|