"""Unit tests for share_file tool.""" from urllib.parse import unquote, urlparse import pytest import navi.session_files as session_files_mod import navi.tools.share_file as share_file_mod from navi.config import Settings from navi.tools._internal.base import ToolContext from navi.tools.share_file import ShareFileTool class TestShareFileTool: @pytest.fixture def tool(self, monkeypatch, tmp_path): async def _to_thread(func, *args, **kwargs): return func(*args, **kwargs) monkeypatch.setattr(share_file_mod.asyncio, "to_thread", _to_thread) _test_settings = Settings( session_files_dir=str(tmp_path / "sessions"), share_file_max_size_mb=1024, public_url="http://localhost:8000", ) monkeypatch.setattr(share_file_mod, "settings", _test_settings) monkeypatch.setattr(session_files_mod, "settings", _test_settings) yield ShareFileTool() async def test_rejects_relative_path(self, tool): result = await tool.execute({"path": "workspace/report.txt"}, ctx=ToolContext(session_id="sess 1")) assert not result.success assert result.error == "path_not_absolute" async def test_copies_file_into_session_dir(self, tool, tmp_path): src = tmp_path / "report.txt" src.write_text("hello") result = await tool.execute({"path": str(src), "filename": "clean report.txt"}, ctx=ToolContext(session_id="sess 1")) assert result.success dest = tmp_path / "sessions" / "sess 1" / "clean report.txt" assert dest.read_text() == "hello" assert result.metadata["filename"] == "clean report.txt" async def test_rejects_files_over_share_limit(self, tool, monkeypatch, tmp_path): _test_settings = Settings( session_files_dir=str(tmp_path / "sessions"), share_file_max_size_mb=0, public_url="http://localhost:8000", ) monkeypatch.setattr(share_file_mod, "settings", _test_settings) monkeypatch.setattr(session_files_mod, "settings", _test_settings) src = tmp_path / "too_large.bin" src.write_bytes(b"x") result = await tool.execute({"path": str(src)}, ctx=ToolContext(session_id="sess 1")) assert not result.success assert result.error == "file_too_large" async def test_url_quotes_session_and_filename(self, tool, tmp_path): src = tmp_path / "source.txt" src.write_text("hello") result = await tool.execute({"path": str(src), "filename": "отчёт #1.txt"}, ctx=ToolContext(session_id="sess 1")) assert result.success parsed = urlparse(result.metadata["url"]) assert parsed.path.endswith( "/api/sessions/sess%201/files/%D0%BE%D1%82%D1%87%D1%91%D1%82%20%231.txt" ) assert unquote(parsed.path).endswith("/api/sessions/sess 1/files/отчёт #1.txt") async def test_duplicate_filename_gets_numbered_copy(self, tool, tmp_path): src = tmp_path / "source.txt" src.write_text("new") existing = tmp_path / "sessions" / "sess 1" / "report.txt" existing.parent.mkdir(parents=True) existing.write_text("old") result = await tool.execute({"path": str(src), "filename": "report.txt"}, ctx=ToolContext(session_id="sess 1")) assert result.success assert existing.read_text() == "old" numbered = tmp_path / "sessions" / "sess 1" / "report_1.txt" assert numbered.read_text() == "new" assert result.metadata["filename"] == "report_1.txt" class TestShareFileNonAdminSources: """A non-admin may share from the sandbox and from the current session dir. The session directory is where uploads land, so a file the user just sent has to be shareable from there — otherwise the agent must smuggle it into its sandbox first, which is exactly the bypass the security policy forbids. """ @pytest.fixture def tool(self, monkeypatch, tmp_path): async def _to_thread(func, *args, **kwargs): return func(*args, **kwargs) monkeypatch.setattr(share_file_mod.asyncio, "to_thread", _to_thread) _test_settings = Settings( session_files_dir=str(tmp_path / "sessions"), share_file_max_size_mb=1024, public_url="http://localhost:8000", ) monkeypatch.setattr(share_file_mod, "settings", _test_settings) monkeypatch.setattr(session_files_mod, "settings", _test_settings) # The sandbox is relative to the process cwd: user_data//. monkeypatch.chdir(tmp_path) yield ShareFileTool() @staticmethod def _ctx(session_id="sess 1", user_id="7"): return ToolContext(session_id=session_id, user_id=user_id, user_role="user") async def test_shares_upload_from_session_dir_in_place(self, tool, tmp_path): session = tmp_path / "sessions" / "sess 1" session.mkdir(parents=True) src = session / "Project.wav" src.write_text("audio") result = await tool.execute({"path": str(src)}, ctx=self._ctx()) assert result.success # Copied onto itself, not duplicated into Project_1.wav. assert result.metadata["filename"] == "Project.wav" assert sorted(p.name for p in session.iterdir()) == ["Project.wav"] async def test_shares_from_sandbox(self, tool, tmp_path): sandbox = tmp_path / "user_data" / "7" sandbox.mkdir(parents=True) src = sandbox / "report.txt" src.write_text("data") result = await tool.execute({"path": str(src)}, ctx=self._ctx()) assert result.success assert (tmp_path / "sessions" / "sess 1" / "report.txt").read_text() == "data" async def test_denies_source_outside_both_roots(self, tool, tmp_path): outside = tmp_path / "elsewhere.txt" outside.write_text("x") result = await tool.execute({"path": str(outside)}, ctx=self._ctx()) assert not result.success assert result.error == "access_denied" # The message names both roots, so the agent can retry in the right one. assert str(tmp_path / "user_data" / "7") in result.output assert str(tmp_path / "sessions" / "sess 1") in result.output async def test_relative_path_still_resolves_into_sandbox(self, tool, tmp_path): sandbox = tmp_path / "user_data" / "7" sandbox.mkdir(parents=True) (sandbox / "notes.txt").write_text("n") result = await tool.execute({"path": "notes.txt"}, ctx=self._ctx()) assert result.success assert (tmp_path / "sessions" / "sess 1" / "notes.txt").read_text() == "n"