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
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""
|
||
tests/test_stepan_v4_farm_state.py
|
||
|
||
v4 Farm State Layer:
|
||
- detect_farm_state_updates (crop, stage, issue, risk)
|
||
- update_farm_state (session merge, dedup, TTL)
|
||
- build_farm_state_prefix (format, empty cases, TTL)
|
||
- Isolation: prefix NOT added in doc/web mode
|
||
- State persists within session
|
||
"""
|
||
import time
|
||
import pytest
|
||
|
||
from crews.agromatrix_crew.farm_state import (
|
||
detect_farm_state_updates,
|
||
update_farm_state,
|
||
build_farm_state_prefix,
|
||
FARM_STATE_TTL,
|
||
)
|
||
|
||
|
||
# ─── 1. detect_farm_state_updates ────────────────────────────────────────────
|
||
|
||
class TestDetectFarmStateUpdates:
|
||
|
||
def test_crop_kukurudza(self):
|
||
u = detect_farm_state_updates("По кукурудзі що робити далі?")
|
||
assert u.get("current_crop") == "кукурудза"
|
||
|
||
def test_crop_inflection_kukurudzu(self):
|
||
u = detect_farm_state_updates("Переглянь стан кукурудзу на полі")
|
||
assert u.get("current_crop") == "кукурудза"
|
||
|
||
def test_crop_wheat(self):
|
||
u = detect_farm_state_updates("Пшениця виглядає кволою")
|
||
assert u.get("current_crop") == "пшениця"
|
||
|
||
def test_crop_sunflower(self):
|
||
u = detect_farm_state_updates("соняшник V6 посуха")
|
||
assert u.get("current_crop") == "соняшник"
|
||
|
||
def test_crop_rapeseed(self):
|
||
u = detect_farm_state_updates("ріпак — йде цвітіння")
|
||
assert u.get("current_crop") == "ріпак"
|
||
|
||
def test_stage_v6(self):
|
||
u = detect_farm_state_updates("Стадія V6, є жовтизна")
|
||
assert u.get("growth_stage") == "V6"
|
||
|
||
def test_stage_kushennya(self):
|
||
u = detect_farm_state_updates("пшениця, фаза кущення")
|
||
assert "кущення" in u.get("growth_stage", "").lower()
|
||
|
||
def test_stage_flowering(self):
|
||
u = detect_farm_state_updates("ріпак іде на цвітіння")
|
||
assert "цвітіння" in u.get("growth_stage", "").lower()
|
||
|
||
def test_issue_zhovtyzna(self):
|
||
u = detect_farm_state_updates("є жовтизна на листі")
|
||
assert "жовтизна" in u.get("recent_issue", "").lower()
|
||
|
||
def test_issue_deficit(self):
|
||
u = detect_farm_state_updates("дефіцит азоту у посівах")
|
||
assert "дефіцит" in u.get("recent_issue", "").lower()
|
||
|
||
def test_issue_shkidnyk(self):
|
||
u = detect_farm_state_updates("з'явились шкідники на кукурудзі")
|
||
assert "шкідник" in u.get("recent_issue", "").lower()
|
||
|
||
def test_risk_posuxa(self):
|
||
u = detect_farm_state_updates("на полі посуха вже 2 тижні")
|
||
assert "посуха" in u.get("risk_flags", [])
|
||
|
||
def test_risk_zamorozok(self):
|
||
u = detect_farm_state_updates("очікується заморозок вночі")
|
||
assert any("заморозок" in r for r in u.get("risk_flags", []))
|
||
|
||
def test_multiple_fields(self):
|
||
u = detect_farm_state_updates("кукурудза V6, дефіцит азоту, посуха")
|
||
assert u.get("current_crop") == "кукурудза"
|
||
assert "V6" in u.get("growth_stage", "")
|
||
assert "дефіцит" in u.get("recent_issue", "").lower()
|
||
assert any("посуха" in r for r in u.get("risk_flags", []))
|
||
|
||
def test_empty_text(self):
|
||
u = detect_farm_state_updates("")
|
||
assert u == {}
|
||
|
||
def test_no_match(self):
|
||
u = detect_farm_state_updates("привіт, як справи сьогодні?")
|
||
assert u == {}
|
||
|
||
def test_fail_safe_garbage(self):
|
||
u = detect_farm_state_updates(None) # type: ignore
|
||
assert u == {}
|
||
|
||
|
||
# ─── 2. update_farm_state ─────────────────────────────────────────────────────
|
||
|
||
class TestUpdateFarmState:
|
||
|
||
def test_creates_farm_state(self):
|
||
session = {}
|
||
update_farm_state(session, {"current_crop": "кукурудза"})
|
||
assert session["farm_state"]["current_crop"] == "кукурудза"
|
||
|
||
def test_sets_last_update_ts(self):
|
||
now = time.time()
|
||
session = {}
|
||
update_farm_state(session, {"current_crop": "соняшник"}, now_ts=now)
|
||
assert abs(session["farm_state"]["last_update_ts"] - now) < 1
|
||
|
||
def test_updates_existing(self):
|
||
session = {"farm_state": {"current_crop": "пшениця", "last_update_ts": 0.0}}
|
||
update_farm_state(session, {"current_crop": "кукурудза"})
|
||
assert session["farm_state"]["current_crop"] == "кукурудза"
|
||
|
||
def test_merges_risk_flags(self):
|
||
session = {"farm_state": {"risk_flags": ["посуха"], "last_update_ts": 0.0}}
|
||
update_farm_state(session, {"risk_flags": ["заморозок"]})
|
||
flags = session["farm_state"]["risk_flags"]
|
||
assert "посуха" in flags
|
||
assert "заморозок" in flags
|
||
|
||
def test_dedup_risk_flags(self):
|
||
session = {"farm_state": {"risk_flags": ["посуха", "спека"], "last_update_ts": 0.0}}
|
||
update_farm_state(session, {"risk_flags": ["посуха"]})
|
||
flags = session["farm_state"]["risk_flags"]
|
||
assert flags.count("посуха") == 1
|
||
|
||
def test_risk_flags_max_5(self):
|
||
session = {"farm_state": {"risk_flags": ["r1", "r2", "r3", "r4"], "last_update_ts": 0.0}}
|
||
update_farm_state(session, {"risk_flags": ["r5", "r6"]})
|
||
assert len(session["farm_state"]["risk_flags"]) <= 5
|
||
|
||
def test_empty_updates_noop(self):
|
||
session = {"farm_state": {"current_crop": "ріпак", "last_update_ts": 0.0}}
|
||
update_farm_state(session, {})
|
||
assert session["farm_state"]["current_crop"] == "ріпак"
|
||
|
||
def test_fail_safe_exception(self):
|
||
# session = None → fail-safe, no exception
|
||
update_farm_state(None, {"current_crop": "соя"}) # type: ignore
|
||
|
||
|
||
# ─── 3. build_farm_state_prefix ──────────────────────────────────────────────
|
||
|
||
class TestBuildFarmStatePrefix:
|
||
|
||
def _session_with_state(self, now: float, **kwargs) -> dict:
|
||
fs = {"last_update_ts": now}
|
||
fs.update(kwargs)
|
||
return {"farm_state": fs}
|
||
|
||
def test_empty_if_no_crop(self):
|
||
now = time.time()
|
||
s = self._session_with_state(now, growth_stage="V6")
|
||
assert build_farm_state_prefix(s) == ""
|
||
|
||
def test_basic_prefix(self):
|
||
now = time.time()
|
||
s = self._session_with_state(now, current_crop="кукурудза")
|
||
p = build_farm_state_prefix(s)
|
||
assert "кукурудза" in p
|
||
assert "[Контекст господарства]" in p
|
||
|
||
def test_includes_stage(self):
|
||
now = time.time()
|
||
s = self._session_with_state(now, current_crop="кукурудза", growth_stage="V6")
|
||
p = build_farm_state_prefix(s)
|
||
assert "V6" in p
|
||
assert "Стадія" in p
|
||
|
||
def test_includes_issue(self):
|
||
now = time.time()
|
||
s = self._session_with_state(now, current_crop="кукурудза", recent_issue="жовтизна")
|
||
p = build_farm_state_prefix(s)
|
||
assert "жовтизна" in p
|
||
assert "Проблема" in p
|
||
|
||
def test_includes_risks(self):
|
||
now = time.time()
|
||
s = self._session_with_state(now, current_crop="соняшник", risk_flags=["посуха", "спека"])
|
||
p = build_farm_state_prefix(s)
|
||
assert "посуха" in p
|
||
assert "Ризики" in p
|
||
|
||
def test_max_lines(self):
|
||
now = time.time()
|
||
s = self._session_with_state(
|
||
now,
|
||
current_crop="кукурудза",
|
||
growth_stage="V6",
|
||
recent_issue="жовтизна",
|
||
risk_flags=["посуха", "заморозок"],
|
||
)
|
||
p = build_farm_state_prefix(s)
|
||
assert len(p.splitlines()) <= 5
|
||
|
||
def test_empty_if_ttl_expired(self):
|
||
old_ts = time.time() - FARM_STATE_TTL - 10
|
||
s = self._session_with_state(old_ts, current_crop="кукурудза")
|
||
assert build_farm_state_prefix(s) == ""
|
||
|
||
def test_empty_if_no_farm_state(self):
|
||
assert build_farm_state_prefix({}) == ""
|
||
|
||
def test_fail_safe(self):
|
||
assert build_farm_state_prefix(None) == "" # type: ignore
|
||
|
||
|
||
# ─── 4. State persists within session ────────────────────────────────────────
|
||
|
||
class TestFarmStatePersistence:
|
||
|
||
def test_crop_persists_across_messages(self):
|
||
session = {}
|
||
# Перше повідомлення — встановлює crop
|
||
u1 = detect_farm_state_updates("По кукурудзі є дефіцит азоту")
|
||
update_farm_state(session, u1)
|
||
# Друге повідомлення — без crop, але є farm_state
|
||
u2 = detect_farm_state_updates("Що робити?")
|
||
update_farm_state(session, u2)
|
||
# Стан зберігся
|
||
assert session["farm_state"]["current_crop"] == "кукурудза"
|
||
|
||
def test_issue_updates_on_new_message(self):
|
||
session = {}
|
||
update_farm_state(session, {"current_crop": "пшениця", "recent_issue": "жовтизна"})
|
||
update_farm_state(session, {"recent_issue": "іржа"})
|
||
assert session["farm_state"]["recent_issue"] == "іржа"
|
||
|
||
def test_stage_updates(self):
|
||
session = {}
|
||
update_farm_state(session, {"current_crop": "кукурудза", "growth_stage": "V4"})
|
||
update_farm_state(session, {"growth_stage": "V6"})
|
||
assert session["farm_state"]["growth_stage"] == "V6"
|
||
|
||
|
||
# ─── 5. Ізоляція: prefix NOT в doc / web mode ─────────────────────────────────
|
||
|
||
def _would_inject(context_mode: str, domain: str, session: dict) -> bool:
|
||
"""Симуляція умови ін'єкції з run.py."""
|
||
if context_mode == "doc" or domain == "web":
|
||
return False
|
||
return bool(build_farm_state_prefix(session))
|
||
|
||
|
||
class TestFarmStatePrefixIsolation:
|
||
|
||
def _fresh_session(self) -> dict:
|
||
now = time.time()
|
||
session = {}
|
||
update_farm_state(session, {"current_crop": "кукурудза"}, now_ts=now)
|
||
return session
|
||
|
||
def test_not_injected_in_doc_mode(self):
|
||
s = self._fresh_session()
|
||
assert not _would_inject("doc", "doc", s)
|
||
|
||
def test_not_injected_in_web_domain(self):
|
||
s = self._fresh_session()
|
||
assert not _would_inject("general", "web", s)
|
||
|
||
def test_injected_in_general_mode(self):
|
||
s = self._fresh_session()
|
||
assert _would_inject("general", "general", s)
|
||
|
||
def test_injected_in_vision_domain(self):
|
||
s = self._fresh_session()
|
||
assert _would_inject("general", "vision", s)
|
||
|
||
def test_not_injected_if_no_crop(self):
|
||
s = {"farm_state": {"growth_stage": "V6", "last_update_ts": time.time()}}
|
||
assert not _would_inject("general", "general", s)
|
||
|
||
def test_not_injected_if_ttl_expired(self):
|
||
old_ts = time.time() - FARM_STATE_TTL - 60
|
||
s = {"farm_state": {"current_crop": "кукурудза", "last_update_ts": old_ts}}
|
||
assert not _would_inject("general", "general", s)
|
||
|
||
|
||
# ─── 6. FARM_STATE_TTL constant ──────────────────────────────────────────────
|
||
|
||
class TestConstants:
|
||
def test_farm_state_ttl_30min(self):
|
||
assert FARM_STATE_TTL == 1800.0
|