AFFAANhandClaude Opus 5
19e1e84fb7
Independent FastAPI backend for the MaskanX agentic growth platform. Includes the agent runtime, MCP client integrations (Meta Ads, LinkedIn, HubSpot, Tavily, Exa, xAI, Citedy, image generation), PostgreSQL storage for chats and cron jobs, provider and secret management, and the CLI. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
69 lines
3.0 KiB
Python
69 lines
3.0 KiB
Python
import os
|
|
import tempfile
|
|
import pytest
|
|
from adclaw.agents.tools.shared_memory import make_read_shared_file, make_write_shared_file, make_list_shared_files
|
|
|
|
|
|
class TestSharedMemory:
|
|
def setup_method(self):
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
self.shared_root = os.path.join(self.tmpdir, "shared")
|
|
os.makedirs(os.path.join(self.shared_root, "researcher"), exist_ok=True)
|
|
os.makedirs(os.path.join(self.shared_root, "content"), exist_ok=True)
|
|
|
|
def test_write_shared_file(self):
|
|
write_fn = make_write_shared_file(self.shared_root, "researcher")
|
|
result = write_fn(filename="report.md", content="# Daily Report\nTrending: AI agents")
|
|
assert "success" in result.lower()
|
|
assert os.path.exists(os.path.join(self.shared_root, "researcher", "report.md"))
|
|
|
|
def test_write_rejects_path_traversal(self):
|
|
write_fn = make_write_shared_file(self.shared_root, "researcher")
|
|
result = write_fn(filename="../../../etc/passwd", content="hack")
|
|
assert "error" in result.lower()
|
|
|
|
def test_read_shared_file(self):
|
|
with open(os.path.join(self.shared_root, "researcher", "report.md"), "w") as f:
|
|
f.write("# Report\nContent here")
|
|
read_fn = make_read_shared_file(self.shared_root)
|
|
result = read_fn(agent_id="researcher", filename="report.md")
|
|
assert "Content here" in result
|
|
|
|
def test_read_nonexistent(self):
|
|
read_fn = make_read_shared_file(self.shared_root)
|
|
result = read_fn(agent_id="researcher", filename="nope.md")
|
|
assert "not found" in result.lower()
|
|
|
|
def test_read_rejects_path_traversal(self):
|
|
read_fn = make_read_shared_file(self.shared_root)
|
|
result = read_fn(agent_id="researcher", filename="../../etc/passwd")
|
|
assert "error" in result.lower()
|
|
|
|
def test_list_shared_files(self):
|
|
with open(os.path.join(self.shared_root, "researcher", "a.md"), "w") as f:
|
|
f.write("a")
|
|
with open(os.path.join(self.shared_root, "researcher", "b.md"), "w") as f:
|
|
f.write("b")
|
|
list_fn = make_list_shared_files(self.shared_root)
|
|
result = list_fn(agent_id="researcher")
|
|
assert "a.md" in result
|
|
assert "b.md" in result
|
|
|
|
def test_list_all_agents_shared(self):
|
|
list_fn = make_list_shared_files(self.shared_root)
|
|
result = list_fn()
|
|
assert "researcher" in result
|
|
assert "content" in result
|
|
|
|
def test_write_only_own_dir(self):
|
|
write_fn = make_write_shared_file(self.shared_root, "researcher")
|
|
result = write_fn(filename="../content/hack.md", content="hacked")
|
|
assert "error" in result.lower()
|
|
|
|
def test_cross_agent_read(self):
|
|
write_r = make_write_shared_file(self.shared_root, "researcher")
|
|
write_r(filename="intel.md", content="Breaking: new model")
|
|
read_c = make_read_shared_file(self.shared_root)
|
|
result = read_c(agent_id="researcher", filename="intel.md")
|
|
assert "Breaking" in result
|