", "
", "
"])
+ assert "Only JSX content" in result.content
+ assert "No markdown here" in result.content
+
+ @patch('requests.get')
+ def test_load_mdx_from_url(self, mock_get):
+ mock_get.return_value = Mock(text="# MDX from URL\n\nContent here.\n\n", raise_for_status=lambda: None)
+ loader = MDXLoader()
+ result = loader.load(SourceContent("https://example.com/content.mdx"))
+ assert "# MDX from URL" in result.content
+ assert "" not in result.content
+
+ @patch('requests.get')
+ def test_load_mdx_with_custom_headers(self, mock_get):
+ mock_get.return_value = Mock(text="# Custom headers test", raise_for_status=lambda: None)
+ loader = MDXLoader()
+ loader.load(SourceContent("https://example.com"), headers={"Authorization": "Bearer token"})
+ assert mock_get.call_args[1]['headers'] == {"Authorization": "Bearer token"}
+
+ @patch('requests.get')
+ def test_mdx_url_fetch_error(self, mock_get):
+ mock_get.side_effect = Exception("Network error")
+ with pytest.raises(ValueError, match="Error fetching MDX from URL"):
+ MDXLoader().load(SourceContent("https://example.com"))
+
+ def test_load_inline_mdx_text(self):
+ content = """# Inline MDX\n\nimport Something from 'somewhere'\n\nContent with .\n\nexport const meta = { title: 'Test' }"""
+ loader = MDXLoader()
+ result = loader.load(SourceContent(content))
+ assert "# Inline MDX" in result.content
+ assert "Content with ." in result.content
+
+ def test_empty_result_after_cleaning(self):
+ content = """
+import Something from 'somewhere'
+export const config = {}
+
+"""
+ result, _ = self._load_from_file(content)
+ assert result.content.strip() == ""
+
+ def test_edge_case_parsing(self):
+ content = """
+# Title
+
+
+Multi-line
+JSX content
+
+
+import { a, b } from 'module'
+
+export { x, y }
+
+Final text.
+"""
+ result, _ = self._load_from_file(content)
+ assert "# Title" in result.content
+ assert "JSX content" in result.content
+ assert "Final text." in result.content
+ assert all(phrase not in result.content for phrase in ["import {", "export {", "
"])
diff --git a/tests/rag/test_text_loaders.py b/tests/rag/test_text_loaders.py
new file mode 100644
index 000000000..e72738778
--- /dev/null
+++ b/tests/rag/test_text_loaders.py
@@ -0,0 +1,160 @@
+import hashlib
+import os
+import tempfile
+import pytest
+
+from crewai_tools.rag.loaders.text_loader import TextFileLoader, TextLoader
+from crewai_tools.rag.base_loader import LoaderResult
+from crewai_tools.rag.source_content import SourceContent
+
+
+def write_temp_file(content, suffix=".txt", encoding="utf-8"):
+ with tempfile.NamedTemporaryFile(mode="w", suffix=suffix, delete=False, encoding=encoding) as f:
+ f.write(content)
+ return f.name
+
+
+def cleanup_temp_file(path):
+ try:
+ os.unlink(path)
+ except FileNotFoundError:
+ pass
+
+
+class TestTextFileLoader:
+ def test_basic_text_file(self):
+ content = "This is test content\nWith multiple lines\nAnd more text"
+ path = write_temp_file(content)
+ try:
+ result = TextFileLoader().load(SourceContent(path))
+ assert isinstance(result, LoaderResult)
+ assert result.content == content
+ assert result.source == path
+ assert result.doc_id
+ assert result.metadata in (None, {})
+ finally:
+ cleanup_temp_file(path)
+
+ def test_empty_file(self):
+ path = write_temp_file("")
+ try:
+ result = TextFileLoader().load(SourceContent(path))
+ assert result.content == ""
+ finally:
+ cleanup_temp_file(path)
+
+ def test_unicode_content(self):
+ content = "Hello 世界 🌍 émojis 🎉 åäö"
+ path = write_temp_file(content)
+ try:
+ result = TextFileLoader().load(SourceContent(path))
+ assert content in result.content
+ finally:
+ cleanup_temp_file(path)
+
+ def test_large_file(self):
+ content = "\n".join(f"Line {i}" for i in range(100))
+ path = write_temp_file(content)
+ try:
+ result = TextFileLoader().load(SourceContent(path))
+ assert "Line 0" in result.content
+ assert "Line 99" in result.content
+ assert result.content.count("\n") == 99
+ finally:
+ cleanup_temp_file(path)
+
+ def test_missing_file(self):
+ with pytest.raises(FileNotFoundError):
+ TextFileLoader().load(SourceContent("/nonexistent/path.txt"))
+
+ def test_permission_denied(self):
+ path = write_temp_file("Some content")
+ os.chmod(path, 0o000)
+ try:
+ with pytest.raises(PermissionError):
+ TextFileLoader().load(SourceContent(path))
+ finally:
+ os.chmod(path, 0o644)
+ cleanup_temp_file(path)
+
+ def test_doc_id_consistency(self):
+ content = "Consistent content"
+ path = write_temp_file(content)
+ try:
+ loader = TextFileLoader()
+ result1 = loader.load(SourceContent(path))
+ result2 = loader.load(SourceContent(path))
+ expected_id = hashlib.sha256((path + content).encode("utf-8")).hexdigest()
+ assert result1.doc_id == result2.doc_id == expected_id
+ finally:
+ cleanup_temp_file(path)
+
+ def test_various_extensions(self):
+ content = "Same content"
+ for ext in [".txt", ".md", ".log", ".json"]:
+ path = write_temp_file(content, suffix=ext)
+ try:
+ result = TextFileLoader().load(SourceContent(path))
+ assert result.content == content
+ finally:
+ cleanup_temp_file(path)
+
+
+class TestTextLoader:
+ def test_basic_text(self):
+ content = "Raw text"
+ result = TextLoader().load(SourceContent(content))
+ expected_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()
+ assert result.content == content
+ assert result.source == expected_hash
+ assert result.doc_id == expected_hash
+
+ def test_multiline_text(self):
+ content = "Line 1\nLine 2\nLine 3"
+ result = TextLoader().load(SourceContent(content))
+ assert "Line 2" in result.content
+
+ def test_empty_text(self):
+ result = TextLoader().load(SourceContent(""))
+ assert result.content == ""
+ assert result.source == hashlib.sha256("".encode("utf-8")).hexdigest()
+
+ def test_unicode_text(self):
+ content = "世界 🌍 émojis 🎉 åäö"
+ result = TextLoader().load(SourceContent(content))
+ assert content in result.content
+
+ def test_special_characters(self):
+ content = "!@#$$%^&*()_+-=~`{}[]\\|;:'\",.<>/?"
+ result = TextLoader().load(SourceContent(content))
+ assert result.content == content
+
+ def test_doc_id_uniqueness(self):
+ result1 = TextLoader().load(SourceContent("A"))
+ result2 = TextLoader().load(SourceContent("B"))
+ assert result1.doc_id != result2.doc_id
+
+ def test_whitespace_text(self):
+ content = " \n\t "
+ result = TextLoader().load(SourceContent(content))
+ assert result.content == content
+
+ def test_long_text(self):
+ content = "A" * 10000
+ result = TextLoader().load(SourceContent(content))
+ assert len(result.content) == 10000
+
+
+class TestTextLoadersIntegration:
+ def test_consistency_between_loaders(self):
+ content = "Consistent content"
+ text_result = TextLoader().load(SourceContent(content))
+ file_path = write_temp_file(content)
+ try:
+ file_result = TextFileLoader().load(SourceContent(file_path))
+
+ assert text_result.content == file_result.content
+ assert text_result.source != file_result.source
+ assert text_result.doc_id != file_result.doc_id
+ finally:
+ cleanup_temp_file(file_path)
diff --git a/tests/rag/test_webpage_loader.py b/tests/rag/test_webpage_loader.py
new file mode 100644
index 000000000..9e02f410b
--- /dev/null
+++ b/tests/rag/test_webpage_loader.py
@@ -0,0 +1,137 @@
+import pytest
+from unittest.mock import patch, Mock
+from crewai_tools.rag.loaders.webpage_loader import WebPageLoader
+from crewai_tools.rag.base_loader import LoaderResult
+from crewai_tools.rag.source_content import SourceContent
+
+
+class TestWebPageLoader:
+ def setup_mock_response(self, text, status_code=200, content_type="text/html"):
+ response = Mock()
+ response.text = text
+ response.apparent_encoding = "utf-8"
+ response.status_code = status_code
+ response.headers = {"content-type": content_type}
+ return response
+
+ def setup_mock_soup(self, text, title=None, script_style_elements=None):
+ soup = Mock()
+ soup.get_text.return_value = text
+ soup.title = Mock(string=title) if title is not None else None
+ soup.return_value = script_style_elements or []
+ return soup
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_load_basic_webpage(self, mock_bs, mock_get):
+ mock_get.return_value = self.setup_mock_response("Test PageTest content
")
+ mock_bs.return_value = self.setup_mock_soup("Test content", title="Test Page")
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com"))
+
+ assert isinstance(result, LoaderResult)
+ assert result.content == "Test content"
+ assert result.metadata["title"] == "Test Page"
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_load_webpage_with_scripts_and_styles(self, mock_bs, mock_get):
+ html = """
+ Page with Scripts
+ Visible content
+ """
+ mock_get.return_value = self.setup_mock_response(html)
+ scripts = [Mock(), Mock()]
+ styles = [Mock()]
+ for el in scripts + styles:
+ el.decompose = Mock()
+ mock_bs.return_value = self.setup_mock_soup("Page with Scripts Visible content", title="Page with Scripts", script_style_elements=scripts + styles)
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com/with-scripts"))
+
+ assert "Visible content" in result.content
+ for el in scripts + styles:
+ el.decompose.assert_called_once()
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_text_cleaning_and_title_handling(self, mock_bs, mock_get):
+ mock_get.return_value = self.setup_mock_response(" Messy text
")
+ mock_bs.return_value = self.setup_mock_soup("Text with extra spaces\n\n More\t text \n\n", title=None)
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com/messy-text"))
+ assert result.content is not None
+ assert result.metadata["title"] == ""
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_empty_or_missing_title(self, mock_bs, mock_get):
+ for title in [None, ""]:
+ mock_get.return_value = self.setup_mock_response("Content")
+ mock_bs.return_value = self.setup_mock_soup("Content", title=title)
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com"))
+ assert result.metadata["title"] == ""
+
+ @patch('requests.get')
+ def test_custom_and_default_headers(self, mock_get):
+ mock_get.return_value = self.setup_mock_response("Test")
+ custom_headers = {"User-Agent": "Bot", "Authorization": "Bearer xyz", "Accept": "text/html"}
+
+ with patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup') as mock_bs:
+ mock_bs.return_value = self.setup_mock_soup("Test")
+ WebPageLoader().load(SourceContent("https://example.com"), headers=custom_headers)
+
+ assert mock_get.call_args[1]['headers'] == custom_headers
+
+ @patch('requests.get')
+ def test_error_handling(self, mock_get):
+ for error in [Exception("Fail"), ValueError("Bad"), ImportError("Oops")]:
+ mock_get.side_effect = error
+ with pytest.raises(ValueError, match="Error loading webpage"):
+ WebPageLoader().load(SourceContent("https://example.com"))
+
+ @patch('requests.get')
+ def test_timeout_and_http_error(self, mock_get):
+ import requests
+ mock_get.side_effect = requests.Timeout("Timeout")
+ with pytest.raises(ValueError):
+ WebPageLoader().load(SourceContent("https://example.com"))
+
+ mock_response = Mock()
+ mock_response.raise_for_status.side_effect = requests.HTTPError("404")
+ mock_get.side_effect = None
+ mock_get.return_value = mock_response
+ with pytest.raises(ValueError):
+ WebPageLoader().load(SourceContent("https://example.com/404"))
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_doc_id_consistency(self, mock_bs, mock_get):
+ mock_get.return_value = self.setup_mock_response("Doc")
+ mock_bs.return_value = self.setup_mock_soup("Doc")
+
+ loader = WebPageLoader()
+ result1 = loader.load(SourceContent("https://example.com"))
+ result2 = loader.load(SourceContent("https://example.com"))
+
+ assert result1.doc_id == result2.doc_id
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_status_code_and_content_type(self, mock_bs, mock_get):
+ for status in [200, 201, 301]:
+ mock_get.return_value = self.setup_mock_response(f"Status {status}", status_code=status)
+ mock_bs.return_value = self.setup_mock_soup(f"Status {status}")
+ result = WebPageLoader().load(SourceContent(f"https://example.com/{status}"))
+ assert result.metadata["status_code"] == status
+
+ for ctype in ["text/html", "text/plain", "application/xhtml+xml"]:
+ mock_get.return_value = self.setup_mock_response("Content", content_type=ctype)
+ mock_bs.return_value = self.setup_mock_soup("Content")
+ result = WebPageLoader().load(SourceContent("https://example.com"))
+ assert result.metadata["content_type"] == ctype
diff --git a/tests/rag/test_xml_loader.py b/tests/rag/test_xml_loader.py
new file mode 100644
index 000000000..9e02f410b
--- /dev/null
+++ b/tests/rag/test_xml_loader.py
@@ -0,0 +1,137 @@
+import pytest
+from unittest.mock import patch, Mock
+from crewai_tools.rag.loaders.webpage_loader import WebPageLoader
+from crewai_tools.rag.base_loader import LoaderResult
+from crewai_tools.rag.source_content import SourceContent
+
+
+class TestWebPageLoader:
+ def setup_mock_response(self, text, status_code=200, content_type="text/html"):
+ response = Mock()
+ response.text = text
+ response.apparent_encoding = "utf-8"
+ response.status_code = status_code
+ response.headers = {"content-type": content_type}
+ return response
+
+ def setup_mock_soup(self, text, title=None, script_style_elements=None):
+ soup = Mock()
+ soup.get_text.return_value = text
+ soup.title = Mock(string=title) if title is not None else None
+ soup.return_value = script_style_elements or []
+ return soup
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_load_basic_webpage(self, mock_bs, mock_get):
+ mock_get.return_value = self.setup_mock_response("Test PageTest content
")
+ mock_bs.return_value = self.setup_mock_soup("Test content", title="Test Page")
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com"))
+
+ assert isinstance(result, LoaderResult)
+ assert result.content == "Test content"
+ assert result.metadata["title"] == "Test Page"
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_load_webpage_with_scripts_and_styles(self, mock_bs, mock_get):
+ html = """
+ Page with Scripts
+ Visible content
+ """
+ mock_get.return_value = self.setup_mock_response(html)
+ scripts = [Mock(), Mock()]
+ styles = [Mock()]
+ for el in scripts + styles:
+ el.decompose = Mock()
+ mock_bs.return_value = self.setup_mock_soup("Page with Scripts Visible content", title="Page with Scripts", script_style_elements=scripts + styles)
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com/with-scripts"))
+
+ assert "Visible content" in result.content
+ for el in scripts + styles:
+ el.decompose.assert_called_once()
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_text_cleaning_and_title_handling(self, mock_bs, mock_get):
+ mock_get.return_value = self.setup_mock_response(" Messy text
")
+ mock_bs.return_value = self.setup_mock_soup("Text with extra spaces\n\n More\t text \n\n", title=None)
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com/messy-text"))
+ assert result.content is not None
+ assert result.metadata["title"] == ""
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_empty_or_missing_title(self, mock_bs, mock_get):
+ for title in [None, ""]:
+ mock_get.return_value = self.setup_mock_response("Content")
+ mock_bs.return_value = self.setup_mock_soup("Content", title=title)
+
+ loader = WebPageLoader()
+ result = loader.load(SourceContent("https://example.com"))
+ assert result.metadata["title"] == ""
+
+ @patch('requests.get')
+ def test_custom_and_default_headers(self, mock_get):
+ mock_get.return_value = self.setup_mock_response("Test")
+ custom_headers = {"User-Agent": "Bot", "Authorization": "Bearer xyz", "Accept": "text/html"}
+
+ with patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup') as mock_bs:
+ mock_bs.return_value = self.setup_mock_soup("Test")
+ WebPageLoader().load(SourceContent("https://example.com"), headers=custom_headers)
+
+ assert mock_get.call_args[1]['headers'] == custom_headers
+
+ @patch('requests.get')
+ def test_error_handling(self, mock_get):
+ for error in [Exception("Fail"), ValueError("Bad"), ImportError("Oops")]:
+ mock_get.side_effect = error
+ with pytest.raises(ValueError, match="Error loading webpage"):
+ WebPageLoader().load(SourceContent("https://example.com"))
+
+ @patch('requests.get')
+ def test_timeout_and_http_error(self, mock_get):
+ import requests
+ mock_get.side_effect = requests.Timeout("Timeout")
+ with pytest.raises(ValueError):
+ WebPageLoader().load(SourceContent("https://example.com"))
+
+ mock_response = Mock()
+ mock_response.raise_for_status.side_effect = requests.HTTPError("404")
+ mock_get.side_effect = None
+ mock_get.return_value = mock_response
+ with pytest.raises(ValueError):
+ WebPageLoader().load(SourceContent("https://example.com/404"))
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_doc_id_consistency(self, mock_bs, mock_get):
+ mock_get.return_value = self.setup_mock_response("Doc")
+ mock_bs.return_value = self.setup_mock_soup("Doc")
+
+ loader = WebPageLoader()
+ result1 = loader.load(SourceContent("https://example.com"))
+ result2 = loader.load(SourceContent("https://example.com"))
+
+ assert result1.doc_id == result2.doc_id
+
+ @patch('requests.get')
+ @patch('crewai_tools.rag.loaders.webpage_loader.BeautifulSoup')
+ def test_status_code_and_content_type(self, mock_bs, mock_get):
+ for status in [200, 201, 301]:
+ mock_get.return_value = self.setup_mock_response(f"Status {status}", status_code=status)
+ mock_bs.return_value = self.setup_mock_soup(f"Status {status}")
+ result = WebPageLoader().load(SourceContent(f"https://example.com/{status}"))
+ assert result.metadata["status_code"] == status
+
+ for ctype in ["text/html", "text/plain", "application/xhtml+xml"]:
+ mock_get.return_value = self.setup_mock_response("Content", content_type=ctype)
+ mock_bs.return_value = self.setup_mock_soup("Content")
+ result = WebPageLoader().load(SourceContent("https://example.com"))
+ assert result.metadata["content_type"] == ctype