File size: 3,454 Bytes
9fc3f98 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | from app.utils.climate_intelligence import (
build_climate_intelligence_context,
extract_nigeria_location,
infer_crop_and_growth_stage,
)
def test_build_climate_intelligence_context_contains_climate_risk_and_recommendations():
context = build_climate_intelligence_context(
latitude=10.0,
longitude=7.0,
crop="maize",
growth_stage="pre-planting",
days=7,
)
assert isinstance(context, dict)
assert "location" in context
assert "current_weather" in context
assert "forecast" in context
assert "climate_risk" in context
assert "recommendations" in context
assert context["location"]["country"] == "Nigeria"
assert context["climate_risk"] in {"low", "medium", "high"}
assert isinstance(context["recommendations"], list)
assert context["recommendations"]
def test_extract_nigeria_location_handles_lagos_and_ikeja():
assert extract_nigeria_location("What is the weather like in Lagos?")[0] == "Lagos"
assert extract_nigeria_location("Ikeja weather forecast")[0] == "Ikeja"
def test_infer_crop_and_growth_stage_handles_farm_questions():
crop, stage = infer_crop_and_growth_stage("Should I plant maize in Kaduna before the rains?")
assert crop == "maize"
assert stage in {"pre-planting", "planting"}
def test_recommendations_are_deduped_within_risk_branch():
"""Recommendations coming out of build_climate_intelligence_context must not
contain near-duplicate bullets, regardless of which branch produced them."""
context = build_climate_intelligence_context(
latitude=12.0,
longitude=8.0,
crop="maize",
growth_stage="pre-planting",
days=7,
)
recs = context["recommendations"]
assert isinstance(recs, list)
assert 1 <= len(recs) <= 3, f"expected 1–3 recs, got {len(recs)}"
normalized = [set(rec.lower().split()) for rec in recs]
for i, words_i in enumerate(normalized):
for j, words_j in enumerate(normalized):
if i >= j:
continue
overlap = len(words_i & words_j) / max(len(words_i), 1)
assert overlap <= 0.5, (
f"near-duplicate recommendations detected: "
f"{recs[i]!r} vs {recs[j]!r} (overlap={overlap:.2f})"
)
def test_risk_reason_is_a_single_readable_sentence():
"""risk_reason must be a single readable summary — no '|' separators and
no obvious self-repetition."""
context = build_climate_intelligence_context(
latitude=12.0,
longitude=8.0,
crop="maize",
growth_stage="pre-planting",
days=7,
)
risk_reason = context["risk_reason"]
assert isinstance(risk_reason, str)
assert "|" not in risk_reason, f"unexpected pipe in risk_reason: {risk_reason!r}"
# Defensive: the first half of the sentence must not be repeated verbatim
# in the second half.
if len(risk_reason) > 40:
half = len(risk_reason) // 2
first = risk_reason[:half].strip().lower()
second = risk_reason[half:].strip().lower()
# 25-char sequences from the first half should not reappear in the second half.
seen = {first[i : i + 25] for i in range(0, max(len(first) - 25, 1), 10)}
seen.discard("")
for chunk in seen:
assert chunk not in second, (
f"risk_reason appears to repeat itself: {risk_reason!r}"
)
|