import pathlib import sys import unittest ROOT = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from services.context_compressor import ( # noqa: E402 compact_history, extract_request_spec, merge_request_spec, render_request_spec, ) class ContextCompressionTests(unittest.TestCase): def test_extracts_stable_data_parameters(self): spec = extract_request_spec( "请下载 2002年8月1日至8月31日 西北太平洋 " "120°E–140°E、20°N–40°N 的 ERA5 v10、u10 数据," "深度0-100m,格式NetCDF" ) self.assertEqual(spec["dataset"], ["ERA5"]) self.assertEqual(sorted(spec["variables"]), ["u10", "v10"]) self.assertIn("120°E", spec["region"]) self.assertIn("40°N", spec["region"]) self.assertEqual(spec["depth"], "0-100m") self.assertEqual(spec["format"], "netcdf") self.assertIn("2002-08-01", spec["date"]) self.assertIn("2002-08-31", spec["date"]) def test_amendment_merges_without_repeating_full_history(self): spec = extract_request_spec( "下载 2002年8月1日至8月31日 ERA5 v10 " "120°E–140°E、20°N–40°N 深度0-100m 格式nc" ) spec = merge_request_spec( spec, "把格式改成 CSV,深度改为 5m", revision=True, ) text = render_request_spec(spec) self.assertIn("格式:csv", text) self.assertIn("深度:5m", text) self.assertIn("变量:v10", text) self.assertIn("2002-08-31", text) self.assertIn("区域:", text) def test_compact_history_is_smaller_and_keeps_parameters(self): original = ( "请下载 2002年8月1日至8月31日 西北太平洋 " "120°E–140°E、20°N–40°N 的 ERA5 v10、u10 数据," "深度0-100m,格式NetCDF。请按上述区域逐日导出," "生成可下载链接后返回。" * 3 ) messages = [ {"role": "user", "text": original}, {"role": "assistant", "text": "已完成,链接 https://x/download/a"}, {"role": "user", "text": "把格式改成 csv"}, {"role": "assistant", "text": "已更新 csv 下载链接。"}, ] compact = compact_history(messages) self.assertLess(len(compact), len(original)) self.assertIn("变量:u10、v10", compact) self.assertIn("2002-08-31", compact) self.assertIn("格式:csv", compact) if __name__ == "__main__": unittest.main()