From 76cb272eb2a8594c6a025d25bb504746ea741e43 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 07:02:37 +0800 Subject: [PATCH] refactor: isolate compact holdings parsing --- notifications/compact_adapter.py | 30 ++- .../test_compact_holdings_characterization.py | 241 ++++++++++++++++++ 2 files changed, 259 insertions(+), 12 deletions(-) create mode 100644 tests/test_compact_holdings_characterization.py diff --git a/notifications/compact_adapter.py b/notifications/compact_adapter.py index 73c4de92..dbf0b9a7 100644 --- a/notifications/compact_adapter.py +++ b/notifications/compact_adapter.py @@ -79,18 +79,8 @@ def label_holding_currencies( return "\n".join(output) -def adapt_compact_sections( - dashboard_text: str, - *, - locale: str, - supplemental_lines: Iterable[object] = (), -) -> tuple[str, ...]: - """Return normalized non-zero holdings followed by explicit supplements. - - Holdings are read from the already-rendered same-cycle dashboard so each - platform keeps ownership of broker-specific valuation and quantity rules. - Supplemental lines must already be localized by the platform translator. - """ +def _compact_holding_lines(dashboard_text: str, *, locale: str) -> list[str]: + """Read localized non-zero holdings from the rendered dashboard.""" holdings: list[str] = [] in_holdings = False for raw_line in str(dashboard_text or "").splitlines(): @@ -111,6 +101,22 @@ def adapt_compact_sections( continue detail = _localize_holding_detail(detail, locale=locale) holdings.append(f"- {symbol}: {detail}") + return holdings + + +def adapt_compact_sections( + dashboard_text: str, + *, + locale: str, + supplemental_lines: Iterable[object] = (), +) -> tuple[str, ...]: + """Return normalized non-zero holdings followed by explicit supplements. + + Holdings are read from the already-rendered same-cycle dashboard so each + platform keeps ownership of broker-specific valuation and quantity rules. + Supplemental lines must already be localized by the platform translator. + """ + holdings = _compact_holding_lines(dashboard_text, locale=locale) lines: list[str] = [] if holdings: diff --git a/tests/test_compact_holdings_characterization.py b/tests/test_compact_holdings_characterization.py new file mode 100644 index 00000000..e4f526d2 --- /dev/null +++ b/tests/test_compact_holdings_characterization.py @@ -0,0 +1,241 @@ +"""Characterize compact holdings through the public notification adapter.""" + +import unittest + +from notifications.compact_adapter import adapt_compact_sections + + +class CompactHoldingsCharacterizationTests(unittest.TestCase): + def test_each_supported_header_starts_holdings(self): + for header in ( + "πŸ’Ό ζŒδ»“", "πŸ’Ό η­–η•₯ζŒδ»“", "πŸ’Ό ε½“ε‰ζŒδ»“", "πŸ’Ό Holdings", + "πŸ’Ό Current Holdings", "πŸ’Ό Strategy Holdings", "πŸ’Ό Strategy holdings", + ): + with self.subTest(header=header): + self.assertEqual( + adapt_compact_sections(f" {header} \n- DEMO: 2 shares", locale="en"), + ("πŸ’Ό Holdings", "- DEMO: 2 shares"), + ) + + def test_unsupported_headers_do_not_start_holdings(self): + for header in ("", "πŸ’Ό holdings", "πŸ’Ό HOLDINGS", "Holdings", "πŸ’Ό Holdings:"): + with self.subTest(header=header): + self.assertEqual(adapt_compact_sections(f"{header}\nDEMO: 2 shares", locale="en"), ()) + + def test_section_terminators_stop_before_later_holdings(self): + for marker in ("━", "πŸ“Œ", "πŸ’΅", "πŸ“Š", "🎯", "🧾", "⏱", "🧩"): + with self.subTest(marker=marker): + dashboard = f"πŸ’Ό Holdings\nFIRST: 2 shares\n {marker} Next\nπŸ’Ό Holdings\nSECOND: 3 shares" + self.assertEqual( + adapt_compact_sections(dashboard, locale="en"), + ("πŸ’Ό Holdings", "- FIRST: 2 shares"), + ) + + def test_blank_and_repeated_headers_keep_the_current_section(self): + dashboard = "\n".join(( + "IGNORED: 1 share", "πŸ’Ό Holdings", "FIRST: 2 shares", "", " \t", + "unknown header", "πŸ’Ό η­–η•₯ζŒδ»“", "SECOND: 3 shares", "🎲 THIRD: 4 shares", + )) + self.assertEqual(adapt_compact_sections(dashboard, locale="en"), ( + "πŸ’Ό Holdings", "- FIRST: 2 shares", "- SECOND: 3 shares", "- 🎲 THIRD: 4 shares", + )) + + def test_symbols_and_separator_precedence_preserve_original_text(self): + for raw, rendered in ( + (" --β€’ MiXeD : 2 shares ", "- MiXeD: 2 shares"), + (" DEMO： 2 shares : note", "- DEMO: 2 shares : note"), + (" A:B： 2 shares", "- A:B: 2 shares"), + (" A:B: 2 shares", "- A: B: 2 shares"), + ("β€’- DEMO: 2θ‚‘", "- DEMO: 2 shares"), + ): + with self.subTest(raw=raw): + self.assertEqual( + adapt_compact_sections(f"πŸ’Ό Holdings\n{raw}", locale="en"), + ("πŸ’Ό Holdings", rendered), + ) + + def test_missing_or_zero_details_are_omitted(self): + for detail in ("", "0 shares", "-0.00", "+0,000.00", "0.000000000001", "-0.000000000001", "no amount", "NaN", "Infinity"): + with self.subTest(detail=detail): + self.assertEqual(adapt_compact_sections(f"πŸ’Ό Holdings\nDEMO: {detail}", locale="en"), ()) + self.assertEqual(adapt_compact_sections("πŸ’Ό Holdings\n: 2 shares\n：2θ‚‘\n2 shares", locale="en"), ()) + + def test_nonzero_detection_and_numeric_text_are_preserved(self): + for detail in ( + "0.0000000000011", "-0.0000000000011", "+1,000.25", "1,2,3", "0,0,0 / 2 shares", + "$0.00 / 2 shares", "$2.00 / 0 shares", "code 2026", "-2 shares", + ): + with self.subTest(detail=detail): + self.assertEqual( + adapt_compact_sections(f"πŸ’Ό Holdings\nDEMO: {detail}", locale="en"), + ("πŸ’Ό Holdings", f"- DEMO: {detail}"), + ) + + def test_localization_keeps_existing_quantity_and_locale_rules(self): + for locale, detail, heading, rendered in ( + ("en", "1θ‚‘", "πŸ’Ό Holdings", "1 share"), + ("en", "-1θ‚‘", "πŸ’Ό Holdings", "-1 share"), + ("en", "+1.0θ‚‘", "πŸ’Ό Holdings", "+1.0 share"), + ("en", "1,000θ‚‘", "πŸ’Ό Holdings", "1,000 shares"), + ("en", "0θ‚‘ / $2", "πŸ’Ό Holdings", "0 shares / $2"), + ("zh", "1 share / 2 SHARES", "πŸ’Ό ζŒδ»“", "1θ‚‘ / 2θ‚‘"), + ("ZH-Hant", "2 shares", "πŸ’Ό ζŒδ»“", "2θ‚‘"), + ("zh-extra", "2shares", "πŸ’Ό ζŒδ»“", "2shares"), + ("fr", "2θ‚‘", "πŸ’Ό Holdings", "2 shares"), + (None, "2θ‚‘", "πŸ’Ό Holdings", "2 shares"), + ): + with self.subTest(locale=locale, detail=detail): + self.assertEqual( + adapt_compact_sections(f"πŸ’Ό Holdings\nDEMO: {detail}", locale=locale), + (heading, f"- DEMO: {rendered}"), + ) + + def test_supplements_keep_order_and_deduplicate_against_rendered_holdings(self): + supplements = [None, False, 0, "", " \t", " πŸ’Ό Holdings ", " - DEMO: 2 shares ", " note ", "note", 7, "7"] + original = list(supplements) + self.assertEqual( + adapt_compact_sections("πŸ’Ό Holdings\nDEMO: 2 shares\nDEMO: 2 shares", locale="en", supplemental_lines=supplements), + ("πŸ’Ό Holdings", "- DEMO: 2 shares", "- DEMO: 2 shares", "note", "7"), + ) + self.assertEqual(supplements, original) + + def test_falsey_dashboards_skip_string_conversion_and_still_accept_supplements(self): + class FalseDashboard: + def __bool__(self): + return False + + def __str__(self): + raise AssertionError("falsey dashboard must not be stringified") + + for dashboard in (None, "", False, 0, FalseDashboard()): + with self.subTest(dashboard_type=type(dashboard).__name__): + self.assertEqual(adapt_compact_sections(dashboard, locale="en", supplemental_lines=("note",)), ("note",)) + + def test_dashboard_locale_and_supplement_evaluation_order(self): + events = [] + + class Dashboard: + def __bool__(self): + events.append("dashboard-bool") + return True + + def __str__(self): + events.append("dashboard-str") + return "πŸ’Ό Holdings\nFIRST: 1θ‚‘\nSECOND: 2θ‚‘" + + class Locale: + def __str__(self): + events.append("locale-str") + return "en" + + class Supplement: + def __bool__(self): + events.append("supplement-bool") + return True + + def __str__(self): + events.append("supplement-str") + return " note " + + def supplements(): + events.append("supplements-start") + yield Supplement() + events.append("supplements-end") + + self.assertEqual(adapt_compact_sections(Dashboard(), locale=Locale(), supplemental_lines=supplements()), ( + "πŸ’Ό Holdings", "- FIRST: 1 share", "- SECOND: 2 shares", "note", + )) + self.assertEqual(events, [ + "dashboard-bool", "dashboard-str", "locale-str", "locale-str", "locale-str", + "supplements-start", "supplement-bool", "supplement-str", "supplements-end", + ]) + + def test_locale_is_not_evaluated_without_nonzero_holdings(self): + class Locale: + def __str__(self): + raise AssertionError("unused locale must not be evaluated") + + self.assertEqual(adapt_compact_sections("πŸ’Ό Holdings\nDEMO: 0θ‚‘", locale=Locale(), supplemental_lines=("note",)), ("note",)) + + def test_dashboard_failure_preserves_exception_identity_and_stops_later_inputs(self): + for stage in ("bool", "str"): + with self.subTest(stage=stage): + error = ValueError("synthetic dashboard failure") + events = [] + + class Dashboard: + def __bool__(self): + events.append("bool") + if stage == "bool": + raise error + return True + + def __str__(self): + events.append("str") + raise error + + def supplements(): + events.append("supplements") + yield "note" + + with self.assertRaises(ValueError) as observed: + adapt_compact_sections(Dashboard(), locale="en", supplemental_lines=supplements()) + self.assertIs(observed.exception, error) + self.assertEqual(events, ["bool"] if stage == "bool" else ["bool", "str"]) + + def test_locale_failure_precedes_supplement_consumption(self): + for failure_call in (1, 2, 3): + with self.subTest(failure_call=failure_call): + error = RuntimeError("synthetic locale failure") + events = [] + + class Locale: + def __str__(self): + events.append("locale") + if len(events) == failure_call: + raise error + return "en" + + def supplements(): + events.append("supplements") + yield "note" + + with self.assertRaises(RuntimeError) as observed: + adapt_compact_sections("πŸ’Ό Holdings\nA: 1θ‚‘\nB: 2θ‚‘", locale=Locale(), supplemental_lines=supplements()) + self.assertIs(observed.exception, error) + self.assertEqual(events, ["locale"] * failure_call) + + def test_supplement_iteration_failure_is_propagated(self): + error = RuntimeError("synthetic iteration failure") + + def supplements(): + yield "note" + raise error + + with self.assertRaises(RuntimeError) as observed: + adapt_compact_sections("πŸ’Ό Holdings\nDEMO: 2 shares", locale="en", supplemental_lines=supplements()) + self.assertIs(observed.exception, error) + with self.assertRaises(TypeError): + adapt_compact_sections("", locale="en", supplemental_lines=None) + + def test_supplement_conversion_failure_is_propagated(self): + for stage in ("bool", "str"): + with self.subTest(stage=stage): + error = ValueError("synthetic supplement failure") + + class Supplement: + def __bool__(self): + if stage == "bool": + raise error + return True + + def __str__(self): + raise error + + with self.assertRaises(ValueError) as observed: + adapt_compact_sections("", locale="en", supplemental_lines=(Supplement(),)) + self.assertIs(observed.exception, error) + + +if __name__ == "__main__": + unittest.main()