hansjone/oclaw
1
0
Fork 0
mirror of https://github.com/hansjone/oclaw.git synced 2026-10-09 07:13:17 +08:00
Code Issues Projects Releases Packages Wiki Activity Actions
oclaw/tests/test_group_ingest.py
oliver 5336f8835f fix(whatsapp): pass quoted UME alarm to agent when @ bot 
Extract quotedText from reply context correctly and enrich inbound only after mention detection, keeping strict @-required group policy.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-23 14:36:14 +08:00

591 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
import pytest
from runtime.orchestration.group_ingest import (
GROUP_SESSION_USER_SENTINEL,
build_group_sender_context,
build_whatsapp_group_reply_metadata,
infer_is_group_from_chat_id,
is_nonsend_channel_reply_text,
normalize_jid,
resolve_group_policy,
session_user_key,
should_process_group_inbound,
text_mentions_bot,
)
from interfaces.channels.base import InboundMessage
from runtime.application.gateway.inbound_service import _parse_generic_inbound, process_inbound_payload
from svc.persistence.db.engine import clear_assistant_engine_cache
from svc.persistence.sqlite_store import SqliteStore
@pytest.fixture
def fresh_sqlite_store(monkeypatch: pytest.MonkeyPatch, tmp_path):
monkeypatch.delenv("AIA_ASSISTANT_DB_BACKEND", raising=False)
monkeypatch.setenv("OPS_WORKSPACE_ROOT", str(tmp_path))
dbfile = tmp_path / "group_ingest.sqlite"
monkeypatch.setenv("AIA_ASSISTANT_DB_PATH", str(dbfile))
clear_assistant_engine_cache()
s = SqliteStore(str(dbfile))
try:
yield s
finally:
clear_assistant_engine_cache()
def test_normalize_jid_strips_device_suffix() -> None:
assert normalize_jid("123456:12@s.whatsapp.net") == "123456@s.whatsapp.net"
def test_should_process_direct_messages_always() -> None:
assert should_process_group_inbound(is_group=False, text="hi", mentions=[], bot_jid="bot@s.whatsapp.net") is True
def test_should_drop_group_without_mention_or_trigger() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="大家好",
mentions=[],
bot_jid="999@s.whatsapp.net",
require_mention=True,
triggers=["/oclaw"],
)
is False
)
def test_should_accept_group_when_bot_mentioned() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@bot hello",
mentions=["999:0@s.whatsapp.net"],
bot_jid="999@s.whatsapp.net",
require_mention=True,
)
is True
)
def test_should_accept_group_mention_with_lid_phone_match() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@bot hello",
mentions=["6281284654304@lid"],
bot_jid="6281284654304@s.whatsapp.net",
require_mention=True,
)
is True
)
def test_should_accept_group_mention_with_bot_lid_identity() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@bot hello",
mentions=["176944565977182@lid"],
bot_jid="6281284654304@s.whatsapp.net",
require_mention=True,
metadata={"bot_lid": "176944565977182:2@lid"},
)
is True
)
def test_jids_same_user_lid_device_suffix() -> None:
from runtime.orchestration.group_ingest import jids_same_user
assert jids_same_user("176944565977182@lid", "176944565977182:2@lid") is True
def test_should_accept_group_when_sidecar_reports_mentions_bot_without_jids() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@bot hi",
mentions=[],
bot_jid="999@s.whatsapp.net",
require_mention=True,
metadata={"mentions_bot": True, "raw": {"mentionsBot": True}},
)
is True
)
def test_should_reject_group_when_multiple_others_mentioned_without_bot() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@alice @bob 开会",
mentions=[
"111111111111@lid",
"222222222222@lid",
],
bot_jid="6281284654304@s.whatsapp.net",
require_mention=True,
metadata={"bot_lid": "176944565977182:2@lid"},
)
is False
)
def test_should_accept_group_when_bot_among_multiple_mentions() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@alice @bot 帮忙",
mentions=[
"111111111111@lid",
"176944565977182@lid",
],
bot_jid="6281284654304@s.whatsapp.net",
require_mention=True,
metadata={"bot_lid": "176944565977182:2@lid"},
)
is True
)
def test_should_reject_group_when_only_other_user_mentioned() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@alice hello",
mentions=["333465375410398@lid"],
bot_jid="6281284654304@s.whatsapp.net",
require_mention=True,
metadata={
"mentions_bot": True,
"bot_lid": "176944565977182:2@lid",
"raw": {"mentionsBot": True, "isReplyToBot": True},
},
)
is False
)
def test_should_reject_group_when_other_mentioned_even_if_reply_to_bot() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="@alice follow up",
mentions=["333465375410398@lid"],
bot_jid="6281284654304@s.whatsapp.net",
require_mention=True,
metadata={
"bot_lid": "176944565977182:2@lid",
"raw": {
"quotedParticipant": "6281284654304@s.whatsapp.net",
"isReplyToBot": True,
},
},
)
is False
)
def test_should_accept_group_reply_to_bot() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="follow up",
mentions=[],
bot_jid="999@s.whatsapp.net",
require_mention=True,
metadata={"raw": {"quotedParticipant": "999:0@s.whatsapp.net", "isReplyToBot": True}},
)
is True
)
def test_text_mentions_bot_phone_fallback() -> None:
assert text_mentions_bot(text="@1234567890 hello", bot_jid="1234567890@s.whatsapp.net") is True
assert text_mentions_bot(text="hello", bot_jid="1234567890@s.whatsapp.net") is False
def test_should_accept_group_trigger_without_mention() -> None:
assert (
should_process_group_inbound(
is_group=True,
text="/oclaw 查天气",
mentions=[],
bot_jid="999@s.whatsapp.net",
require_mention=True,
triggers=["/oclaw"],
)
is True
)
def test_enrich_quoted_ume_alarm_when_mentioned() -> None:
from runtime.orchestration.group_ingest import (
enrich_alert_group_question,
extract_quoted_ume_alert_text,
)
meta = {"raw": {"quotedText": "[UME Alarm Raised]\nDevice: NE1", "mentionsBot": True}}
assert extract_quoted_ume_alert_text(metadata=meta).startswith("[UME")
enriched = enrich_alert_group_question(
user_text="what happened?",
quoted_alert="[UME Alarm Raised]\nDevice: NE1",
)
assert "[Quoted UME alarm]" in enriched
assert "what happened?" in enriched
def test_session_user_key_group_sentinel() -> None:
assert session_user_key(is_group=True, external_user_id="111@s.whatsapp.net") == GROUP_SESSION_USER_SENTINEL
assert session_user_key(is_group=False, external_user_id="111@s.whatsapp.net") == "111@s.whatsapp.net"
def test_resolve_group_policy_from_account_config() -> None:
policy = resolve_group_policy(
account={
"config": {
"group_policy": {
"require_mention": False,
"triggers": ["!ask"],
}
}
}
)
assert policy.require_mention is False
assert policy.triggers == ("!ask",)
def test_infer_is_group_from_chat_id() -> None:
assert infer_is_group_from_chat_id("120363012345678@g.us") is True
assert infer_is_group_from_chat_id("111@s.whatsapp.net") is False
def test_parse_generic_inbound_infers_whatsapp_group() -> None:
inbound = _parse_generic_inbound(
"whatsapp",
{
"user_id": "111@s.whatsapp.net",
"chat_id": "120363012345678@g.us",
"text": "hello",
},
)
assert inbound.is_group is True
def test_is_nonsend_channel_reply_text() -> None:
assert is_nonsend_channel_reply_text("") is True
assert is_nonsend_channel_reply_text("(silent)") is True
assert is_nonsend_channel_reply_text("（silent）") is True
assert is_nonsend_channel_reply_text("NO_REPLY") is True
assert is_nonsend_channel_reply_text("你好") is False
def test_inbound_group_without_is_group_flag_is_still_silent(
monkeypatch: pytest.MonkeyPatch, fresh_sqlite_store: SqliteStore
) -> None:
store = fresh_sqlite_store
_setup_whatsapp_identity(store)
monkeypatch.setattr("svc.persistence.assistant_store.get_assistant_store", lambda: store)
out = process_inbound_payload(
{
"channel": "whatsapp",
"account_id": "wa-default",
"user_id": "111@s.whatsapp.net",
"chat_id": "120363012345678@g.us",
"text": "Test alarm",
"metadata": {"bot_jid": "999@s.whatsapp.net", "source": "test"},
}
)
assert out.get("replies") == []
def test_inbound_suppresses_silent_llm_reply(
monkeypatch: pytest.MonkeyPatch, fresh_sqlite_store: SqliteStore
) -> None:
store = fresh_sqlite_store
_setup_whatsapp_identity(store)
monkeypatch.setattr("svc.persistence.assistant_store.get_assistant_store", lambda: store)
class _Turn:
turn_uuid = "turn-s"
reply_text = "(silent)"
class _Gw:
def __init__(self, *, store: object) -> None:
_ = store
def handle_turn(self, **kwargs: object) -> _Turn:
return _Turn()
monkeypatch.setattr("runtime.gateway.OclawGateway", _Gw)
out = process_inbound_payload(
{
"channel": "whatsapp",
"account_id": "wa-default",
"user_id": "111@s.whatsapp.net",
"chat_id": "111@s.whatsapp.net",
"text": "ping",
"is_group": False,
"metadata": {"bot_jid": "999@s.whatsapp.net"},
}
)
assert out.get("replies") == []
def test_build_group_sender_context() -> None:
ctx = build_group_sender_context(
metadata={"raw": {"pushName": "Alice"}},
external_user_id="111@s.whatsapp.net",
)
assert "Alice" in ctx
assert "111@s.whatsapp.net" in ctx
def test_build_whatsapp_group_reply_metadata() -> None:
inbound = InboundMessage(
channel="whatsapp",
external_user_id="111:12@s.whatsapp.net",
external_chat_id="120363012345678@g.us",
text="明天几点？",
is_group=True,
metadata={
"raw": {
"id": "MSG123",
"participant": "111:12@s.whatsapp.net",
"pushName": "Alice",
}
},
)
meta = build_whatsapp_group_reply_metadata(inbound=inbound)
assert meta["quote_stanza_id"] == "MSG123"
assert meta["mention_jids"] == ["111:12@s.whatsapp.net"]
assert meta["quote_text"] == "明天几点？"
assert meta["quote_participant"] == "111:12@s.whatsapp.net"
def test_shared_group_session_for_multiple_senders(fresh_sqlite_store: SqliteStore) -> None:
store = fresh_sqlite_store
tenant = store.create_tenant("WA")
chat_id = "120363012345678@g.us"
sid_a = store.get_or_create_channel_session_v2(
tenant_id=str(tenant["id"]),
channel="whatsapp",
account_id="wa-default",
external_chat_id=chat_id,
external_user_id=session_user_key(is_group=True, external_user_id="111@s.whatsapp.net"),
session_title="whatsapp|test+Family",
)
sid_b = store.get_or_create_channel_session_v2(
tenant_id=str(tenant["id"]),
channel="whatsapp",
account_id="wa-default",
external_chat_id=chat_id,
external_user_id=session_user_key(is_group=True, external_user_id="222@s.whatsapp.net"),
session_title="whatsapp|test+Family",
)
assert sid_a == sid_b
def _setup_whatsapp_identity(store: SqliteStore, *, extra_user_ids: list[str] | None = None) -> tuple[str, str]:
tenant = store.create_tenant("WA")
tenant_id = str(tenant["id"])
user = store.create_user(tenant_id=tenant_id, display_name="Admin", role="owner")
user_id = str(user["id"])
store.upsert_user_channel_account(
tenant_id=tenant_id,
user_id=user_id,
channel="whatsapp",
account_id="wa-default",
name="wa-default",
config={"group_policy": {"require_mention": True, "triggers": ["/oclaw"]}},
is_active=True,
)
for ext_uid in ["111@s.whatsapp.net", *(extra_user_ids or [])]:
store.upsert_channel_identity_v2(
tenant_id=tenant_id,
channel="whatsapp",
account_id="wa-default",
external_user_id=ext_uid,
user_id=user_id,
)
return tenant_id, user_id
def test_inbound_group_without_mention_is_silent(monkeypatch: pytest.MonkeyPatch, fresh_sqlite_store: SqliteStore) -> None:
store = fresh_sqlite_store
_setup_whatsapp_identity(store)
monkeypatch.setattr("svc.persistence.assistant_store.get_assistant_store", lambda: store)
out = process_inbound_payload(
{
"channel": "whatsapp",
"account_id": "wa-default",
"user_id": "111@s.whatsapp.net",
"chat_id": "120363012345678@g.us",
"text": "大家晚上好",
"is_group": True,
"mentions": [],
"metadata": {"bot_jid": "999@s.whatsapp.net", "source": "test"},
}
)
assert out.get("ok") is True
assert out.get("replies") == []
def test_inbound_dm_still_processes_without_mention(monkeypatch: pytest.MonkeyPatch, fresh_sqlite_store: SqliteStore) -> None:
store = fresh_sqlite_store
_setup_whatsapp_identity(store)
monkeypatch.setattr("svc.persistence.assistant_store.get_assistant_store", lambda: store)
captured: dict[str, str] = {}
class _Turn:
turn_uuid = "turn-1"
reply_text = "pong"
class _Gw:
def __init__(self, *, store: object) -> None:
_ = store
def handle_turn(self, **kwargs: object) -> _Turn:
msg = kwargs.get("msg")
captured["text"] = str(getattr(msg, "text", "") or "")
return _Turn()
monkeypatch.setattr("runtime.gateway.OclawGateway", _Gw)
out = process_inbound_payload(
{
"channel": "whatsapp",
"account_id": "wa-default",
"user_id": "111@s.whatsapp.net",
"chat_id": "111@s.whatsapp.net",
"text": "ping",
"is_group": False,
"mentions": [],
"metadata": {"bot_jid": "999@s.whatsapp.net"},
}
)
replies = out.get("replies") if isinstance(out.get("replies"), list) else []
assert replies and str(replies[0].get("text") or "") == "pong"
assert "[群成员:" not in captured.get("text", "")
def test_inbound_group_mention_uses_shared_session_and_sender_prefix(
monkeypatch: pytest.MonkeyPatch, fresh_sqlite_store: SqliteStore
) -> None:
store = fresh_sqlite_store
tenant_id, user_id = _setup_whatsapp_identity(store, extra_user_ids=["222@s.whatsapp.net"])
monkeypatch.setattr("svc.persistence.assistant_store.get_assistant_store", lambda: store)
captured: dict[str, str] = {}
session_ids: list[str] = []
class _Turn:
turn_uuid = "turn-g"
reply_text = "group-ok"
class _Gw:
def __init__(self, *, store: object) -> None:
_ = store
def handle_turn(self, **kwargs: object) -> _Turn:
msg = kwargs.get("msg")
captured["text"] = str(getattr(msg, "text", "") or "")
session_ids.append(str(getattr(msg, "session_id", "") or ""))
return _Turn()
monkeypatch.setattr("runtime.gateway.OclawGateway", _Gw)
chat_id = "120363012345678@g.us"
base = {
"channel": "whatsapp",
"account_id": "wa-default",
"chat_id": chat_id,
"is_group": True,
"metadata": {"bot_jid": "999@s.whatsapp.net", "source": "test"},
}
process_inbound_payload(
{
**base,
"user_id": "111@s.whatsapp.net",
"text": "@bot hi",
"mentions": ["999@s.whatsapp.net"],
"metadata": {**base["metadata"], "raw": {"pushName": "Alice"}},
}
)
process_inbound_payload(
{
**base,
"user_id": "222@s.whatsapp.net",
"text": "@bot again",
"mentions": ["999@s.whatsapp.net"],
"metadata": {**base["metadata"], "raw": {"pushName": "Bob"}},
}
)
assert len(session_ids) == 2
assert session_ids[0] == session_ids[1]
assert "[群成员:" in captured["text"]
assert "Bob" in captured["text"]
sid = store.get_or_create_channel_session_v2(
tenant_id=tenant_id,
channel="whatsapp",
account_id="wa-default",
external_chat_id=chat_id,
external_user_id=GROUP_SESSION_USER_SENTINEL,
session_title="whatsapp|wa-default+Family",
)
assert sid == session_ids[0]
_ = tenant_id, user_id
def test_inbound_group_reply_includes_quote_and_mention_metadata(
monkeypatch: pytest.MonkeyPatch, fresh_sqlite_store: SqliteStore
) -> None:
store = fresh_sqlite_store
_setup_whatsapp_identity(store)
monkeypatch.setattr("svc.persistence.assistant_store.get_assistant_store", lambda: store)
class _Turn:
turn_uuid = "turn-g"
reply_text = "下午三点"
class _Gw:
def __init__(self, *, store: object) -> None:
_ = store
def handle_turn(self, **kwargs: object) -> _Turn:
return _Turn()
monkeypatch.setattr("runtime.gateway.OclawGateway", _Gw)
out = process_inbound_payload(
{
"channel": "whatsapp",
"account_id": "wa-default",
"user_id": "111@s.whatsapp.net",
"chat_id": "120363012345678@g.us",
"text": "@bot 明天几点？",
"is_group": True,
"mentions": ["999@s.whatsapp.net"],
"metadata": {
"bot_jid": "999@s.whatsapp.net",
"raw": {"id": "ABC123", "participant": "111@s.whatsapp.net", "pushName": "Alice"},
},
}
)
replies = out.get("replies") if isinstance(out.get("replies"), list) else []
assert replies
meta = replies[0].get("metadata") if isinstance(replies[0], dict) else {}
assert isinstance(meta, dict)
assert meta.get("quote_stanza_id") == "ABC123"
assert meta.get("mention_jids") == ["111@s.whatsapp.net"]
assert meta.get("quote_text") == "@bot 明天几点？"
Reference in a new issue View git blame Copy permalink