feat: add Feishu document comment intelligent reply with 3-tier access control
- Full comment handler: parse drive.notice.comment_add_v1 events, build timeline, run agent, deliver reply with chunking support. - 5 tools: feishu_doc_read, feishu_drive_list_comments, feishu_drive_list_comment_replies, feishu_drive_reply_comment, feishu_drive_add_comment. - 3-tier access control rules (exact doc > wildcard "*" > top-level > defaults) with per-field fallback. Config via ~/.hermes/feishu_comment_rules.json, mtime-cached hot-reload. - Self-reply filter using generalized self_open_id (supports future user-identity subscriptions). Receiver check: only process events where the bot is the @mentioned target. - Smart timeline selection, long text chunking, semantic text extraction, session sharing per document, wiki link resolution. Change-Id: I31e82fd6355173dbcc400b8934b6d9799e3137b9
This commit is contained in:
liujinkun
261
tests/gateway/test_feishu_comment.py
Normal file
261
tests/gateway/test_feishu_comment.py
Normal file
@@ -0,0 +1,261 @@
|
||||
"""Tests for feishu_comment — event filtering, access control integration, wiki reverse lookup."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
from gateway.platforms.feishu_comment import (
|
||||
parse_drive_comment_event,
|
||||
_ALLOWED_NOTICE_TYPES,
|
||||
_sanitize_comment_text,
|
||||
)
|
||||
|
||||
|
||||
def _make_event(
|
||||
comment_id="c1",
|
||||
reply_id="r1",
|
||||
notice_type="add_reply",
|
||||
file_token="docx_token",
|
||||
file_type="docx",
|
||||
from_open_id="ou_user",
|
||||
to_open_id="ou_bot",
|
||||
is_mentioned=True,
|
||||
):
|
||||
"""Build a minimal drive comment event SimpleNamespace."""
|
||||
return SimpleNamespace(event={
|
||||
"event_id": "evt_1",
|
||||
"comment_id": comment_id,
|
||||
"reply_id": reply_id,
|
||||
"is_mentioned": is_mentioned,
|
||||
"timestamp": "1713200000",
|
||||
"notice_meta": {
|
||||
"file_token": file_token,
|
||||
"file_type": file_type,
|
||||
"notice_type": notice_type,
|
||||
"from_user_id": {"open_id": from_open_id},
|
||||
"to_user_id": {"open_id": to_open_id},
|
||||
},
|
||||
})
|
||||
|
||||
|
||||
class TestParseEvent(unittest.TestCase):
|
||||
def test_parse_valid_event(self):
|
||||
evt = _make_event()
|
||||
parsed = parse_drive_comment_event(evt)
|
||||
self.assertIsNotNone(parsed)
|
||||
self.assertEqual(parsed["comment_id"], "c1")
|
||||
self.assertEqual(parsed["file_type"], "docx")
|
||||
self.assertEqual(parsed["from_open_id"], "ou_user")
|
||||
self.assertEqual(parsed["to_open_id"], "ou_bot")
|
||||
|
||||
def test_parse_missing_event_attr(self):
|
||||
self.assertIsNone(parse_drive_comment_event(object()))
|
||||
|
||||
def test_parse_none_event(self):
|
||||
self.assertIsNone(parse_drive_comment_event(SimpleNamespace()))
|
||||
|
||||
|
||||
class TestEventFiltering(unittest.TestCase):
|
||||
"""Test the filtering logic in handle_drive_comment_event."""
|
||||
|
||||
def _run(self, coro):
|
||||
return asyncio.get_event_loop().run_until_complete(coro)
|
||||
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
|
||||
def test_self_reply_filtered(self, mock_allowed, mock_resolve, mock_load):
|
||||
"""Events where from_open_id == self_open_id should be dropped."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
|
||||
evt = _make_event(from_open_id="ou_bot", to_open_id="ou_bot")
|
||||
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
|
||||
mock_load.assert_not_called()
|
||||
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
|
||||
def test_wrong_receiver_filtered(self, mock_allowed, mock_resolve, mock_load):
|
||||
"""Events where to_open_id != self_open_id should be dropped."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
|
||||
evt = _make_event(to_open_id="ou_other_bot")
|
||||
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
|
||||
mock_load.assert_not_called()
|
||||
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
|
||||
def test_empty_to_open_id_filtered(self, mock_allowed, mock_resolve, mock_load):
|
||||
"""Events with empty to_open_id should be dropped."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
|
||||
evt = _make_event(to_open_id="")
|
||||
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
|
||||
mock_load.assert_not_called()
|
||||
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed")
|
||||
def test_invalid_notice_type_filtered(self, mock_allowed, mock_resolve, mock_load):
|
||||
"""Events with unsupported notice_type should be dropped."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
|
||||
evt = _make_event(notice_type="resolve_comment")
|
||||
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
|
||||
mock_load.assert_not_called()
|
||||
|
||||
def test_allowed_notice_types(self):
|
||||
self.assertIn("add_comment", _ALLOWED_NOTICE_TYPES)
|
||||
self.assertIn("add_reply", _ALLOWED_NOTICE_TYPES)
|
||||
self.assertNotIn("resolve_comment", _ALLOWED_NOTICE_TYPES)
|
||||
|
||||
|
||||
class TestAccessControlIntegration(unittest.TestCase):
|
||||
def _run(self, coro):
|
||||
return asyncio.get_event_loop().run_until_complete(coro)
|
||||
|
||||
@patch("gateway.platforms.feishu_comment_rules.has_wiki_keys", return_value=False)
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed", return_value=False)
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
def test_denied_user_no_side_effects(self, mock_load, mock_resolve, mock_allowed, mock_wiki_keys):
|
||||
"""Denied user should not trigger typing reaction or agent."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
from gateway.platforms.feishu_comment_rules import ResolvedCommentRule
|
||||
|
||||
mock_resolve.return_value = ResolvedCommentRule(True, "allowlist", frozenset(), "top")
|
||||
mock_load.return_value = Mock()
|
||||
|
||||
client = Mock()
|
||||
evt = _make_event()
|
||||
self._run(handle_drive_comment_event(client, evt, self_open_id="ou_bot"))
|
||||
|
||||
# No API calls should be made for denied users
|
||||
client.request.assert_not_called()
|
||||
|
||||
@patch("gateway.platforms.feishu_comment_rules.has_wiki_keys", return_value=False)
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed", return_value=False)
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
def test_disabled_comment_skipped(self, mock_load, mock_resolve, mock_allowed, mock_wiki_keys):
|
||||
"""Disabled comments should return immediately."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
from gateway.platforms.feishu_comment_rules import ResolvedCommentRule
|
||||
|
||||
mock_resolve.return_value = ResolvedCommentRule(False, "allowlist", frozenset(), "top")
|
||||
mock_load.return_value = Mock()
|
||||
|
||||
evt = _make_event()
|
||||
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
|
||||
mock_allowed.assert_not_called()
|
||||
|
||||
|
||||
class TestSanitizeCommentText(unittest.TestCase):
|
||||
def test_angle_brackets_escaped(self):
|
||||
self.assertEqual(_sanitize_comment_text("List<String>"), "List<String>")
|
||||
|
||||
def test_ampersand_escaped_first(self):
|
||||
self.assertEqual(_sanitize_comment_text("a & b"), "a & b")
|
||||
|
||||
def test_ampersand_not_double_escaped(self):
|
||||
result = _sanitize_comment_text("a < b & c > d")
|
||||
self.assertEqual(result, "a < b & c > d")
|
||||
self.assertNotIn("&lt;", result)
|
||||
self.assertNotIn("&gt;", result)
|
||||
|
||||
def test_plain_text_unchanged(self):
|
||||
self.assertEqual(_sanitize_comment_text("hello world"), "hello world")
|
||||
|
||||
def test_empty_string(self):
|
||||
self.assertEqual(_sanitize_comment_text(""), "")
|
||||
|
||||
def test_code_snippet(self):
|
||||
text = 'if (a < b && c > 0) { return "ok"; }'
|
||||
result = _sanitize_comment_text(text)
|
||||
self.assertNotIn("<", result)
|
||||
self.assertNotIn(">", result)
|
||||
self.assertIn("<", result)
|
||||
self.assertIn(">", result)
|
||||
|
||||
|
||||
class TestWikiReverseLookup(unittest.TestCase):
|
||||
def _run(self, coro):
|
||||
return asyncio.get_event_loop().run_until_complete(coro)
|
||||
|
||||
@patch("gateway.platforms.feishu_comment._exec_request")
|
||||
def test_reverse_lookup_success(self, mock_exec):
|
||||
from gateway.platforms.feishu_comment import _reverse_lookup_wiki_token
|
||||
|
||||
mock_exec.return_value = (0, "Success", {
|
||||
"node": {"node_token": "WIKI_TOKEN_123", "obj_token": "docx_abc"},
|
||||
})
|
||||
result = self._run(_reverse_lookup_wiki_token(Mock(), "docx", "docx_abc"))
|
||||
self.assertEqual(result, "WIKI_TOKEN_123")
|
||||
# Verify correct API params
|
||||
call_args = mock_exec.call_args
|
||||
queries = call_args[1].get("queries") or call_args[0][3]
|
||||
query_dict = dict(queries)
|
||||
self.assertEqual(query_dict["token"], "docx_abc")
|
||||
self.assertEqual(query_dict["obj_type"], "docx")
|
||||
|
||||
@patch("gateway.platforms.feishu_comment._exec_request")
|
||||
def test_reverse_lookup_not_wiki(self, mock_exec):
|
||||
from gateway.platforms.feishu_comment import _reverse_lookup_wiki_token
|
||||
|
||||
mock_exec.return_value = (131001, "not found", {})
|
||||
result = self._run(_reverse_lookup_wiki_token(Mock(), "docx", "docx_abc"))
|
||||
self.assertIsNone(result)
|
||||
|
||||
@patch("gateway.platforms.feishu_comment._exec_request")
|
||||
def test_reverse_lookup_service_error(self, mock_exec):
|
||||
from gateway.platforms.feishu_comment import _reverse_lookup_wiki_token
|
||||
|
||||
mock_exec.return_value = (500, "internal error", {})
|
||||
result = self._run(_reverse_lookup_wiki_token(Mock(), "docx", "docx_abc"))
|
||||
self.assertIsNone(result)
|
||||
|
||||
@patch("gateway.platforms.feishu_comment._reverse_lookup_wiki_token", new_callable=AsyncMock)
|
||||
@patch("gateway.platforms.feishu_comment_rules.has_wiki_keys", return_value=True)
|
||||
@patch("gateway.platforms.feishu_comment_rules.is_user_allowed", return_value=True)
|
||||
@patch("gateway.platforms.feishu_comment_rules.resolve_rule")
|
||||
@patch("gateway.platforms.feishu_comment_rules.load_config")
|
||||
@patch("gateway.platforms.feishu_comment.add_comment_reaction", new_callable=AsyncMock)
|
||||
@patch("gateway.platforms.feishu_comment.batch_query_comment", new_callable=AsyncMock)
|
||||
@patch("gateway.platforms.feishu_comment.query_document_meta", new_callable=AsyncMock)
|
||||
def test_wiki_lookup_triggered_when_no_exact_match(
|
||||
self, mock_meta, mock_batch, mock_reaction,
|
||||
mock_load, mock_resolve, mock_allowed, mock_wiki_keys, mock_lookup,
|
||||
):
|
||||
"""Wiki reverse lookup should fire when rule falls to wildcard/top and wiki keys exist."""
|
||||
from gateway.platforms.feishu_comment import handle_drive_comment_event
|
||||
from gateway.platforms.feishu_comment_rules import ResolvedCommentRule
|
||||
|
||||
# First resolve returns wildcard (no exact match), second returns exact wiki match
|
||||
mock_resolve.side_effect = [
|
||||
ResolvedCommentRule(True, "allowlist", frozenset(), "wildcard"),
|
||||
ResolvedCommentRule(True, "allowlist", frozenset(), "exact:wiki:WIKI123"),
|
||||
]
|
||||
mock_load.return_value = Mock()
|
||||
mock_lookup.return_value = "WIKI123"
|
||||
mock_meta.return_value = {"title": "Test", "url": ""}
|
||||
mock_batch.return_value = {"is_whole": False, "quote": ""}
|
||||
|
||||
evt = _make_event()
|
||||
# Will proceed past access control but fail later — that's OK, we just test the lookup
|
||||
try:
|
||||
self._run(handle_drive_comment_event(Mock(), evt, self_open_id="ou_bot"))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
mock_lookup.assert_called_once_with(unittest.mock.ANY, "docx", "docx_token")
|
||||
self.assertEqual(mock_resolve.call_count, 2)
|
||||
# Second call should include wiki_token
|
||||
second_call_kwargs = mock_resolve.call_args_list[1]
|
||||
self.assertEqual(second_call_kwargs[1].get("wiki_token") or second_call_kwargs[0][3], "WIKI123")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
320
tests/gateway/test_feishu_comment_rules.py
Normal file
320
tests/gateway/test_feishu_comment_rules.py
Normal file
@@ -0,0 +1,320 @@
|
||||
"""Tests for feishu_comment_rules — 3-tier access control rule engine."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from gateway.platforms.feishu_comment_rules import (
|
||||
CommentsConfig,
|
||||
CommentDocumentRule,
|
||||
ResolvedCommentRule,
|
||||
_MtimeCache,
|
||||
_parse_document_rule,
|
||||
has_wiki_keys,
|
||||
is_user_allowed,
|
||||
load_config,
|
||||
pairing_add,
|
||||
pairing_list,
|
||||
pairing_remove,
|
||||
resolve_rule,
|
||||
)
|
||||
|
||||
|
||||
class TestCommentDocumentRuleParsing(unittest.TestCase):
|
||||
def test_parse_full_rule(self):
|
||||
rule = _parse_document_rule({
|
||||
"enabled": False,
|
||||
"policy": "allowlist",
|
||||
"allow_from": ["ou_a", "ou_b"],
|
||||
})
|
||||
self.assertFalse(rule.enabled)
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertEqual(rule.allow_from, frozenset(["ou_a", "ou_b"]))
|
||||
|
||||
def test_parse_partial_rule(self):
|
||||
rule = _parse_document_rule({"policy": "allowlist"})
|
||||
self.assertIsNone(rule.enabled)
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertIsNone(rule.allow_from)
|
||||
|
||||
def test_parse_empty_rule(self):
|
||||
rule = _parse_document_rule({})
|
||||
self.assertIsNone(rule.enabled)
|
||||
self.assertIsNone(rule.policy)
|
||||
self.assertIsNone(rule.allow_from)
|
||||
|
||||
def test_invalid_policy_ignored(self):
|
||||
rule = _parse_document_rule({"policy": "invalid_value"})
|
||||
self.assertIsNone(rule.policy)
|
||||
|
||||
|
||||
class TestResolveRule(unittest.TestCase):
|
||||
def test_exact_match(self):
|
||||
cfg = CommentsConfig(
|
||||
policy="pairing",
|
||||
allow_from=frozenset(["ou_top"]),
|
||||
documents={
|
||||
"docx:abc": CommentDocumentRule(policy="allowlist"),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "abc")
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertTrue(rule.match_source.startswith("exact:"))
|
||||
|
||||
def test_wildcard_match(self):
|
||||
cfg = CommentsConfig(
|
||||
policy="pairing",
|
||||
documents={
|
||||
"*": CommentDocumentRule(policy="allowlist"),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "unknown")
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertEqual(rule.match_source, "wildcard")
|
||||
|
||||
def test_top_level_fallback(self):
|
||||
cfg = CommentsConfig(policy="pairing", allow_from=frozenset(["ou_top"]))
|
||||
rule = resolve_rule(cfg, "docx", "whatever")
|
||||
self.assertEqual(rule.policy, "pairing")
|
||||
self.assertEqual(rule.allow_from, frozenset(["ou_top"]))
|
||||
self.assertEqual(rule.match_source, "top")
|
||||
|
||||
def test_exact_overrides_wildcard(self):
|
||||
cfg = CommentsConfig(
|
||||
policy="pairing",
|
||||
documents={
|
||||
"*": CommentDocumentRule(policy="pairing"),
|
||||
"docx:abc": CommentDocumentRule(policy="allowlist"),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "abc")
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertTrue(rule.match_source.startswith("exact:"))
|
||||
|
||||
def test_field_by_field_fallback(self):
|
||||
"""Exact sets policy, wildcard sets allow_from, enabled from top."""
|
||||
cfg = CommentsConfig(
|
||||
enabled=True,
|
||||
policy="pairing",
|
||||
allow_from=frozenset(["ou_top"]),
|
||||
documents={
|
||||
"*": CommentDocumentRule(allow_from=frozenset(["ou_wildcard"])),
|
||||
"docx:abc": CommentDocumentRule(policy="allowlist"),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "abc")
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertEqual(rule.allow_from, frozenset(["ou_wildcard"]))
|
||||
self.assertTrue(rule.enabled)
|
||||
|
||||
def test_explicit_empty_allow_from_does_not_fall_through(self):
|
||||
"""allow_from=[] on exact should NOT inherit from wildcard or top."""
|
||||
cfg = CommentsConfig(
|
||||
allow_from=frozenset(["ou_top"]),
|
||||
documents={
|
||||
"*": CommentDocumentRule(allow_from=frozenset(["ou_wildcard"])),
|
||||
"docx:abc": CommentDocumentRule(
|
||||
policy="allowlist",
|
||||
allow_from=frozenset(),
|
||||
),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "abc")
|
||||
self.assertEqual(rule.allow_from, frozenset())
|
||||
|
||||
def test_wiki_token_match(self):
|
||||
cfg = CommentsConfig(
|
||||
policy="pairing",
|
||||
documents={
|
||||
"wiki:WIKI123": CommentDocumentRule(policy="allowlist"),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "obj_token", wiki_token="WIKI123")
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertTrue(rule.match_source.startswith("exact:wiki:"))
|
||||
|
||||
def test_exact_takes_priority_over_wiki(self):
|
||||
cfg = CommentsConfig(
|
||||
documents={
|
||||
"docx:abc": CommentDocumentRule(policy="allowlist"),
|
||||
"wiki:WIKI123": CommentDocumentRule(policy="pairing"),
|
||||
},
|
||||
)
|
||||
rule = resolve_rule(cfg, "docx", "abc", wiki_token="WIKI123")
|
||||
self.assertEqual(rule.policy, "allowlist")
|
||||
self.assertTrue(rule.match_source.startswith("exact:docx:"))
|
||||
|
||||
def test_default_config(self):
|
||||
cfg = CommentsConfig()
|
||||
rule = resolve_rule(cfg, "docx", "anything")
|
||||
self.assertTrue(rule.enabled)
|
||||
self.assertEqual(rule.policy, "pairing")
|
||||
self.assertEqual(rule.allow_from, frozenset())
|
||||
|
||||
|
||||
class TestHasWikiKeys(unittest.TestCase):
|
||||
def test_no_wiki_keys(self):
|
||||
cfg = CommentsConfig(documents={
|
||||
"docx:abc": CommentDocumentRule(policy="allowlist"),
|
||||
"*": CommentDocumentRule(policy="pairing"),
|
||||
})
|
||||
self.assertFalse(has_wiki_keys(cfg))
|
||||
|
||||
def test_has_wiki_keys(self):
|
||||
cfg = CommentsConfig(documents={
|
||||
"wiki:WIKI123": CommentDocumentRule(policy="allowlist"),
|
||||
})
|
||||
self.assertTrue(has_wiki_keys(cfg))
|
||||
|
||||
def test_empty_documents(self):
|
||||
cfg = CommentsConfig()
|
||||
self.assertFalse(has_wiki_keys(cfg))
|
||||
|
||||
|
||||
class TestIsUserAllowed(unittest.TestCase):
|
||||
def test_allowlist_allows_listed(self):
|
||||
rule = ResolvedCommentRule(True, "allowlist", frozenset(["ou_a"]), "top")
|
||||
self.assertTrue(is_user_allowed(rule, "ou_a"))
|
||||
|
||||
def test_allowlist_denies_unlisted(self):
|
||||
rule = ResolvedCommentRule(True, "allowlist", frozenset(["ou_a"]), "top")
|
||||
self.assertFalse(is_user_allowed(rule, "ou_b"))
|
||||
|
||||
def test_allowlist_empty_denies_all(self):
|
||||
rule = ResolvedCommentRule(True, "allowlist", frozenset(), "top")
|
||||
self.assertFalse(is_user_allowed(rule, "ou_anyone"))
|
||||
|
||||
def test_pairing_allows_in_allow_from(self):
|
||||
rule = ResolvedCommentRule(True, "pairing", frozenset(["ou_a"]), "top")
|
||||
self.assertTrue(is_user_allowed(rule, "ou_a"))
|
||||
|
||||
def test_pairing_checks_store(self):
|
||||
rule = ResolvedCommentRule(True, "pairing", frozenset(), "top")
|
||||
with patch(
|
||||
"gateway.platforms.feishu_comment_rules._load_pairing_approved",
|
||||
return_value={"ou_approved"},
|
||||
):
|
||||
self.assertTrue(is_user_allowed(rule, "ou_approved"))
|
||||
self.assertFalse(is_user_allowed(rule, "ou_unknown"))
|
||||
|
||||
|
||||
class TestMtimeCache(unittest.TestCase):
|
||||
def test_returns_empty_dict_for_missing_file(self):
|
||||
cache = _MtimeCache(Path("/nonexistent/path.json"))
|
||||
self.assertEqual(cache.load(), {})
|
||||
|
||||
def test_reads_file_and_caches(self):
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump({"key": "value"}, f)
|
||||
f.flush()
|
||||
path = Path(f.name)
|
||||
try:
|
||||
cache = _MtimeCache(path)
|
||||
data = cache.load()
|
||||
self.assertEqual(data, {"key": "value"})
|
||||
# Second load should use cache (same mtime)
|
||||
data2 = cache.load()
|
||||
self.assertEqual(data2, {"key": "value"})
|
||||
finally:
|
||||
path.unlink()
|
||||
|
||||
def test_reloads_on_mtime_change(self):
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump({"v": 1}, f)
|
||||
f.flush()
|
||||
path = Path(f.name)
|
||||
try:
|
||||
cache = _MtimeCache(path)
|
||||
self.assertEqual(cache.load(), {"v": 1})
|
||||
# Modify file
|
||||
time.sleep(0.05)
|
||||
with open(path, "w") as f2:
|
||||
json.dump({"v": 2}, f2)
|
||||
# Force mtime change detection
|
||||
os.utime(path, (time.time() + 1, time.time() + 1))
|
||||
self.assertEqual(cache.load(), {"v": 2})
|
||||
finally:
|
||||
path.unlink()
|
||||
|
||||
|
||||
class TestLoadConfig(unittest.TestCase):
|
||||
def test_load_with_documents(self):
|
||||
raw = {
|
||||
"enabled": True,
|
||||
"policy": "allowlist",
|
||||
"allow_from": ["ou_a"],
|
||||
"documents": {
|
||||
"*": {"policy": "pairing"},
|
||||
"docx:abc": {"policy": "allowlist", "allow_from": ["ou_b"]},
|
||||
},
|
||||
}
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(raw, f)
|
||||
path = Path(f.name)
|
||||
try:
|
||||
with patch("gateway.platforms.feishu_comment_rules.RULES_FILE", path):
|
||||
with patch("gateway.platforms.feishu_comment_rules._rules_cache", _MtimeCache(path)):
|
||||
cfg = load_config()
|
||||
self.assertTrue(cfg.enabled)
|
||||
self.assertEqual(cfg.policy, "allowlist")
|
||||
self.assertEqual(cfg.allow_from, frozenset(["ou_a"]))
|
||||
self.assertIn("*", cfg.documents)
|
||||
self.assertIn("docx:abc", cfg.documents)
|
||||
self.assertEqual(cfg.documents["docx:abc"].policy, "allowlist")
|
||||
finally:
|
||||
path.unlink()
|
||||
|
||||
def test_load_missing_file_returns_defaults(self):
|
||||
with patch("gateway.platforms.feishu_comment_rules._rules_cache", _MtimeCache(Path("/nonexistent"))):
|
||||
cfg = load_config()
|
||||
self.assertTrue(cfg.enabled)
|
||||
self.assertEqual(cfg.policy, "pairing")
|
||||
self.assertEqual(cfg.allow_from, frozenset())
|
||||
self.assertEqual(cfg.documents, {})
|
||||
|
||||
|
||||
class TestPairingStore(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmpdir = tempfile.mkdtemp()
|
||||
self._pairing_file = Path(self._tmpdir) / "pairing.json"
|
||||
with open(self._pairing_file, "w") as f:
|
||||
json.dump({"approved": {}}, f)
|
||||
self._patcher_file = patch("gateway.platforms.feishu_comment_rules.PAIRING_FILE", self._pairing_file)
|
||||
self._patcher_cache = patch(
|
||||
"gateway.platforms.feishu_comment_rules._pairing_cache",
|
||||
_MtimeCache(self._pairing_file),
|
||||
)
|
||||
self._patcher_file.start()
|
||||
self._patcher_cache.start()
|
||||
|
||||
def tearDown(self):
|
||||
self._patcher_cache.stop()
|
||||
self._patcher_file.stop()
|
||||
if self._pairing_file.exists():
|
||||
self._pairing_file.unlink()
|
||||
os.rmdir(self._tmpdir)
|
||||
|
||||
def test_add_and_list(self):
|
||||
self.assertTrue(pairing_add("ou_new"))
|
||||
approved = pairing_list()
|
||||
self.assertIn("ou_new", approved)
|
||||
|
||||
def test_add_duplicate(self):
|
||||
pairing_add("ou_a")
|
||||
self.assertFalse(pairing_add("ou_a"))
|
||||
|
||||
def test_remove(self):
|
||||
pairing_add("ou_a")
|
||||
self.assertTrue(pairing_remove("ou_a"))
|
||||
self.assertNotIn("ou_a", pairing_list())
|
||||
|
||||
def test_remove_nonexistent(self):
|
||||
self.assertFalse(pairing_remove("ou_nobody"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
62
tests/tools/test_feishu_tools.py
Normal file
62
tests/tools/test_feishu_tools.py
Normal file
@@ -0,0 +1,62 @@
|
||||
"""Tests for feishu_doc_tool and feishu_drive_tool — registration and schema validation."""
|
||||
|
||||
import importlib
|
||||
import unittest
|
||||
|
||||
from tools.registry import registry
|
||||
|
||||
# Trigger tool discovery so feishu tools get registered
|
||||
importlib.import_module("tools.feishu_doc_tool")
|
||||
importlib.import_module("tools.feishu_drive_tool")
|
||||
|
||||
|
||||
class TestFeishuToolRegistration(unittest.TestCase):
|
||||
"""Verify feishu tools are registered and have valid schemas."""
|
||||
|
||||
EXPECTED_TOOLS = {
|
||||
"feishu_doc_read": "feishu_doc",
|
||||
"feishu_drive_list_comments": "feishu_drive",
|
||||
"feishu_drive_list_comment_replies": "feishu_drive",
|
||||
"feishu_drive_reply_comment": "feishu_drive",
|
||||
"feishu_drive_add_comment": "feishu_drive",
|
||||
}
|
||||
|
||||
def test_all_tools_registered(self):
|
||||
for tool_name, toolset in self.EXPECTED_TOOLS.items():
|
||||
entry = registry.get_entry(tool_name)
|
||||
self.assertIsNotNone(entry, f"{tool_name} not registered")
|
||||
self.assertEqual(entry.toolset, toolset)
|
||||
|
||||
def test_schemas_have_required_fields(self):
|
||||
for tool_name in self.EXPECTED_TOOLS:
|
||||
entry = registry.get_entry(tool_name)
|
||||
schema = entry.schema
|
||||
self.assertIn("name", schema)
|
||||
self.assertEqual(schema["name"], tool_name)
|
||||
self.assertIn("description", schema)
|
||||
self.assertIn("parameters", schema)
|
||||
self.assertIn("type", schema["parameters"])
|
||||
self.assertEqual(schema["parameters"]["type"], "object")
|
||||
|
||||
def test_handlers_are_callable(self):
|
||||
for tool_name in self.EXPECTED_TOOLS:
|
||||
entry = registry.get_entry(tool_name)
|
||||
self.assertTrue(callable(entry.handler))
|
||||
|
||||
def test_doc_read_schema_params(self):
|
||||
entry = registry.get_entry("feishu_doc_read")
|
||||
props = entry.schema["parameters"].get("properties", {})
|
||||
self.assertIn("doc_token", props)
|
||||
|
||||
def test_drive_tools_require_file_token(self):
|
||||
for tool_name in self.EXPECTED_TOOLS:
|
||||
if tool_name == "feishu_doc_read":
|
||||
continue
|
||||
entry = registry.get_entry(tool_name)
|
||||
props = entry.schema["parameters"].get("properties", {})
|
||||
self.assertIn("file_token", props, f"{tool_name} missing file_token param")
|
||||
self.assertIn("file_type", props, f"{tool_name} missing file_type param")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user