"""Tests for SQLCoach intent detection and specialized renderers."""
import unittest
from unittest.mock import Mock
from app.coach_local import SQLCoach
class TestIntentDetection(unittest.TestCase):
"""Test SQLCoach._detect_intent routing."""
def setUp(self):
"""Set up coach instance for non-static _detect_intent."""
model_manager = Mock()
self.coach = SQLCoach(model_manager)
def test_greeting_intent_hello(self):
self.assertEqual(self.coach._detect_intent("hello"), "GREETING")
def test_greeting_intent_hi(self):
self.assertEqual(self.coach._detect_intent("hi"), "GREETING")
def test_greeting_intent_test(self):
self.assertEqual(self.coach._detect_intent("test"), "GREETING")
def test_greeting_intent_case_insensitive(self):
self.assertEqual(self.coach._detect_intent("HELLO"), "GREETING")
self.assertEqual(self.coach._detect_intent("Hi there"), "GREETING")
def test_meta_intent_help(self):
self.assertEqual(self.coach._detect_intent("help"), "META")
def test_meta_intent_capabilities(self):
self.assertEqual(self.coach._detect_intent("capabilities"), "META")
def test_meta_intent_how_does_this_work(self):
self.assertEqual(self.coach._detect_intent("how does this work"), "META")
def test_meta_intent_case_insensitive(self):
self.assertEqual(self.coach._detect_intent("HELP"), "META")
self.assertEqual(self.coach._detect_intent("How To Use"), "META")
def test_skill_delta_intent_fell_off(self):
self.assertEqual(self.coach._detect_intent("which skill fell off"), "SKILL_DELTA")
def test_skill_delta_intent_dropped(self):
self.assertEqual(self.coach._detect_intent("skill dropped"), "SKILL_DELTA")
def test_skill_delta_intent_decline(self):
self.assertEqual(self.coach._detect_intent("skill decline"), "SKILL_DELTA")
def test_report_intent_report(self):
self.assertEqual(self.coach._detect_intent("report"), "REPORT")
def test_report_intent_full_review(self):
self.assertEqual(self.coach._detect_intent("full review"), "REPORT")
def test_report_intent_coach_me(self):
self.assertEqual(self.coach._detect_intent("coach me"), "REPORT")
def test_crit_bucket_trend_intent_crit(self):
self.assertEqual(self.coach._detect_intent("crit"), "CRIT_BUCKET_TREND")
def test_crit_bucket_trend_intent_crit_rate(self):
self.assertEqual(self.coach._detect_intent("crit rate"), "CRIT_BUCKET_TREND")
def test_crit_bucket_trend_intent_trend(self):
self.assertEqual(self.coach._detect_intent("trend"), "CRIT_BUCKET_TREND")
def test_crit_bucket_trend_intent_bucket(self):
self.assertEqual(self.coach._detect_intent("how do crit rates trend"), "CRIT_BUCKET_TREND")
def test_skills_intent_skill(self):
self.assertEqual(self.coach._detect_intent("skill"), "SKILLS")
def test_skills_intent_which_skill(self):
self.assertEqual(self.coach._detect_intent("which skill should I focus on"), "SKILLS")
def test_skills_intent_top(self):
self.assertEqual(self.coach._detect_intent("top damage"), "SKILLS")
def test_skills_intent_rotation(self):
self.assertEqual(self.coach._detect_intent("rotation"), "SKILLS")
def test_runs_intent_run(self):
self.assertEqual(self.coach._detect_intent("run"), "RUNS")
def test_runs_intent_best_run(self):
self.assertEqual(self.coach._detect_intent("best run"), "RUNS")
def test_runs_intent_improve(self):
self.assertEqual(self.coach._detect_intent("improve"), "RUNS")
def test_runs_intent_better(self):
self.assertEqual(self.coach._detect_intent("can I do better"), "RUNS")
def test_default_intent_empty(self):
self.assertEqual(self.coach._detect_intent(""), "DEFAULT")
def test_default_intent_none(self):
self.assertEqual(self.coach._detect_intent(None), "DEFAULT")
def test_default_intent_generic(self):
self.assertEqual(self.coach._detect_intent("tell me about my performance"), "DEFAULT")
def test_intent_priority_greeting_over_default(self):
"""Test that 'hello' is GREETING not DEFAULT."""
self.assertEqual(self.coach._detect_intent("hello"), "GREETING")
def test_intent_priority_skill_delta_over_skills(self):
"""Test that 'skill fell off' is SKILL_DELTA not SKILLS."""
self.assertEqual(self.coach._detect_intent("which skill fell off"), "SKILL_DELTA")
def test_intent_priority_report_over_default(self):
"""Test that 'report' is REPORT not DEFAULT."""
self.assertEqual(self.coach._detect_intent("provide a report"), "REPORT")
def test_action_plan_intent_top_3(self):
"""Test ACTION_PLAN intent detection."""
self.assertEqual(self.coach._detect_intent("give me top 3 changes"), "ACTION_PLAN")
def test_action_plan_intent_three_changes(self):
self.assertEqual(self.coach._detect_intent("what are three changes to improve"), "ACTION_PLAN")
def test_action_plan_intent_action_plan(self):
self.assertEqual(self.coach._detect_intent("action plan"), "ACTION_PLAN")
def test_spike_analysis_intent_spike(self):
"""Test SPIKE_ANALYSIS intent detection."""
self.assertEqual(self.coach._detect_intent("where did my damage spike"), "SPIKE_ANALYSIS")
def test_spike_analysis_intent_burst(self):
self.assertEqual(self.coach._detect_intent("burst damage"), "SPIKE_ANALYSIS")
def test_spike_analysis_intent_peak_damage(self):
self.assertEqual(self.coach._detect_intent("peak damage"), "SPIKE_ANALYSIS")
def test_frontload_intent_frontload(self):
"""Test FRONTLOAD intent detection."""
self.assertEqual(self.coach._detect_intent("should I front-load damage"), "FRONTLOAD")
def test_frontload_intent_opener(self):
self.assertEqual(self.coach._detect_intent("opener rotation"), "FRONTLOAD")
def test_frontload_intent_reorder_rotation(self):
self.assertEqual(self.coach._detect_intent("reorder rotation"), "FRONTLOAD")
def test_intent_keyword_substring_match(self):
"""Test that keywords work as substrings."""
self.assertEqual(self.coach._detect_intent("how do crit rates trend"), "CRIT_BUCKET_TREND")
self.assertEqual(self.coach._detect_intent("what skills did I use"), "SKILLS")
self.assertEqual(self.coach._detect_intent("was my previous run better"), "RUNS")
class TestGreetingRoute(unittest.TestCase):
"""Test greeting intent handling (no tool calls, no model inference)."""
def setUp(self):
"""Set up mock model manager."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_greeting_answer_no_tool_trace(self):
"""Greeting should return immediately with empty tool trace."""
response, trace = self.coach.answer(
"hello",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: {},
)
self.assertEqual(trace, [])
self.assertIn("Hello", response)
self.assertIn("DPS Coach", response)
def test_greeting_answer_contains_help_text(self):
"""Greeting response should offer guidance."""
response, _ = self.coach.answer(
"hi",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: {},
)
self.assertIn("Improve DPS", response)
self.assertIn("combat performance", response)
def test_test_message_answer(self):
"""'test' should also trigger greeting."""
response, trace = self.coach.answer(
"test",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: {},
)
self.assertEqual(trace, [])
self.assertIn("DPS Coach", response)
class TestMetaRoute(unittest.TestCase):
"""Test meta intent handling (no tool calls, static response)."""
def setUp(self):
"""Set up mock model manager."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_meta_answer_no_tool_trace(self):
"""Meta should return immediately with empty tool trace."""
response, trace = self.coach.answer(
"help",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: {},
)
self.assertEqual(trace, [])
def test_meta_answer_capability_card(self):
"""Meta response should be capability card."""
response, _ = self.coach.answer(
"capabilities",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: {},
)
self.assertIn("Capability card", response)
self.assertIn("Qwen2.5-7B", response)
def test_how_does_this_work_meta(self):
"""'how does this work' should trigger meta."""
response, trace = self.coach.answer(
"how does this work",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: {},
)
self.assertEqual(trace, [])
class TestCritBucketTrendRoute(unittest.TestCase):
"""Test CRIT_BUCKET_TREND renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_crit_trend_includes_timeline_analysis(self):
"""Crit trend response should analyze bucket timeline."""
packet = {
"meta": {"run_id": "run_123", "limits": {"bucket_seconds": 5}},
"run_summary": {"rows": [["run_123", 10, 1000, 20, 50, 25]]}, # [id, count, dmg, dur, dps, crit%]
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {
"columns": ["bucket_s", "hits", "crit_hits", "crit_rate_pct", "damage"],
"rows": [
[0, 12, 2, 16.7, 200], # bucket 0s: 12 hits, 2 crits = 16.7%
[5, 10, 5, 50.0, 160], # bucket 5s: 10 hits, 5 crits = 50%
[10, 15, 3, 20.0, 240], # bucket 10s: 15 hits, 3 crits = 20%
],
},
"skill_deltas": {"rows": []},
"notes": [],
}
response, trace = self.coach.answer(
"how do crit rates trend",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Should have one tool trace (get_analysis_packet)
self.assertEqual(len(trace), 1)
self.assertEqual(trace[0]["tool"], "get_analysis_packet")
# Response should contain crit analysis keywords
self.assertIn("Crit Rate Timeline", response)
self.assertIn("run_123", response)
self.assertIn("Bucket Details", response)
# Check for mm:ss time format
self.assertIn("00:00", response) # First bucket at 0s = 00:00
def test_crit_trend_shows_trend_direction(self):
"""Crit trend should show direction: rising, falling, or stable."""
packet = {
"meta": {"run_id": "run_1", "limits": {"bucket_seconds": 5}},
"run_summary": {"rows": [["run_1", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {
"columns": ["bucket_s", "hits", "crit_hits", "crit_rate_pct", "damage"],
"rows": [
[0, 15, 3, 20.0, 200], # early: 20%
[5, 15, 8, 53.3, 200],
[10, 15, 12, 80.0, 200], # late: 80%
],
},
"skill_deltas": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"crit trend",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Rising trend should be detected
self.assertIn("rising", response.lower())
# Check for mm:ss format
self.assertIn(":", response)
def test_crit_trend_shows_peak_bucket(self):
"""Crit trend should identify peak bucket."""
packet = {
"meta": {"run_id": "run_2", "limits": {"bucket_seconds": 5}},
"run_summary": {"rows": [["run_2", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {
"columns": ["bucket_s", "hits", "crit_hits", "crit_rate_pct", "damage"],
"rows": [
[0, 15, 3, 20.0, 200], # 20%
[5, 15, 15, 100.0, 200], # 100% ← peak
[10, 15, 8, 53.3, 200], # 53%
],
},
"skill_deltas": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"crit trend",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Peak crit", response)
self.assertIn("100", response)
class TestSkillsRoute(unittest.TestCase):
"""Test SKILLS intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_skills_response_includes_skill_table(self):
"""Skills response should include damage table by skill."""
packet = {
"meta": {"run_id": "run_s1", "limits": {}},
"run_summary": {"rows": [["run_s1", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {
"columns": ["skill_name", "total_damage", "total_hits", "avg_damage", "dmg_rank", "dmg_pct"],
"rows": [
["Fireball", 400, 20, 20, 1, 40],
["Ice Storm", 300, 15, 20, 2, 30],
],
},
"skill_efficiency": {
"columns": ["skill_name", "avg_damage"],
"rows": [["Ice Storm", 20]],
},
"timeline": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"which skill should I focus",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Skill Focus", response)
self.assertIn("Fireball", response)
self.assertIn("Ice Storm", response)
self.assertIn("Skill | Damage", response)
def test_skills_response_includes_efficiency_highlight(self):
"""Skills response should highlight best efficiency skill."""
packet = {
"meta": {"run_id": "run_s2", "limits": {}},
"run_summary": {"rows": [["run_s2", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {
"columns": ["skill_name", "total_damage", "total_hits", "avg_damage", "dmg_rank", "dmg_pct"],
"rows": [["Ability A", 500, 20, 25, 1, 50]],
},
"skill_efficiency": {
"columns": ["skill_name", "avg_damage"],
"rows": [["Ability B", 30]], # Higher avg hit
},
"timeline": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"top damage",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Ability A", response)
self.assertIn("Ability B", response)
self.assertIn("Best efficiency", response)
class TestRunsRoute(unittest.TestCase):
"""Test RUNS intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_runs_response_includes_run_table(self):
"""Runs response should include historical run data."""
packet = {
"meta": {"run_id": "run_r1", "limits": {}},
"run_summary": {"rows": [["run_r1", 20, 1000, 20, 50, 25]]},
"runs_last_n": [
{"run_id": "run_r1", "total_hits": 100, "total_damage": 1000, "duration_seconds": 20, "dps": 50, "crit_rate_pct": 25},
{"run_id": "run_r2", "total_hits": 95, "total_damage": 900, "duration_seconds": 20, "dps": 45, "crit_rate_pct": 22},
{"run_id": "run_r3", "total_hits": 110, "total_damage": 1100, "duration_seconds": 20, "dps": 55, "crit_rate_pct": 28},
],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"best run today",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Run Performance", response)
self.assertIn("Recent Run History", response)
self.assertIn("best", response.lower())
def test_runs_response_shows_variability(self):
"""Runs response should highlight DPS variability."""
packet = {
"meta": {"run_id": "run_r4", "limits": {}},
"run_summary": {"rows": [["run_r4", 20, 1000, 20, 50, 25]]},
"runs_last_n": [
{"run_id": "run_r4", "total_hits": 120, "total_damage": 1200, "duration_seconds": 20, "dps": 60, "crit_rate_pct": 30}, # best
{"run_id": "run_r5", "total_hits": 60, "total_damage": 600, "duration_seconds": 20, "dps": 30, "crit_rate_pct": 15}, # worst
],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"improve my runs",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Should compare best vs worst
self.assertIn("Variability", response)
class TestDefaultRoute(unittest.TestCase):
"""Test DEFAULT (generic summary) intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_default_response_includes_all_sections(self):
"""Default response should include insights, evidence, actions, next questions."""
packet = {
"meta": {"run_id": "run_d1", "limits": {"bucket_seconds": 5, "top_k_skills": 10, "last_n_runs": 5}},
"run_summary": {
"rows": [["run_d1", 10, 1000, 20, 50, 25]]
},
"runs_last_n": [],
"top_skills": {
"columns": ["skill_name", "damage", "hits", "avg"],
"rows": [["Skill1", 500, 20, 25]],
},
"skill_efficiency": {"rows": [["Skill2", 30]]},
"timeline": {"rows": [[0, 10, 2, 20.0, 200]]},
"skill_deltas": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"tell me about my performance",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Insights:", response)
self.assertIn("Evidence:", response)
self.assertIn("Actions:", response)
self.assertIn("Next", response)
def test_default_response_generic_question(self):
"""Generic questions should use DEFAULT intent and packet rendering."""
packet = {
"meta": {"run_id": "run_d2", "limits": {"bucket_seconds": 5, "top_k_skills": 10, "last_n_runs": 5}},
"run_summary": {"rows": [["run_d2", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"skill_deltas": {"rows": []},
"notes": [],
}
response, trace = self.coach.answer(
"summarize my session",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Should have called analysis_callback (one trace)
self.assertEqual(len(trace), 1)
class TestSkillDeltaRoute(unittest.TestCase):
"""Test SKILL_DELTA intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_skill_delta_detects_falloffs(self):
"""Skill delta should identify fall-offs vs prior runs."""
packet = {
"meta": {"run_id": "run_sd1", "limits": {}},
"run_summary": {"rows": [["run_sd1", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"skill_deltas": {
"columns": [
"skill_name", "last_share_pct", "prior_avg_share_pct", "delta_share_pp",
"last_hits", "prior_avg_hits", "delta_hits",
"last_crit_rate_pct", "prior_avg_crit_rate_pct", "delta_crit_pp"
],
"rows": [
["Fireball", 20.0, 25.0, -5.0, 10, 15, -5, 25.0, 30.0, -5.0], # Big fall-off
["Ice Storm", 30.0, 28.0, 2.0, 20, 18, 2, 20.0, 18.0, 2.0], # Improvement
],
},
"notes": [],
}
response, _ = self.coach.answer(
"which skill fell off",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Fall-Off", response)
self.assertIn("Fireball", response)
self.assertIn("-5", response) # Delta value
self.assertIn("prior runs", response.lower())
def test_skill_delta_shows_improvements(self):
"""Skill delta should show improvements as well."""
packet = {
"meta": {"run_id": "run_sd2", "limits": {}},
"run_summary": {"rows": [["run_sd2", 10, 1000, 20, 50, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"skill_deltas": {
"columns": [
"skill_name", "last_share_pct", "prior_avg_share_pct", "delta_share_pp",
"last_hits", "prior_avg_hits", "delta_hits",
"last_crit_rate_pct", "prior_avg_crit_rate_pct", "delta_crit_pp"
],
"rows": [
["Thunder", 35.0, 30.0, 5.0, 25, 20, 5, 30.0, 25.0, 5.0], # Improvement
],
},
"notes": [],
}
response, _ = self.coach.answer(
"skill dropped",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Improvement", response)
self.assertIn("Thunder", response)
class TestReportRoute(unittest.TestCase):
"""Test REPORT intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_report_includes_strengths_leaks_actions(self):
"""Report should include Strengths, Leaks, and Top 3 changes."""
packet = {
"meta": {"run_id": "run_r1", "limits": {}},
"run_summary": {"rows": [["run_r1", 100, 10000, 100, 100, 28]]},
"runs_last_n": [
{"run_id": "run_r1", "duration_seconds": 100, "dps": 100, "total_damage": 10000, "total_hits": 100, "crit_rate_pct": 28},
{"run_id": "run_r2", "duration_seconds": 100, "dps": 90, "total_damage": 9500, "total_hits": 98, "crit_rate_pct": 26},
{"run_id": "run_r3", "duration_seconds": 100, "dps": 95, "total_damage": 9800, "total_hits": 101, "crit_rate_pct": 27},
],
"top_skills": {
"columns": ["skill_name", "damage", "hits", "avg", "crit", "share"],
"rows": [
["BigHit", 5000, 50, 100, 30, 50],
],
},
"skill_efficiency": {"rows": []},
"timeline": {"rows": [[0, 50, 10, 20.0, 1000]]},
"skill_deltas": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"provide a full report",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Strengths", response)
self.assertIn("Leaks", response)
self.assertIn("Top 3 Changes", response)
self.assertIn("Data-Backed", response)
def test_report_detects_variability_leak(self):
"""Report should detect high DPS variability as a leak."""
packet = {
"meta": {"run_id": "run_r2", "limits": {}},
"run_summary": {"rows": [["run_r2", 100, 10000, 100, 100, 25]]},
"runs_last_n": [
{"run_id": "run_r2", "duration_seconds": 100, "dps": 100, "total_damage": 10000, "total_hits": 100, "crit_rate_pct": 25},
{"run_id": "run_r3", "duration_seconds": 100, "dps": 50, "total_damage": 5000, "total_hits": 90, "crit_rate_pct": 20}, # Big spread
{"run_id": "run_r4", "duration_seconds": 100, "dps": 120, "total_damage": 12000, "total_hits": 110, "crit_rate_pct": 27},
],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"skill_deltas": {"rows": []},
"notes": [],
}
response, _ = self.coach.answer(
"coach me",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("variability", response.lower())
class TestFollowupWhyRoute(unittest.TestCase):
"""Test FOLLOWUP_WHY intent with state management."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_followup_why_after_skill_delta(self):
"""FOLLOWUP_WHY should only trigger after SKILL_DELTA intent."""
packet_with_deltas = {
"meta": {"run_id": "run_f1", "limits": {}},
"run_summary": {"rows": [["run_f1", 100, 10000, 100, 100, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"skill_deltas": {
"rows": [
["Fireball", 20.0, 25.0, -5.0, 20, 25, -5, 30.0, 35.0, -5.0],
],
"notes": [],
},
"windows": {},
"actions": {},
"notes": [],
}
# First, ask about skill delta to set state
_, _ = self.coach.answer(
"which skill fell off",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet_with_deltas,
)
# Now ask followup
response, trace = self.coach.answer(
"why did the top fall-off skill lose damage share",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet_with_deltas,
)
self.assertEqual(len(trace), 1) # One packet call
self.assertIn("Fireball", response)
self.assertIn("fall off", response.lower())
self.assertIn("delta", response.lower() or "share" in response.lower())
def test_followup_why_uses_stored_skill(self):
"""FOLLOWUP_WHY should reference the top fall-off skill from state."""
packet = {
"meta": {"run_id": "run_f2", "limits": {}},
"run_summary": {"rows": [["run_f2", 100, 10000, 100, 100, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": []},
"skill_deltas": {
"rows": [
["IceStorm", 15.0, 20.0, -5.0, 15, 20, -5, 25.0, 30.0, -5.0],
["Thunder", 10.0, 12.0, -2.0, 10, 12, -2, 20.0, 22.0, -2.0],
],
"notes": [],
},
"windows": {},
"actions": {},
"notes": [],
}
# Set state with SKILL_DELTA
self.coach.answer(
"which skill fell off",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Followup should reference IceStorm (top fall-off)
response, _ = self.coach.answer(
"why that skill",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("IceStorm", response)
class TestActionPlanRoute(unittest.TestCase):
"""Test ACTION_PLAN intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_action_plan_contains_numbered_items(self):
"""ACTION_PLAN should return 3 numbered changes."""
packet = {
"meta": {"run_id": "run_a1", "limits": {}},
"run_summary": {"rows": [["run_a1", 100, 10000, 100, 100, 25]]},
"runs_last_n": [
{"run_id": "run_a1", "duration_seconds": 100, "dps": 100, "total_damage": 10000, "total_hits": 100, "crit_rate_pct": 25},
{"run_id": "run_a2", "duration_seconds": 100, "dps": 50, "total_damage": 5000, "total_hits": 90, "crit_rate_pct": 20},
{"run_id": "run_a3", "duration_seconds": 100, "dps": 120, "total_damage": 12000, "total_hits": 110, "crit_rate_pct": 28},
],
"top_skills": {
"rows": [
["BigHit", 5000, 50, 100, 30, 50],
],
},
"skill_efficiency": {
"rows": [
["Strong", 100, 5000, 100, 25, 50],
["Weak", 50, 1000, 20, 10, 20],
],
},
"skill_deltas": {
"rows": [
["Fireball", 20.0, 25.0, -5.0, 20, 25, -5, 30.0, 35.0, -5.0],
],
"notes": [],
},
"timeline": {"rows": []},
"windows": {},
"actions": {"top_levers": ["Restore Fireball", "Optimize Weak", "Consistency"]},
"notes": [],
}
response, trace = self.coach.answer(
"give me top 3 changes next run to increase DPS",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertEqual(len(trace), 1)
self.assertIn("1.", response)
self.assertIn("2.", response)
self.assertIn("3.", response)
def test_action_plan_includes_numbers(self):
"""ACTION_PLAN should include data-backed numbers."""
packet = {
"meta": {"run_id": "run_a2", "limits": {}},
"run_summary": {"rows": [["run_a2", 100, 10000, 100, 100, 25]]},
"runs_last_n": [
{"run_id": "run_a2", "duration_seconds": 100, "dps": 100, "total_damage": 10000, "total_hits": 100, "crit_rate_pct": 25},
],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": [["Weak", 50, 1000, 20, 10, 20]]},
"skill_deltas": {
"rows": [
["Fireball", 20.0, 28.0, -8.0, 20, 28, -8, 30.0, 35.0, -5.0],
],
"notes": [],
},
"timeline": {"rows": []},
"windows": {},
"actions": {},
"notes": [],
}
response, _ = self.coach.answer(
"top 3 changes",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Should include delta or efficiency numbers
has_numbers = any(x in response for x in ["pp", "%", "hits", "avg"])
self.assertTrue(has_numbers, "ACTION_PLAN should include data numbers")
class TestSpikeAnalysisRoute(unittest.TestCase):
"""Test SPIKE_ANALYSIS intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_spike_analysis_includes_mmss_times(self):
"""SPIKE_ANALYSIS should show mm:ss times for spike windows."""
packet = {
"meta": {"run_id": "run_s1", "limits": {}},
"run_summary": {"rows": [["run_s1", 100, 10000, 100, 100, 25]]},
"runs_last_n": [],
"top_skills": {
"rows": [
["Thunder", 5000, 50, 100, 30, 50],
],
},
"skill_efficiency": {"rows": []},
"timeline": {
"rows": [
[0, 50, 10, 20.0, 1000],
[5, 60, 15, 25.0, 2000],
[10, 55, 12, 22.0, 1500],
],
},
"skill_deltas": {"rows": []},
"windows": {
"top_damage_windows": [
{"start_s": 5, "end_s": 10, "damage": 2000, "top_skills": ["Thunder"]},
{"start_s": 10, "end_s": 15, "damage": 1500, "top_skills": ["Fireball"]},
],
},
"actions": {},
"notes": [],
}
response, trace = self.coach.answer(
"where did my damage spike and why",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertEqual(len(trace), 1)
self.assertIn("00:", response) # mm:ss format
self.assertIn("damage", response.lower())
def test_spike_analysis_attributes_to_skills(self):
"""SPIKE_ANALYSIS should attribute spikes to top skills."""
packet = {
"meta": {"run_id": "run_s2", "limits": {}},
"run_summary": {"rows": [["run_s2", 100, 10000, 100, 100, 25]]},
"runs_last_n": [],
"top_skills": {"rows": []},
"skill_efficiency": {"rows": []},
"timeline": {"rows": [[0, 50, 10, 20.0, 1000]]},
"skill_deltas": {"rows": []},
"windows": {
"top_damage_windows": [
{"start_s": 15, "end_s": 20, "damage": 3000, "top_skills": ["Meteor", "Inferno"]},
],
},
"actions": {},
"notes": [],
}
response, _ = self.coach.answer(
"damage spike",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Meteor", response)
class TestFrontloadRoute(unittest.TestCase):
"""Test FRONTLOAD intent renderer."""
def setUp(self):
"""Set up mock model manager and coach."""
from unittest.mock import Mock
self.model_manager = Mock()
self.coach = SQLCoach(self.model_manager)
def test_frontload_includes_early_vs_overall(self):
"""FRONTLOAD should show early damage share vs overall."""
packet = {
"meta": {"run_id": "run_f1", "limits": {}},
"run_summary": {"rows": [["run_f1", 100, 10000, 100, 100, 25]]},
"runs_last_n": [],
"top_skills": {
"rows": [
["BigHit", 5000, 50, 100, 30, 50],
],
},
"skill_efficiency": {"rows": []},
"timeline": {
"rows": [
[0, 50, 10, 20.0, 1000],
[55, 60, 15, 25.0, 2000],
],
},
"skill_deltas": {"rows": []},
"windows": {
"early_window": {"start_s": 0, "end_s": 60, "damage": 3000, "top_skills": []},
},
"actions": {},
"notes": [],
}
response, trace = self.coach.answer(
"should I reorder rotation to front-load damage",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertEqual(len(trace), 1)
self.assertIn("Early", response)
self.assertIn("%", response) # Share percentage
self.assertIn("60", response) # 60s window
def test_frontload_recommends_opener_reorder(self):
"""FRONTLOAD should recommend reordering when early share is low."""
packet = {
"meta": {"run_id": "run_f2", "limits": {}},
"run_summary": {"rows": [["run_f2", 100, 10000, 100, 100, 25]]},
"runs_last_n": [],
"top_skills": {
"rows": [
["BigHit", 5000, 50, 100, 30, 50],
],
},
"skill_efficiency": {"rows": []},
"timeline": {
"rows": [
[0, 50, 10, 20.0, 1000],
[65, 60, 15, 25.0, 2000],
],
},
"skill_deltas": {"rows": []},
"windows": {
"early_window": {"start_s": 0, "end_s": 60, "damage": 2000, "top_skills": ["BigHit"]},
},
"actions": {},
"notes": [],
}
response, _ = self.coach.answer(
"front-load",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
self.assertIn("Recommendation", response)
class TestSessionPersistence(unittest.TestCase):
"""Test that coach session ID persists and state tracks across consecutive calls."""
def setUp(self):
"""Set up coach instance."""
model_manager = Mock()
self.coach = SQLCoach(model_manager)
# Record initial session ID
self.initial_session_id = self.coach._coach_session_id
def test_coach_session_id_stable(self):
"""Verify that coach_session_id is created and stable."""
self.assertIsNotNone(self.initial_session_id)
self.assertEqual(len(self.initial_session_id), 32) # uuid4().hex is 32 chars
# Call answer and verify session ID doesn't change
self.assertEqual(self.coach._coach_session_id, self.initial_session_id)
def test_consecutive_calls_with_state_persistence(self):
"""Test SKILL_DELTA followed by FOLLOWUP_WHY with shared state.
1. Call SKILL_DELTA question → coach stores focus_skill in state
2. Call FOLLOWUP_WHY question → coach uses stored focus_skill
3. Verify same session_id in both traces
4. Verify follow-up uses correct skill from state
"""
# Create a stable packet
packet = {
"meta": {"run_id": "run_123"},
"run_summary": {
"rows": [["run_123", 250, 50000, 120.0, 416.7, 48.0]]
},
"top_skills": {"rows": [["Fireball", 5000, 100, 50]]},
"runs_last_n": [
{"run_id": "run_123", "duration_seconds": 100, "dps": 10000, "total_damage": 1000000, "total_hits": 250, "crit_rate_pct": 48.0},
],
"timeline": {"rows": []},
"skill_deltas": {
"columns": [
"skill_name", "last_share_pct", "prior_avg_share_pct", "delta_share_pp",
"last_hits", "prior_avg_hits", "delta_hits",
"last_crit_rate_pct", "prior_avg_crit_rate_pct", "delta_crit_pp"
],
"rows": [
["Fireball", 46.0, 50.0, -4.0, 100, 104, -4, 50.0, 50.0, 0.0], # Fireball fell off
["IceShower", 52.0, 50.0, 2.0, 120, 118, 2, 45.0, 45.0, 0.0],
]
},
"windows": {
"early_window": {"start_s": 0, "end_s": 60, "damage": 2000, "top_skills": ["BigHit"]},
},
"actions": {},
"notes": [],
}
# Call 1: SKILL_DELTA question
q1_answer, q1_trace = self.coach.answer(
"which skill fell off",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Verify trace includes session_id
self.assertEqual(len(q1_trace), 1)
self.assertIn("coach_session_id", q1_trace[0])
self.assertEqual(q1_trace[0]["coach_session_id"], self.initial_session_id)
session_id_from_q1 = q1_trace[0]["coach_session_id"]
# Verify state was updated with focus skill
self.assertEqual(self.coach._state["last_intent"], "SKILL_DELTA")
stored_skill = self.coach._state["last_focus_skill"]
self.assertIsNotNone(stored_skill)
# Call 2: FOLLOWUP_WHY question (follow-up to SKILL_DELTA)
q2_answer, q2_trace = self.coach.answer(
"why did that skill fall off",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Verify trace includes same session_id
self.assertEqual(len(q2_trace), 1)
self.assertIn("coach_session_id", q2_trace[0])
self.assertEqual(q2_trace[0]["coach_session_id"], self.initial_session_id)
session_id_from_q2 = q2_trace[0]["coach_session_id"]
# Both traces should have identical session IDs
self.assertEqual(session_id_from_q1, session_id_from_q2)
# Verify FOLLOWUP_WHY was detected
self.assertEqual(self.coach._state["last_intent"], "FOLLOWUP_WHY")
# Verify response uses the stored focus skill (Fireball)
self.assertIn("Fireball", q2_answer)
class TestRunsAnalysisDPSBound(unittest.TestCase):
"""Test that RUNS view displays correct DPS metrics, not confused with total_damage."""
def setUp(self):
model_manager = Mock()
self.coach = SQLCoach(model_manager)
def test_runs_analysis_trace_handles_list_runs_payload(self):
"""Ensure analysis trace works when runs_last_n is list-of-dicts."""
packet = {
"meta": {"run_id": "run_list", "limits": {}},
"run_summary": {"rows": [["run_list", 10, 1000, 20, 50, 25]]},
"runs_last_n": [
{"run_id": "run_list", "total_hits": 10, "total_damage": 1000, "duration_seconds": 20, "dps": 50, "crit_rate_pct": 25},
{"run_id": "run_other", "total_hits": 9, "total_damage": 900, "duration_seconds": 19, "dps": 47, "crit_rate_pct": 22},
],
"top_skills": {"rows": [["Fireball", 500, 5, 100, 40, 50]]},
"timeline": {"rows": [[0, 5, 2, 40.0, 300], [10, 5, 3, 60.0, 700]]},
}
response, trace = self.coach.answer(
"lowest DPS run?",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Should not raise and should count runs correctly
self.assertIsNotNone(trace)
counts = trace[0].get("counts", {}) if trace else {}
self.assertEqual(counts.get("runs_last_n"), 2)
self.assertEqual(counts.get("top_skills_last_run"), 1)
self.assertEqual(counts.get("timeline_buckets_last_run"), 2)
def test_runs_analysis_dps_not_exceeding_bound(self):
"""Verify DPS values in RUNS response are reasonable (< 1M for sample data).
Sample data has ~350 hits over 180s = ~1.9 DPS per hit with ~240 damage per hit
= ~472 DPS expected. If we see values like 85,000 or 13M, thats total_damage
being mislabeled as DPS.
"""
packet = {
"meta": {"run_id": "run_123"},
"run_summary": {
"rows": [["run_123", 350, 85000, 180.0, 472.2, 52.0]]
},
"runs_last_n": [
{"run_id": "run_123", "total_hits": 350, "total_damage": 85000, "duration_seconds": 180.0, "dps": 472.2, "crit_rate_pct": 52.0, "last_ts": "2025-12-06T00:00:00"},
{"run_id": "run_122", "total_hits": 340, "total_damage": 80000, "duration_seconds": 170.0, "dps": 470.6, "crit_rate_pct": 50.0, "last_ts": "2025-12-05T00:00:00"},
{"run_id": "run_121", "total_hits": 345, "total_damage": 82000, "duration_seconds": 175.0, "dps": 468.6, "crit_rate_pct": 51.0, "last_ts": "2025-12-04T00:00:00"},
],
}
response, trace = self.coach.answer(
"lowest DPS run?",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
# Verify DPS column values are reasonable (< 1,000,000) and not confused with total_damage
lines = response.split("\n")
table_rows = [
l for l in lines
if "|" in l and "Run #" not in l and "---" not in l and l.strip().startswith(" ")
]
for line in table_rows:
parts = [p.strip() for p in line.split("|")]
if len(parts) < 5:
continue
dps_part = parts[2]
dmg_part = parts[3]
dps_val = int(dps_part.replace(",", "")) if dps_part else 0
dmg_val = int(dmg_part.replace(",", "")) if dmg_part else 0
self.assertLess(
dps_val, 1_000_000,
f"DPS value {dps_val:,} exceeds bound (likely total_damage mislabeled). Line: {line}"
)
# Sanity: total damage should remain larger than DPS for sample data
self.assertGreater(dmg_val, dps_val, "Total damage should exceed DPS in run table")
def test_runs_analysis_labels_damage_and_dps_separately(self):
"""Ensure DPS and total damage columns remain distinct and correctly labeled.
Output Contract:
- RUNS response formatting (whitespace, separators, emoji markers) may change
- Must always include distinct DPS and Total Damage column headers
- Must include at least 1 data row per run
- DPS values must pass sanity check (< 1M for sample data, > 0)
- Total Damage must exceed DPS for sample data
This test is robust to cosmetic table changes while enforcing the core contract.
"""
packet = {
"meta": {"run_id": "run_200"},
"run_summary": {"rows": [["run_200", 200, 2_400_000, 120.0, 20_000, 48.0]]},
"runs_last_n": [
{"run_id": "run_200", "total_hits": 200, "total_damage": 2_400_000, "duration_seconds": 120.0, "dps": 20_000, "crit_rate_pct": 48.0},
{"run_id": "run_199", "total_hits": 210, "total_damage": 2_520_000, "duration_seconds": 125.0, "dps": 20_160, "crit_rate_pct": 46.0},
],
}
response, _ = self.coach.answer(
"lowest DPS run?",
payload={},
schema={},
query_callback=lambda x: {},
analysis_callback=lambda: packet,
)
lines = response.split("\n")
# Find header line containing both DPS and Total (case-insensitive)
header_line = None
header_idx = -1
for idx, line in enumerate(lines):
line_upper = line.upper()
# Look for line with "DPS" and "TOTAL" or "DMG"
if "DPS" in line_upper and ("TOTAL" in line_upper or "DMG" in line_upper):
header_line = line
header_idx = idx
break
self.assertIsNotNone(header_line, "Header should include DPS and Total Damage labels")
# Collect data rows after header (skip separator line with dashes)
data_rows = []
in_table = False
for i in range(header_idx + 1, len(lines)):
line = lines[i]
# Skip separator line with dashes
if "---" in line or "===" in line:
in_table = True
continue
# Stop at blank line or "Next Questions" or "Improvement"
if not line.strip() or "Next Questions" in line or "Improvement" in line:
break
# Collect lines with pipe separators (data rows)
if "|" in line and in_table:
# Must have at least 2 pipes (3 columns minimum)
if line.count("|") >= 2:
data_rows.append(line)
self.assertGreater(len(data_rows), 0, f"Expected at least one run row in table. Lines after header: {lines[header_idx+1:header_idx+10]}")
# Parse first data row
first_row = data_rows[0]
parts = [p.strip() for p in first_row.split("|") if p.strip()]
# Expect columns: Run #, Duration(s), DPS, Total Dmg, Crit%
self.assertGreaterEqual(len(parts), 4, f"Expected at least 4 columns, got {len(parts)}: {parts}")
# Find DPS and Total Damage columns by header
header_parts = [p.strip() for p in header_line.split("|") if p.strip()]
dps_col_idx = None
damage_col_idx = None
for idx, h in enumerate(header_parts):
h_upper = h.upper()
if "DPS" in h_upper and "TOTAL" not in h_upper:
dps_col_idx = idx
elif "TOTAL" in h_upper or ("DMG" in h_upper and "DPS" not in h_upper):
damage_col_idx = idx
self.assertIsNotNone(dps_col_idx, f"Could not find DPS column in header: {header_parts}")
self.assertIsNotNone(damage_col_idx, f"Could not find Total Damage column in header: {header_parts}")
self.assertNotEqual(dps_col_idx, damage_col_idx, "DPS and Total Damage must be separate columns")
# Extract values
dps_part = parts[dps_col_idx]
dmg_part = parts[damage_col_idx]
# Parse as numbers (handle commas and potential suffixes like "★ best")
dps_str = dps_part.replace(",", "").replace(" ", "").split("★")[0].split("…")[0]
dmg_str = dmg_part.replace(",", "").replace(" ", "").split("★")[0].split("…")[0]
dps_val = float(dps_str)
dmg_val = float(dmg_str)
# Contract assertions
self.assertLess(dps_val, 1_000_000, f"DPS should be reasonable (<1M), got {dps_val}")
self.assertNotEqual(dps_val, dmg_val, "DPS and total damage should not be identical")
self.assertGreater(dmg_val, dps_val,
f"Total damage ({dmg_val}) should be greater than DPS ({dps_val}) for sample data")
if __name__ == "__main__":
unittest.main()
