Spaces:
Running on Zero
Running on Zero
| import sys | |
| sys.path.insert(0, 'src') | |
| from sunnyware.memory import _summarize_old_events | |
| def _row(event_type, content): | |
| return {"event_type": event_type, "payload": {"content": content}} | |
| def test_summarize_empty(): | |
| assert _summarize_old_events([]) == "" | |
| def test_summarize_no_user_messages(): | |
| rows = [_row("assistant_message", "hello")] | |
| assert _summarize_old_events(rows) == "" | |
| def test_summarize_single_user(): | |
| rows = [_row("user_message", "what is AAPL price")] | |
| result = _summarize_old_events(rows) | |
| assert "AAPL price" in result | |
| assert result.startswith("Earlier in this conversation") | |
| def test_summarize_multiple_users(): | |
| rows = [ | |
| _row("user_message", "AAPL price"), | |
| _row("assistant_message", "$180"), | |
| _row("user_message", "MSFT price"), | |
| _row("assistant_message", "$400"), | |
| ] | |
| result = _summarize_old_events(rows) | |
| assert "AAPL" in result | |
| assert "MSFT" in result | |
| assert "; " in result | |
| def test_summarize_respects_max_chars(): | |
| rows = [_row("user_message", "question " + str(i) + " " + "x" * 80) for i in range(10)] | |
| result = _summarize_old_events(rows, max_chars=100) | |
| assert len(result) < 200 | |
| assert result.endswith("...") | |
| def test_summarize_caps_each_snippet(): | |
| rows = [_row("user_message", "x" * 200)] | |
| result = _summarize_old_events(rows) | |
| assert len(result) < 200 | |
| import asyncio | |
| def test_build_context_with_summary_no_pool(): | |
| from sunnyware.memory import build_context_with_summary | |
| result = asyncio.run(build_context_with_summary("", max_turns=4)) | |
| assert result == [] | |
| def test_should_skip_message_empty(): | |
| from sunnyware.memory import _should_skip_message | |
| assert _should_skip_message("") is True | |
| assert _should_skip_message(" ") is True | |
| def test_should_skip_message_short(): | |
| from sunnyware.memory import _should_skip_message | |
| assert _should_skip_message("ok") is True | |
| assert _should_skip_message("yes") is True | |
| assert _should_skip_message("no") is True | |
| def test_should_skip_message_low_value(): | |
| from sunnyware.memory import _should_skip_message | |
| assert _should_skip_message("thanks") is True | |
| assert _should_skip_message("got it") is True | |
| assert _should_skip_message("sure") is True | |
| def test_should_not_skip_substantive(): | |
| from sunnyware.memory import _should_skip_message | |
| assert _should_skip_message("what is AAPL price") is False | |
| assert _should_skip_message("AAPL price") is False | |
| assert _should_skip_message("MSFT earnings report") is False | |
| def test_pruning_threshold_env_override(): | |
| import os, importlib | |
| os.environ["SUNNYWARE_MIN_HISTORY_CHARS"] = "10" | |
| import sunnyware.memory as m | |
| importlib.reload(m) | |
| assert m.MIN_HISTORY_CONTENT_CHARS == 10 | |
| assert m._should_skip_message("short") is True | |
| assert m._should_skip_message("this is long enough") is False | |
| os.environ.pop("SUNNYWARE_MIN_HISTORY_CHARS", None) | |
| importlib.reload(m) | |
| assert m.MIN_HISTORY_CONTENT_CHARS == 4 | |
| def test_score_event_empty(): | |
| from sunnyware.memory import _score_event_text | |
| assert _score_event_text("", ["ai"]) == 0.0 | |
| def test_score_event_no_match(): | |
| from sunnyware.memory import _score_event_text | |
| assert _score_event_text("hello world", ["xyz"]) == 0.0 | |
| def test_score_event_single_match(): | |
| from sunnyware.memory import _score_event_text | |
| score = _score_event_text("what is AI", ["ai"]) | |
| assert score == 2.0 | |
| def test_score_event_multiple_matches(): | |
| from sunnyware.memory import _score_event_text | |
| score = _score_event_text("ai and ai", ["ai"]) | |
| assert score == 3.0 | |
| def test_score_longer_keyword_weighted(): | |
| from sunnyware.memory import _score_event_text | |
| score_short = _score_event_text("machine", ["ai"]) | |
| score_long = _score_event_text("machine", ["machine"]) | |
| assert score_long > score_short | |
| def test_score_two_keywords(): | |
| from sunnyware.memory import _score_event_text | |
| score = _score_event_text("ai and machine", ["ai", "machine"]) | |
| assert score == 9.0 | |
| def test_format_context_empty(): | |
| from sunnyware.memory import format_context | |
| assert format_context([]) == "" | |
| def test_format_context_single_event(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": "hello world"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[user] hello world" | |
| def test_format_context_assistant_role(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "assistant_message", "payload": {"content": "hi there"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[assistant] hi there" | |
| def test_format_context_string_payload(): | |
| from sunnyware.memory import format_context | |
| import json | |
| events = [ | |
| {"event_type": "user_message", "payload": json.dumps({"content": "from string"})}, | |
| ] | |
| result = format_context(events) | |
| assert "[user] from string" in result | |
| def test_format_context_multiple_events(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": "first"}}, | |
| {"event_type": "assistant_message", "payload": {"content": "second"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[user] first\n[assistant] second" | |
| def test_format_context_skips_empty_content(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": ""}}, | |
| {"event_type": "user_message", "payload": {"content": "real content"}}, | |
| ] | |
| result = format_context(events) | |
| assert "real content" in result | |
| assert result.count("[user]") == 1 | |
| def test_format_context_truncates_long_content(): | |
| from sunnyware.memory import format_context | |
| long_text = "x" * 500 | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": long_text}}, | |
| ] | |
| result = format_context(events, max_chars_per_event=50) | |
| assert len(result) < 100 | |
| def test_format_context_replaces_newlines(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": "line1\nline2"}}, | |
| ] | |
| result = format_context(events) | |
| assert "\n" not in result.replace("[user] ", "") | |
| def test_format_context_unknown_event_type(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "other_event", "payload": {"content": "misc"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[assistant] misc" | |
| def test_format_context_empty(): | |
| from sunnyware.memory import format_context | |
| assert format_context([]) == "" | |
| def test_format_context_single_event(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": "hello world"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[user] hello world" | |
| def test_format_context_assistant_role(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "assistant_message", "payload": {"content": "hi there"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[assistant] hi there" | |
| def test_format_context_string_payload(): | |
| from sunnyware.memory import format_context | |
| import json | |
| events = [ | |
| {"event_type": "user_message", "payload": json.dumps({"content": "from string"})}, | |
| ] | |
| result = format_context(events) | |
| assert "[user] from string" in result | |
| def test_format_context_multiple_events(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": "first"}}, | |
| {"event_type": "assistant_message", "payload": {"content": "second"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[user] first\n[assistant] second" | |
| def test_format_context_skips_empty_content(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": ""}}, | |
| {"event_type": "user_message", "payload": {"content": "real content"}}, | |
| ] | |
| result = format_context(events) | |
| assert "real content" in result | |
| assert result.count("[user]") == 1 | |
| def test_format_context_truncates_long_content(): | |
| from sunnyware.memory import format_context | |
| long_text = "x" * 500 | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": long_text}}, | |
| ] | |
| result = format_context(events, max_chars_per_event=50) | |
| assert len(result) < 100 | |
| def test_format_context_replaces_newlines(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "user_message", "payload": {"content": "line1\nline2"}}, | |
| ] | |
| result = format_context(events) | |
| assert "\n" not in result.replace("[user] ", "") | |
| def test_format_context_unknown_event_type(): | |
| from sunnyware.memory import format_context | |
| events = [ | |
| {"event_type": "other_event", "payload": {"content": "misc"}}, | |
| ] | |
| result = format_context(events) | |
| assert result == "[assistant] misc" | |
| def test_extract_content_dict(): | |
| from sunnyware.memory import _extract_content | |
| assert _extract_content({"content": "hello"}) == "hello" | |
| assert _extract_content({"content": ""}) == "" | |
| assert _extract_content({}) == "" | |
| assert _extract_content({"other": "x"}) == "" | |
| def test_extract_content_string(): | |
| from sunnyware.memory import _extract_content | |
| import json | |
| assert _extract_content(json.dumps({"content": "from string"})) == "from string" | |
| assert _extract_content('{"content": "nested"}') == "nested" | |
| def test_extract_content_invalid_string(): | |
| from sunnyware.memory import _extract_content | |
| assert _extract_content("not json") == "" | |
| assert _extract_content("") == "" | |
| def test_extract_content_none(): | |
| from sunnyware.memory import _extract_content | |
| assert _extract_content(None) == "" | |
| assert _extract_content(123) == "" | |
| def test_sanitize_content_empty(): | |
| from sunnyware.memory import _sanitize_content | |
| assert _sanitize_content("", 100) == "" | |
| assert _sanitize_content(None, 100) == "" | |
| def test_sanitize_content_short(): | |
| from sunnyware.memory import _sanitize_content | |
| result = _sanitize_content("hello world", 100) | |
| assert result == "hello world" | |
| def test_sanitize_content_truncates(): | |
| from sunnyware.memory import _sanitize_content | |
| result = _sanitize_content("x" * 200, 50) | |
| assert result.startswith("x" * 50) | |
| assert "[truncated, orig 200 chars]" in result | |
| def test_sanitize_content_redacts_base64(): | |
| from sunnyware.memory import _sanitize_content | |
| # Create a 600-char base64-ish string with uppercase, lowercase, and digit | |
| b64 = "A" * 200 + "a" * 200 + "1" * 200 | |
| result = _sanitize_content(b64, 5000) | |
| assert "[redacted:" in result | |
| assert len(result) < len(b64) | |
| def test_sanitize_content_does_not_redact_short_runs(): | |
| from sunnyware.memory import _sanitize_content | |
| text = "Hello World 123" | |
| result = _sanitize_content(text, 1000) | |
| assert "redacted" not in result | |
| assert result == text | |