256 lines
10 KiB
Python
256 lines
10 KiB
Python
"""Narrate-stage robustness: FCR/EEF daily insights must come out as narrative
|
|
paragraphs, not per-metric one-liners (`fcr: ...`, `bw: ...`)."""
|
|
|
|
import re
|
|
|
|
from django.test import SimpleTestCase
|
|
|
|
from apps.operations.services.insight_service import (
|
|
ANTI_HALLUCINATION_RULES,
|
|
generate_insight,
|
|
local_fallback_insight,
|
|
parse_llm_json,
|
|
sanitize_insight_narrative,
|
|
)
|
|
|
|
_METRIC_ONE_LINER = re.compile(
|
|
r"(?mi)^\s*(?:fcr|eef|bw|bobot|mortality|mortalitas|environment)\s*:\s*\S"
|
|
)
|
|
|
|
|
|
class ParseLlmJsonTests(SimpleTestCase):
|
|
def test_parses_standard_two_field_json(self):
|
|
raw = '{"kesimpulan":"Kesimpulan utuh.","insight":"Paragraf insight utuh."}'
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertEqual(out["kesimpulan"], "Kesimpulan utuh.")
|
|
self.assertEqual(out["insight"], "Paragraf insight utuh.")
|
|
|
|
def test_tolerates_trailing_comma_and_line_comment(self):
|
|
raw = (
|
|
"{\n"
|
|
' "kesimpulan": "Kesimpulan utuh.",\n'
|
|
' "insight": "Paragraf insight utuh.",\n'
|
|
" // model menulis komentar\n"
|
|
"}"
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertEqual(out["kesimpulan"], "Kesimpulan utuh.")
|
|
|
|
def test_salvages_metric_keyed_json_into_narrative_without_label_lines(self):
|
|
"""Model echoing graded metrics as keys must not yield one-liners."""
|
|
raw = (
|
|
'{"fcr": "FCR 1.70 melebihi standar CP 707 (1.615) pada hari ke-48.",'
|
|
' "bw": "Bobot badan aktual 142.1g jauh di bawah standar CP 707 (231g).",'
|
|
' "mortality": "Data mortalitas tidak tersedia."}'
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertTrue(out["kesimpulan"].startswith("FCR 1.70"))
|
|
self.assertIn("Bobot badan aktual", out["insight"])
|
|
self.assertIn("Data mortalitas tidak tersedia", out["insight"])
|
|
self.assertFalse(_METRIC_ONE_LINER.search(out["kesimpulan"]))
|
|
self.assertFalse(_METRIC_ONE_LINER.search(out["insight"]))
|
|
|
|
def test_salvages_prose_response_without_json(self):
|
|
raw = (
|
|
"Kandang 1 pada hari ke-8 menunjukkan FCR 0.18 yang lebih baik dari "
|
|
"standar CP 707 (0.864), sehingga manajemen pakan tampak efisien. "
|
|
"Namun bobot badan 142.1 gram masih jauh di bawah standar 231 gram "
|
|
"dan perlu ditindaklanjuti dengan grading segera agar pertumbuhan "
|
|
"tidak tertinggal dari target siklus."
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertIn("FCR 0.18", out["kesimpulan"])
|
|
self.assertIn("grading segera", out["insight"])
|
|
self.assertFalse(_METRIC_ONE_LINER.search(out["kesimpulan"]))
|
|
|
|
def test_tolerates_literal_newlines_inside_json_strings(self):
|
|
"""qwen2.5:3b wraps JSON string values across raw lines — seen live."""
|
|
raw = (
|
|
"```json\n"
|
|
"{\n"
|
|
' "kesimpulan": "Pada hari ke-8 di\n'
|
|
'Kandang 1, FCR 0.18 lebih rendah dari standar CP 707 (0.864).",\n'
|
|
' "insight": "Manajemen pakan perlu dijaga\n'
|
|
'agar FCR tetap ideal hingga panen."\n'
|
|
"}\n"
|
|
"```"
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertIn("Kandang 1", out["kesimpulan"])
|
|
self.assertIn("FCR 0.18", out["kesimpulan"])
|
|
self.assertIn("hingga panen", out["insight"])
|
|
|
|
def test_short_junk_without_json_returns_none(self):
|
|
self.assertIsNone(parse_llm_json("Maaf, saya tidak bisa."))
|
|
|
|
def test_oversized_prose_rejected_over_length_cap(self):
|
|
"""Salvage path must not accept unbounded untrusted prose (security MEDIUM)."""
|
|
from apps.operations.services.insight_service import _SALVAGE_MAX_CHARS
|
|
|
|
prose = (
|
|
"Paragraf narasi yang sangat panjang berisi klaim analisis AI. "
|
|
+ "Kalimat padding untuk melewati batas cap. " * 100
|
|
)
|
|
self.assertGreater(len(prose), _SALVAGE_MAX_CHARS)
|
|
self.assertIsNone(parse_llm_json(prose))
|
|
|
|
def test_oversized_metric_salvage_rejected_over_length_cap(self):
|
|
"""Metric-keyed salvage is also capped — many long values fold into prose."""
|
|
from apps.operations.services.insight_service import _SALVAGE_MAX_CHARS
|
|
|
|
big = "Kalimat metrik panjang tanpa batas. " * 100
|
|
raw = (
|
|
'{"fcr": "'
|
|
+ big
|
|
+ '", "bw": "'
|
|
+ big
|
|
+ '", "mortality": "'
|
|
+ big
|
|
+ '"}'
|
|
)
|
|
self.assertGreater(len(big) * 3, _SALVAGE_MAX_CHARS)
|
|
self.assertIsNone(parse_llm_json(raw))
|
|
|
|
def test_metric_salvage_under_cap_still_accepted(self):
|
|
raw = '{"fcr": "FCR 1.70 di atas standar.", "bw": "Bobot kurang."}'
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertIn("FCR 1.70", out["kesimpulan"])
|
|
|
|
def test_line_comment_inside_string_value_survives(self):
|
|
"""`//` inside a JSON string must not be stripped as a comment (review MEDIUM)."""
|
|
raw = (
|
|
'{"kesimpulan": "Pakan // revisi besok dikirim ke kandang 1.", '
|
|
'"insight": "Manajemen pakan perlu penyesuaian rasio harian."}'
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertIn("// revisi besok", out["kesimpulan"])
|
|
|
|
def test_trailing_comma_lookalike_inside_string_value_survives(self):
|
|
"""`, }` inside a JSON string must not be eaten by trailing-comma cleanup (review LOW)."""
|
|
raw = (
|
|
'{"kesimpulan": "Kandang 2, } blok selatan aman dari insiden.", '
|
|
'"insight": "Tidak ada anomali lingkungan pada periode berjalan ini."}'
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertIn(", }", out["kesimpulan"])
|
|
|
|
def test_prose_without_sentence_split_not_duplicated(self):
|
|
"""No `.!?` split: prose stored once as kesimpulan, insight must differ (review LOW)."""
|
|
raw = (
|
|
"FCR membaik ke 0,18 dan lebih baik dari standar CP 707 berkat "
|
|
"manajemen pakan yang konsisten ditinjau setiap hari tanpa kendala "
|
|
"berarti di kandang blok selatan periode berjalan ini"
|
|
)
|
|
out = parse_llm_json(raw)
|
|
self.assertIsNotNone(out)
|
|
self.assertIn("FCR membaik", out["kesimpulan"])
|
|
self.assertNotEqual(out["insight"], out["kesimpulan"])
|
|
|
|
|
|
class TopicWhitelistTests(SimpleTestCase):
|
|
def test_unknown_topic_rejected_before_any_work(self):
|
|
for bad in ("topic_asing<script>", "free_text_prompt_injection", "x" * 200):
|
|
with self.assertRaises(ValueError) as ctx:
|
|
generate_insight(
|
|
cycle_id=1, kandang_id=1, topic=bad, context={}
|
|
)
|
|
self.assertIn("topic", str(ctx.exception))
|
|
|
|
def test_known_frontend_topics_pass_validation(self):
|
|
"""Whitelist covers every topic the frontend can send."""
|
|
from apps.operations.services.insight_service import KNOWN_TOPICS
|
|
|
|
expected = {
|
|
"hitung_ayam",
|
|
"berat_ayam",
|
|
"fcr",
|
|
"eef",
|
|
"iot_panel",
|
|
"hitung_karung",
|
|
"dashboard",
|
|
}
|
|
self.assertEqual(set(KNOWN_TOPICS), expected)
|
|
|
|
|
|
class SanitizeNarrativeTests(SimpleTestCase):
|
|
def test_strips_metric_label_lines_into_paragraph(self):
|
|
out = sanitize_insight_narrative(
|
|
{
|
|
"kesimpulan": (
|
|
"bw: Bobot badan aktual 142.1g jauh di bawah standar "
|
|
"CP 707 (231g) pada hari ke-8."
|
|
),
|
|
"insight": (
|
|
"fcr: FCR 0.18 lebih baik dari standar CP 707 (0.864).\n"
|
|
"mortality: Data mortalitas tidak tersedia."
|
|
),
|
|
}
|
|
)
|
|
self.assertEqual(
|
|
out["kesimpulan"],
|
|
"Bobot badan aktual 142.1g jauh di bawah standar CP 707 (231g) pada hari ke-8.",
|
|
)
|
|
self.assertFalse(_METRIC_ONE_LINER.search(out["insight"]))
|
|
self.assertIn("FCR 0.18", out["insight"])
|
|
self.assertIn("Data mortalitas tidak tersedia", out["insight"])
|
|
self.assertNotIn("\n", out["insight"])
|
|
|
|
def test_leaves_normal_paragraphs_and_colon_words_alone(self):
|
|
parsed = {
|
|
"kesimpulan": "Catatan lapangan: data lengkap.",
|
|
"insight": "Paragraf pertama.\n\nParagraf kedua.",
|
|
}
|
|
out = sanitize_insight_narrative(parsed)
|
|
self.assertEqual(out["kesimpulan"], parsed["kesimpulan"])
|
|
self.assertEqual(out["insight"], parsed["insight"])
|
|
|
|
def test_still_collapses_duplicate_kandang(self):
|
|
out = sanitize_insight_narrative(
|
|
{"kesimpulan": "Kandang Kandang 2 mortalitas tinggi.", "insight": "Pantau."}
|
|
)
|
|
self.assertEqual(out["kesimpulan"], "Kandang 2 mortalitas tinggi.")
|
|
|
|
|
|
class LocalFallbackNarrativeTests(SimpleTestCase):
|
|
GRADED = {
|
|
"analyses": {
|
|
"bw": {
|
|
"status": "critical",
|
|
"message": "Bobot badan aktual 142.1g jauh di bawah standar CP 707 (231g).",
|
|
},
|
|
"fcr": {"status": "ok", "message": "FCR 0.18 sesuai standar CP 707 (0.864)."},
|
|
"mortality": {"status": "unknown", "message": "Data mortalitas tidak tersedia."},
|
|
}
|
|
}
|
|
|
|
def test_fallback_through_sanitize_has_no_metric_one_liners(self):
|
|
out = sanitize_insight_narrative(local_fallback_insight(dict(self.GRADED), "fcr"))
|
|
self.assertFalse(_METRIC_ONE_LINER.search(out["kesimpulan"]))
|
|
self.assertFalse(_METRIC_ONE_LINER.search(out["insight"]))
|
|
self.assertIn("Bobot badan aktual", out["kesimpulan"])
|
|
self.assertIn("FCR 0.18", out["insight"])
|
|
|
|
def test_fallback_without_messages_still_reports_unknown(self):
|
|
out = local_fallback_insight({"analyses": {}}, "eef")
|
|
self.assertIn("tidak cukup", out["kesimpulan"])
|
|
|
|
|
|
class NarrativePromptRuleTests(SimpleTestCase):
|
|
def test_rules_require_paragraph_output_and_ban_metric_one_liners(self):
|
|
rules = ANTI_HALLUCINATION_RULES.lower()
|
|
self.assertIn("paragraf", rules)
|
|
self.assertIn("fcr: ...", rules)
|
|
|
|
def test_end_cycle_format_replace_target_still_present(self):
|
|
"""generate_insight swaps this exact block for end-cycle rules."""
|
|
target = 'FORMAT OUTPUT (JSON SAJA):\n{"kesimpulan":"...","insight":"..."}'
|
|
self.assertIn(target, ANTI_HALLUCINATION_RULES)
|