test: fix missing coverage for decompress_to_cache (#514)

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: n24q02m

tests/test_model_management_extra.py

@@ -0,0 +1,135 @@
+import io
+import tarfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+from loguru import logger
+
+from qwen3_embed.common.model_management import ModelManagement
+
+
+class TestModelManagementExtra:
+    """Extra tests to cover missing branches in model_management.py."""
+
+    def test_get_session_init(self):
+        """Test that _get_session initializes the session if it's None."""
+        # Save original session to restore it later
+        original_session = ModelManagement._session
+        ModelManagement._session = None
+        try:
+            session = ModelManagement._get_session()
+            assert isinstance(session, requests.Session)
+            assert ModelManagement._session is session
+
+            # Second call should return the same session
+            session2 = ModelManagement._get_session()
+            assert session2 is session
+        finally:
+            ModelManagement._session = original_session
+
+    def test_decompress_absolute_path_mock(self, tmp_path):
+        """Mock getmembers to return an absolute path to trigger line 371."""
+        tar_path = tmp_path / "test.tar.gz"
+        # Create a dummy tar file
+        with tarfile.open(tar_path, "w:gz") as tar:
+            info = tarfile.TarInfo(name="safe.txt")
+            info.size = 0
+            tar.addfile(info, io.BytesIO(b""))
+
+        cache_dir = tmp_path / "cache"
+        cache_dir.mkdir()
+
+        # Mock tarfile.open to return a tar with a malicious member
+        mock_member = MagicMock()
+        mock_member.name = "/absolute/path"
+
+        with patch("tarfile.open") as mock_open:
+            mock_tar = mock_open.return_value.__enter__.return_value
+            mock_tar.getmembers.return_value = [mock_member]
+
+            with pytest.raises(tarfile.TarError, match="Attempted path traversal"):
+                ModelManagement.decompress_to_cache(str(tar_path), str(cache_dir))
+
+    def test_decompress_safe_symlink_and_hardlink(self, tmp_path):
+        """Test safe symlinks and hardlinks to cover safe branch for links (line 399)."""
+        cache_dir = tmp_path / "cache_links"
+        cache_dir.mkdir()
+
+        tar_path = tmp_path / "links.tar.gz"
+        with tarfile.open(tar_path, "w:gz") as tar:
+            # Regular file
+            info = tarfile.TarInfo(name="file.txt")
+            info.size = 4
+            tar.addfile(info, io.BytesIO(b"data"))
+
+            # Safe symlink: points to a file within the same directory
+            sym_info = tarfile.TarInfo(name="symlink.txt")
+            sym_info.type = tarfile.SYMTYPE
+            sym_info.linkname = "file.txt"
+            tar.addfile(sym_info)
+
+            # Safe hardlink: points to a file within the same directory (relative to root)
+            hard_info = tarfile.TarInfo(name="hardlink.txt")
+            hard_info.type = tarfile.LNKTYPE
+            hard_info.linkname = "file.txt"
+            tar.addfile(hard_info)
+
+        result = ModelManagement.decompress_to_cache(str(tar_path), str(cache_dir))
+        assert result == str(cache_dir)
+        assert (cache_dir / "file.txt").exists()
+        # On some systems/Python versions, symlink/hardlink might not be fully
+        # supported or behaved differently in tests, but the logic should pass.
+        assert (cache_dir / "symlink.txt").exists()
+        assert (cache_dir / "hardlink.txt").exists()
+
+    def test_decompress_no_data_filter(self, tmp_path):
+        """Cover fallback line 412 by mocking tarfile to lack data_filter."""
+        tar_path = tmp_path / "test.tar.gz"
+        with tarfile.open(tar_path, "w:gz") as tar:
+            info = tarfile.TarInfo(name="file.txt")
+            info.size = 0
+            tar.addfile(info, io.BytesIO(b""))
+
+        cache_dir = tmp_path / "cache_no_filter"
+        cache_dir.mkdir()
+
+        # We patch the tarfile module in the model_management namespace
+        with patch("qwen3_embed.common.model_management.tarfile") as mock_tarfile_mod:
+            # Setup mock_tar
+            mock_tar = MagicMock()
+            mock_tarfile_mod.open.return_value.__enter__.return_value = mock_tar
+
+            # Mock getmembers to return a list of members
+            member = MagicMock()
+            member.name = "file.txt"
+            member.issym.return_value = False
+            member.islnk.return_value = False
+            mock_tar.getmembers.return_value = [member]
+
+            # Ensure hasattr(tarfile, 'data_filter') returns False
+            del mock_tarfile_mod.data_filter
+
+            ModelManagement.decompress_to_cache(str(tar_path), str(cache_dir))
+
+            # Verify extractall was called without filter keyword
+            mock_tar.extractall.assert_called_once()
+            args, kwargs = mock_tar.extractall.call_args
+            assert "filter" not in kwargs
+
+    def test_decompress_logging_on_error(self, tmp_path):
+        """Verify logger.error is called on TarError."""
+        tar_path = tmp_path / "corrupt.tar.gz"
+        tar_path.write_text("not a tar file")
+
+        cache_dir = tmp_path / "cache_err"
+        cache_dir.mkdir()
+
+        with patch.object(logger, "error") as mock_log_error:
+            with pytest.raises(tarfile.TarError):
+                ModelManagement.decompress_to_cache(str(tar_path), str(cache_dir))
+
+            mock_log_error.assert_called()
+            # Verify the log message contains the filename
+            args, _ = mock_log_error.call_args
+            assert str(tar_path) in args[0]