"""Unit tests for topic-analysis markdown table column normalisation.""" import unittest import pandas as pd from tools.llm_api_call import ( TOPIC_TABLE_EXPECTED_COLS, _ensure_standard_topic_table_columns, _four_column_table_has_sentiment, reconstruct_markdown_table_from_reference_df, ) class TestEnsureStandardTopicTableColumns(unittest.TestCase): def test_six_column_table_with_name_collision_is_unique(self): """Positional rename used to create duplicate 'General topic' labels.""" df = pd.DataFrame( { "Topic": ["Housing"], "Detail": ["Repairs delay"], "General topic": ["Services"], "Response ID": ["1"], "Summary": ["Delayed repairs"], "Extra": ["noise"], } ) out = _ensure_standard_topic_table_columns(df, batch_size_number=1) self.assertEqual(list(out.columns), TOPIC_TABLE_EXPECTED_COLS) self.assertFalse(out.columns.duplicated().any()) self.assertEqual(out.loc[0, "General topic"], "Services") self.assertEqual(out.loc[0, "Response ID"], "1") # Concat must not raise the historical reindexing error other = pd.DataFrame( { "General topic": ["Other"], "Subtopic": ["X"], "Sentiment": ["Not assessed"], } ) combined = pd.concat( [out[["General topic", "Subtopic", "Sentiment"]], other], ignore_index=True, ) self.assertEqual(len(combined), 2) def test_float_nan_general_topic_becomes_string(self): df = pd.DataFrame( { "General topic": [float("nan"), float("nan")], "Subtopic": ["A", "B"], "Sentiment": ["Neutral", "Positive"], "Response ID": [1.0, 2.0], "Summary": ["s1", "s2"], } ) out = _ensure_standard_topic_table_columns(df, batch_size_number=5) self.assertEqual(out["General topic"].dtype, object) # .str accessor must work after normalisation stripped = out["General topic"].str.strip() self.assertTrue((stripped == "").all()) self.assertEqual(out.loc[0, "Response ID"], "1.0") def test_empty_named_general_topic_prefers_positional_values(self): """Blank 'General topic' header should not block real topic text in col0.""" df = pd.DataFrame( { "Topic": ["Housing"], "Subtopic": ["Repairs"], "General topic": [float("nan")], "Response ID": ["1"], "Summary": ["Delayed repairs"], "Extra": ["noise"], } ) out = _ensure_standard_topic_table_columns(df, batch_size_number=5) self.assertEqual(out.loc[0, "General topic"], "Housing") self.assertEqual(out.loc[0, "Subtopic"], "Repairs") self.assertFalse(out.columns.duplicated().any()) def test_three_column_table_gets_defaults(self): df = pd.DataFrame( { "General topic": ["Transport"], "Subtopic": ["Buses"], "Summary": ["Need more buses"], } ) out = _ensure_standard_topic_table_columns(df, batch_size_number=1) self.assertEqual(list(out.columns), TOPIC_TABLE_EXPECTED_COLS) self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") self.assertEqual(out.loc[0, "Response ID"], "1") self.assertEqual(out.loc[0, "General topic"], "Transport") def test_empty_dataframe(self): out = _ensure_standard_topic_table_columns(pd.DataFrame(), batch_size_number=5) self.assertTrue(out.empty) self.assertEqual(list(out.columns), TOPIC_TABLE_EXPECTED_COLS) def test_duplicate_input_columns_are_deduped(self): df = pd.DataFrame( [["Housing", "Repairs", "Negative", "1", "summary"]], columns=[ "General topic", "Subtopic", "Sentiment", "Response ID", "Summary", ], ) # Force duplicate labels the way a bad select used to df = pd.concat([df, df[["General topic"]]], axis=1) out = _ensure_standard_topic_table_columns(df, batch_size_number=5) self.assertEqual(list(out.columns), TOPIC_TABLE_EXPECTED_COLS) self.assertFalse(out.columns.duplicated().any()) self.assertEqual(out.loc[0, "General topic"], "Housing") class TestNoSentimentTableLayouts(unittest.TestCase): def test_named_four_column_preserves_response_ids(self): df = pd.DataFrame( { "General topic": ["Housing"], "Subtopic": ["Repairs"], "Response ID": ["3, 4"], "Summary": ["Delayed repairs"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False ) self.assertEqual(out.loc[0, "Response ID"], "3, 4") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_unnamed_four_column_maps_third_column_as_response_id(self): df = pd.DataFrame([["Housing", "Repairs", "3, 4", "Delayed repairs"]]) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False ) self.assertEqual(out.loc[0, "Response ID"], "3, 4") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_five_column_trailing_extra_does_not_steal_response_id(self): df = pd.DataFrame( { "General topic": ["Housing"], "Subtopic": ["Repairs"], "Response ID": ["3, 4"], "Summary": ["Delayed repairs"], "Extra": [""], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False ) self.assertEqual(out.loc[0, "Response ID"], "3, 4") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_unnamed_five_column_trailing_extra_without_sentiment(self): df = pd.DataFrame([["Housing", "Repairs", "3, 4", "Delayed repairs", ""]]) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False ) self.assertEqual(out.loc[0, "Response ID"], "3, 4") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_three_column_summary_is_not_used_as_response_id(self): df = pd.DataFrame( { "General topic": ["Transport"], "Subtopic": ["Buses"], "Summary": ["Need more buses"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False ) self.assertEqual(out.loc[0, "Summary"], "Need more buses") self.assertEqual(out.loc[0, "Response ID"], "") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_sentiment_layout_still_maps_when_assessing_sentiment(self): df = pd.DataFrame([["Housing", "Repairs", "Negative", "3", "summary"]]) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=True ) self.assertEqual(out.loc[0, "Sentiment"], "Negative") self.assertEqual(out.loc[0, "Response ID"], "3") self.assertEqual(out.loc[0, "Summary"], "summary") def test_three_column_placeholder_table_maps_ids_not_sentiment(self): """Force-zero-shot tables often have Placeholder, Subtopic, Response ID.""" df = pd.DataFrame( { "Placeholder": ["Not assessed", "Not assessed"], "Subtopics": ["Housing repairs", "Buses"], "Response refs": ["1, 2", "3"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=True ) self.assertEqual(out.loc[0, "Response ID"], "1, 2") self.assertEqual(out.loc[1, "Response ID"], "3") self.assertEqual(out.loc[0, "Subtopic"], "Housing repairs") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_four_column_with_response_id_header_is_not_sentiment(self): df = pd.DataFrame( { "General topic": ["Housing"], "Subtopic": ["Repairs"], "Response ID": ["3, 4"], "Summary": ["Delayed repairs"], } ) self.assertFalse(_four_column_table_has_sentiment(df, assess_sentiment=False)) def test_four_column_numeric_ids_are_not_treated_as_sentiment(self): df = pd.DataFrame([["Housing", "Repairs", "1, 2", "Delayed repairs"]]) self.assertFalse(_four_column_table_has_sentiment(df, assess_sentiment=False)) def test_reconstruct_omits_sentiment_column_when_not_assessed(self): reference_df = pd.DataFrame( { "General topic": ["Housing", "Housing"], "Subtopic": ["Repairs", "Repairs"], "Sentiment": ["Not assessed", "Not assessed"], "Response ID": [1, 2], "Summary": ["Delayed repairs", "Delayed repairs"], } ) markdown, out_df = reconstruct_markdown_table_from_reference_df( reference_df, sentiment_checkbox="Do not assess sentiment" ) self.assertIn("| General topic | Subtopic | Response ID | Summary |", markdown) self.assertNotIn("| Sentiment |", markdown) self.assertEqual(out_df.loc[0, "Sentiment"], "Not assessed") self.assertIn("1", out_df.loc[0, "Response ID"]) self.assertIn("2", out_df.loc[0, "Response ID"]) class TestTopicConfidenceColumn(unittest.TestCase): def test_named_six_column_table_maps_confidence(self): df = pd.DataFrame( { "General topic": ["Housing"], "Subtopic": ["Repairs"], "Sentiment": ["Negative"], "Response ID": ["1, 2"], "Confidence": ["0.82"], "Summary": ["Delayed repairs"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=True, include_confidence=True, ) self.assertIn("Confidence", out.columns) self.assertEqual(out.loc[0, "Confidence"], "0.82") self.assertEqual(out.loc[0, "Response ID"], "1, 2") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") def test_unnamed_six_column_maps_confidence_before_summary(self): df = pd.DataFrame( [["Housing", "Repairs", "Negative", "3", "0.4", "Delayed repairs"]] ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=True, include_confidence=True, ) self.assertEqual(out.loc[0, "Response ID"], "3") self.assertEqual(out.loc[0, "Confidence"], "0.4") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") def test_five_column_without_sentiment_maps_confidence(self): df = pd.DataFrame( { "General topic": ["Housing"], "Subtopic": ["Repairs"], "Response ID": ["3, 4"], "Confidence": ["80%"], "Summary": ["Delayed repairs"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False, include_confidence=True, ) self.assertEqual(out.loc[0, "Response ID"], "3, 4") self.assertEqual(out.loc[0, "Confidence"], "80%") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") self.assertEqual(out.loc[0, "Sentiment"], "Not assessed") def test_missing_confidence_does_not_steal_summary(self): df = pd.DataFrame( { "General topic": ["Transport"], "Subtopic": ["Buses"], "Sentiment": ["Negative"], "Response ID": ["1"], "Summary": ["Need more buses"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=True, include_confidence=True, ) self.assertEqual(out.loc[0, "Summary"], "Need more buses") self.assertEqual(out.loc[0, "Response ID"], "1") self.assertEqual(out.loc[0, "Confidence"], "") def test_response_ids_are_not_consumed_as_confidence(self): df = pd.DataFrame([["Housing", "Repairs", "1, 2", "Delayed repairs"]]) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False, include_confidence=True, ) self.assertEqual(out.loc[0, "Response ID"], "1, 2") self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") self.assertEqual(out.loc[0, "Confidence"], "") def test_without_flag_confidence_is_omitted(self): df = pd.DataFrame( { "General topic": ["Housing"], "Subtopic": ["Repairs"], "Sentiment": ["Negative"], "Response ID": ["1"], "Confidence": ["0.9"], "Summary": ["Delayed repairs"], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, include_confidence=False ) self.assertEqual(list(out.columns), TOPIC_TABLE_EXPECTED_COLS) self.assertNotIn("Confidence", out.columns) self.assertEqual(out.loc[0, "Summary"], "Delayed repairs") def test_reconstruct_includes_confidence_column(self): reference_df = pd.DataFrame( { "General topic": ["Housing", "Housing"], "Subtopic": ["Repairs", "Repairs"], "Sentiment": ["Negative", "Negative"], "Response ID": [1, 2], "Confidence": [0.9, 0.4], "Summary": ["Delayed repairs", "Delayed repairs"], } ) markdown, out_df = reconstruct_markdown_table_from_reference_df( reference_df, sentiment_checkbox="Negative or Positive", include_topic_confidence_radio="Yes", ) self.assertIn( "| General topic | Subtopic | Sentiment | Response ID | Confidence | Summary |", markdown, ) self.assertEqual(len(out_df), 2) self.assertIn("0.90", markdown) self.assertIn("0.40", markdown) def test_main_heading_alias_maps_to_general_topic(self): df = pd.DataFrame( { "Main heading": ["Behaviour at school"], "Subheading": ["Absences"], "Summary": ["Alex had increasing absences."], } ) out = _ensure_standard_topic_table_columns( df, batch_size_number=5, assess_sentiment=False ) self.assertEqual(out.loc[0, "General topic"], "Behaviour at school") self.assertEqual(out.loc[0, "Subtopic"], "Absences") self.assertEqual(out.loc[0, "Summary"], "Alex had increasing absences.") self.assertNotEqual(out.loc[0, "Sentiment"], "Alex had increasing absences.") if __name__ == "__main__": unittest.main()