import unittest
import tempfile
import shutil
import os
from pathlib import Path
from rlm_mcp_server.ingest import read_paths
class TestIngest(unittest.TestCase):
def setUp(self):
self.test_dir = tempfile.mkdtemp()
self.root = Path(self.test_dir)
# Create some files
(self.root / "file1.txt").write_text("content1")
(self.root / "sub").mkdir()
(self.root / "sub" / "file2.py").write_text("content2")
# Create file outside root
self.outside_dir = tempfile.mkdtemp()
(Path(self.outside_dir) / "secret.txt").write_text("secret")
def tearDown(self):
shutil.rmtree(self.test_dir)
shutil.rmtree(self.outside_dir)
def test_basic_read(self):
res = read_paths(self.root, ["**/*.txt"])
self.assertIn("file1.txt", res)
self.assertEqual(res["file1.txt"], "content1")
self.assertNotIn("sub/file2.py", res)
def test_recursive_glob(self):
res = read_paths(self.root, ["**/*"])
self.assertIn("file1.txt", res)
self.assertIn("sub/file2.py", res)
def test_repo_boundary(self):
# Symlink pointing outside (if OS supports) or just absolute path attempt
try:
os.symlink(Path(self.outside_dir) / "secret.txt", self.root / "link_to_secret")
res = read_paths(self.root, ["link_to_secret"])
self.assertEqual(len(res), 0)
except OSError:
pass
def test_max_file_bytes(self):
(self.root / "big.txt").write_text("a" * 1000)
res = read_paths(self.root, ["big.txt"], max_file_bytes=500)
self.assertNotIn("big.txt", res)
def test_max_total_bytes(self):
# Isolate this test in a subdir to avoid picking up file1.txt
sub_root = self.root / "limit_test"
sub_root.mkdir()
(sub_root / "a.txt").write_text("a" * 100)
(sub_root / "b.txt").write_text("b" * 100)
# Search only in sub_root
res = read_paths(sub_root, ["*.txt"], max_total_bytes=150)
# Should only have one file
self.assertTrue(len(res) == 1)
if __name__ == "__main__":
unittest.main()
