127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from runtime import (
|
|
AgentApi,
|
|
AppFlowyMcp,
|
|
CoverMedia,
|
|
LocalArticleGenerator,
|
|
_json_from_text,
|
|
)
|
|
|
|
|
|
class FakeAppFlowy(AppFlowyMcp):
|
|
def __init__(self):
|
|
super().__init__(Path("unused"))
|
|
self.edits = []
|
|
|
|
def _call(self, tool, arguments):
|
|
if tool == "get_workspace_folder":
|
|
return {
|
|
"data": {
|
|
"view_id": "3a743541-abbf-495a-a18a-74f405fa5d70",
|
|
"children": [
|
|
{"view_id": "a", "name": "A", "children": [
|
|
{"view_id": "b", "name": "B", "children": []},
|
|
]},
|
|
],
|
|
}
|
|
}
|
|
if tool == "get_page_blocks":
|
|
return {"blocks": [{"id": "x", "text": "body"}]}
|
|
if tool == "edit_block_text":
|
|
self.edits.append(arguments)
|
|
return {"code": 0}
|
|
raise AssertionError(tool)
|
|
|
|
|
|
class FakeAgentApi(AgentApi):
|
|
def __init__(self):
|
|
super().__init__("test-key")
|
|
self.responses = []
|
|
self.requests = []
|
|
self.public = (200, b"page")
|
|
self.rsc = (200, b"rsc")
|
|
|
|
def _request(self, method, path, body=None, *, anonymous=False):
|
|
self.requests.append((method, path, body, anonymous))
|
|
if self.responses:
|
|
return self.responses.pop(0)
|
|
return 200, b"{}", "application/json"
|
|
|
|
def _public_get(self, url, *, rsc=False):
|
|
return self.rsc if rsc else self.public
|
|
|
|
|
|
class RuntimeTests(unittest.TestCase):
|
|
def test_unwraps_nested_mcp_result(self):
|
|
nested = json.dumps({"result": json.dumps({"data": {"ok": True}})})
|
|
self.assertEqual(_json_from_text(nested), {"data": {"ok": True}})
|
|
|
|
def test_appflowy_enforces_enumerated_subtree(self):
|
|
client = FakeAppFlowy()
|
|
self.assertEqual([page["view_id"] for page in client.list_pages()], ["a", "b"])
|
|
self.assertEqual(client.get_blocks("b")[0]["id"], "x")
|
|
client.edit_block("a", "x", "new")
|
|
self.assertEqual(client.edits[0]["page_id"], "a")
|
|
with self.assertRaises(PermissionError):
|
|
client.get_blocks("outside")
|
|
|
|
def test_agent_api_records_body_probe_and_decodes_create(self):
|
|
api = FakeAgentApi()
|
|
api.responses = [(201, json.dumps({"slug": "guide"}).encode(), "application/json")]
|
|
body = {
|
|
"slug": "guide",
|
|
"content": {"root": {"children": [
|
|
{"children": [{"text": "distinctive protected body sentence"}]},
|
|
]}},
|
|
}
|
|
self.assertEqual(api.create(body)["slug"], "guide")
|
|
self.assertEqual(api.body_probes["guide"], "distinctive protected body sentence")
|
|
|
|
def test_agent_api_updates_with_patch(self):
|
|
api = FakeAgentApi()
|
|
api.responses = [(200, json.dumps({"slug": "guide"}).encode(), "application/json")]
|
|
self.assertEqual(api.update("guide", {"publish": True})["slug"], "guide")
|
|
self.assertEqual(api.requests[0][0:2], ("PATCH", "/api/agent/posts/guide"))
|
|
|
|
@patch("runtime.time.sleep", return_value=None)
|
|
def test_vip_verification_rejects_body_leak(self, _sleep):
|
|
api = FakeAgentApi()
|
|
api.body_probes["guide"] = "VIP_BODY_MUST_NOT_LEAK"
|
|
api.public = (200, b"VIP_BODY_MUST_NOT_LEAK")
|
|
record = {"slug": "guide", "access": "vip", "url": "https://anwarpy.ai/posts/guide"}
|
|
self.assertFalse(api.verify_public(record))
|
|
|
|
def test_local_article_parser_enforces_word_count(self):
|
|
article = {
|
|
"title": "دليل عربي",
|
|
"slug": "arabic-guide",
|
|
"dek": "وصف عملي",
|
|
"teaser": "",
|
|
"sections": [
|
|
{"heading": "أ", "paragraphs": ["كلمة " * 300], "code": None},
|
|
{"heading": "ب", "paragraphs": ["كلمة " * 300], "code": None},
|
|
{"heading": "ج", "paragraphs": ["كلمة " * 300], "code": None},
|
|
],
|
|
}
|
|
parsed = LocalArticleGenerator._parse(json.dumps(article, ensure_ascii=False), "free", None)
|
|
self.assertEqual(parsed["slug"], "arabic-guide")
|
|
article["sections"][0]["paragraphs"] = ["قصير"]
|
|
with self.assertRaisesRegex(Exception, "900 to 2500"):
|
|
LocalArticleGenerator._parse(json.dumps(article, ensure_ascii=False), "free", None)
|
|
|
|
def test_cover_media_returns_cover_without_video(self):
|
|
media = CoverMedia(Path(tempfile.mkdtemp())).render(
|
|
{"slug": "arabic-guide"}, {"category": "أدوات"}
|
|
)
|
|
self.assertIn("coverImage", media)
|
|
self.assertNotIn("video_path", media)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|