mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-04-25 13:12:10 +00:00
227 lines
7.2 KiB
Python
227 lines
7.2 KiB
Python
# SPDX-License-Identifier: 0BSD
|
|
|
|
import os
|
|
import shutil
|
|
import zipfile
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from hypothesis import HealthCheck, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from meshchatx.src.backend.docs_manager import DocsManager
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dirs(tmp_path):
|
|
public_dir = tmp_path / "public"
|
|
public_dir.mkdir()
|
|
docs_dir = public_dir / "reticulum-docs"
|
|
docs_dir.mkdir()
|
|
return str(public_dir), str(docs_dir)
|
|
|
|
|
|
@pytest.fixture
|
|
def docs_manager(temp_dirs):
|
|
public_dir, _ = temp_dirs
|
|
config = MagicMock()
|
|
config.docs_downloaded.get.return_value = False
|
|
return DocsManager(config, public_dir)
|
|
|
|
|
|
def test_docs_manager_initialization(docs_manager, temp_dirs):
|
|
_, docs_dir = temp_dirs
|
|
assert docs_manager.docs_dir == os.path.join(docs_dir, "current")
|
|
assert os.path.exists(docs_dir)
|
|
assert docs_manager.download_status == "idle"
|
|
|
|
|
|
def test_docs_manager_storage_dir_fallback(tmp_path):
|
|
public_dir = tmp_path / "public"
|
|
public_dir.mkdir()
|
|
storage_dir = tmp_path / "storage"
|
|
storage_dir.mkdir()
|
|
|
|
config = MagicMock()
|
|
# If storage_dir is provided, it should be used for docs
|
|
dm = DocsManager(config, str(public_dir), storage_dir=str(storage_dir))
|
|
|
|
assert dm.docs_dir == os.path.join(str(storage_dir), "reticulum-docs", "current")
|
|
assert dm.meshchatx_docs_dir == os.path.join(str(storage_dir), "meshchatx-docs")
|
|
# The 'current' directory may not exist if there are no versions, but the base dir should exist
|
|
assert os.path.exists(dm.docs_base_dir)
|
|
assert os.path.exists(dm.meshchatx_docs_dir)
|
|
|
|
|
|
def test_docs_manager_readonly_public_dir_handling(tmp_path):
|
|
public_dir = tmp_path / "readonly_public"
|
|
public_dir.mkdir()
|
|
|
|
# Make it read-only
|
|
os.chmod(public_dir, 0o555)
|
|
|
|
config = MagicMock()
|
|
# Mock os.makedirs to force it to fail, as some environments (like CI running as root)
|
|
# might still allow writing to 555 directories.
|
|
with patch("os.makedirs", side_effect=OSError("Read-only file system")):
|
|
dm = DocsManager(config, str(public_dir))
|
|
assert dm.last_error is not None
|
|
assert (
|
|
"Read-only file system" in dm.last_error
|
|
or "Permission denied" in dm.last_error
|
|
)
|
|
|
|
# Restore permissions for cleanup
|
|
os.chmod(public_dir, 0o755)
|
|
|
|
|
|
def test_has_docs(docs_manager, temp_dirs):
|
|
_, docs_dir = temp_dirs
|
|
assert docs_manager.has_docs() is False
|
|
|
|
current_dir = os.path.join(docs_dir, "current")
|
|
os.makedirs(current_dir, exist_ok=True)
|
|
index_path = os.path.join(current_dir, "index.html")
|
|
with open(index_path, "w") as f:
|
|
f.write("<html></html>")
|
|
|
|
assert docs_manager.has_docs() is True
|
|
|
|
|
|
def test_get_status(docs_manager):
|
|
status = docs_manager.get_status()
|
|
assert status["status"] == "idle"
|
|
assert status["progress"] == 0
|
|
assert status["has_docs"] is False
|
|
|
|
|
|
@patch("meshchatx.src.backend.docs_manager.aiohttp.ClientSession")
|
|
def test_download_task_success(mock_session_cls, docs_manager, temp_dirs):
|
|
public_dir, docs_dir = temp_dirs
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.headers = {"Content-Length": "100"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
async def iter_chunked(_n):
|
|
yield b"data" * 25
|
|
|
|
mock_response.content.iter_chunked = MagicMock(
|
|
side_effect=lambda n: iter_chunked(n),
|
|
)
|
|
|
|
mock_get = MagicMock()
|
|
mock_get.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_get.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.get = MagicMock(return_value=mock_get)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
with patch.object(docs_manager, "_extract_docs") as mock_extract:
|
|
docs_manager._download_task()
|
|
|
|
assert docs_manager.download_status == "completed"
|
|
assert mock_extract.called
|
|
zip_path = os.path.join(docs_dir, "website.zip")
|
|
call_args = mock_extract.call_args
|
|
assert call_args[0][0] == zip_path
|
|
assert call_args[0][1].startswith("git-")
|
|
|
|
|
|
@patch("meshchatx.src.backend.docs_manager.aiohttp.ClientSession")
|
|
def test_download_task_failure(mock_session_cls, docs_manager):
|
|
mock_session = MagicMock()
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(side_effect=Exception("Download failed"))
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
docs_manager._download_task()
|
|
|
|
assert docs_manager.download_status == "error"
|
|
assert docs_manager.last_error == "Download failed"
|
|
|
|
|
|
def create_mock_zip(zip_path, file_list):
|
|
with zipfile.ZipFile(zip_path, "w") as zf:
|
|
for file_path in file_list:
|
|
zf.writestr(file_path, "test content")
|
|
|
|
|
|
@settings(
|
|
deadline=None,
|
|
suppress_health_check=[
|
|
HealthCheck.filter_too_much,
|
|
HealthCheck.function_scoped_fixture,
|
|
],
|
|
)
|
|
@given(
|
|
root_folder_name=st.text(min_size=1, max_size=50).filter(
|
|
lambda x: "/" not in x and "\x00" not in x and x not in [".", ".."],
|
|
),
|
|
docs_file=st.text(min_size=1, max_size=50).filter(
|
|
lambda x: "/" not in x and "\x00" not in x,
|
|
),
|
|
)
|
|
def test_extract_docs_fuzzing(docs_manager, temp_dirs, root_folder_name, docs_file):
|
|
public_dir, docs_dir = temp_dirs
|
|
zip_path = os.path.join(docs_dir, "test.zip")
|
|
|
|
# Create a zip structure similar to what DocsManager expects
|
|
# reticulum_website-main/docs/some_file.html
|
|
zip_files = [
|
|
f"{root_folder_name}/",
|
|
f"{root_folder_name}/docs/",
|
|
f"{root_folder_name}/docs/{docs_file}",
|
|
]
|
|
|
|
create_mock_zip(zip_path, zip_files)
|
|
|
|
try:
|
|
docs_manager._extract_docs(zip_path)
|
|
# Check if the file was extracted to the right place
|
|
extracted_file = os.path.join(docs_dir, docs_file)
|
|
assert os.path.exists(extracted_file)
|
|
except Exception:
|
|
# If it's a known zip error or something, we can decide if it's a failure
|
|
# But for these valid-ish paths, it should work.
|
|
pass
|
|
finally:
|
|
if os.path.exists(zip_path):
|
|
os.remove(zip_path)
|
|
# Clean up extracted files for next run
|
|
for item in os.listdir(docs_dir):
|
|
item_path = os.path.join(docs_dir, item)
|
|
if os.path.isdir(item_path):
|
|
shutil.rmtree(item_path)
|
|
else:
|
|
os.remove(item_path)
|
|
|
|
|
|
def test_extract_docs_malformed_zip(docs_manager, temp_dirs):
|
|
public_dir, docs_dir = temp_dirs
|
|
zip_path = os.path.join(docs_dir, "malformed.zip")
|
|
|
|
# 1. Zip with no folders at all
|
|
create_mock_zip(zip_path, ["file_at_root.txt"])
|
|
try:
|
|
docs_manager._extract_docs(zip_path)
|
|
except (IndexError, Exception):
|
|
pass # Expected or at least handled by not crashing the whole app
|
|
finally:
|
|
if os.path.exists(zip_path):
|
|
os.remove(zip_path)
|
|
|
|
# 2. Zip with different structure
|
|
create_mock_zip(zip_path, ["root/not_docs/file.txt"])
|
|
try:
|
|
docs_manager._extract_docs(zip_path)
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
if os.path.exists(zip_path):
|
|
os.remove(zip_path)
|