Refactor variable system and use pytest.

Author: ChenPi11

.gitignore

@@ -126,6 +126,9 @@ dmypy.json
 # VizTracer
 result.json
 
+# Test data
+!tests/*
+
 # Patterns for all subdirectories: all kinds of automatic backup files.
 *.orig
 *.rej

extensions/subpackages/rubisco.json

@@ -130,56 +130,56 @@ def _show_pkgs(pkg: ExtensionPackageInfo) -> None:
         IKernelTrigger.on_output,
         message=format_str(
             _("Package: [green]${{name}}[/green]"),
-            {"name": pkg.name},
+            fmt={"name": pkg.name},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("Version: [cyan]${{ver}}[/cyan]"),
-            {"ver": str(pkg.version)},
+            fmt={"ver": str(pkg.version)},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("Installation Location: ${{loc}}"),
-            {"loc": _get_envtype_str(pkg.env_type)},
+            fmt={"loc": _get_envtype_str(pkg.env_type)},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("Maintainers: ${{maintainers}}"),
-            {"maintainers": ", ".join(pkg.maintainers)},
+            fmt={"maintainers": ", ".join(pkg.maintainers)},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("Homepage: ${{homepage}}"),
-            {"homepage": pkg.homepage},
+            fmt={"homepage": pkg.homepage},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("License: ${{license}}"),
-            {"license": pkg.pkg_license},
+            fmt={"license": pkg.pkg_license},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("Tags: ${{tags}}"),
-            {"tags": ", ".join(pkg.tags)},
+            fmt={"tags": ", ".join(pkg.tags)},
         ),
     )
     call_ktrigger(
         IKernelTrigger.on_output,
         message=format_str(
             _("Description: ${{desc}}"),
-            {"desc": pkg.description},
+            fmt={"desc": pkg.description},
         ),
     )
 

rubisco/cli/main/extman_cmds.py

@@ -41,7 +41,7 @@
 TIMEOUT = 15
 COPY_BUFSIZE = 1024 * 1024 if os.name == "nt" else 64 * 1024
 
-# Lib onfigurations.
+# Lib configurations.
 WORKSPACE_LIB_DIR = Path(f".{APP_NAME}")
 WORKSPACE_CONFIG_DIR = WORKSPACE_LIB_DIR
 WORKSPACE_CONFIG_FILE = WORKSPACE_LIB_DIR / "config.json"

rubisco/config.py

@@ -27,18 +27,20 @@
 import os
 import tarfile
 import zipfile
-from typing import TYPE_CHECKING
+from pathlib import Path
 
 import py7zr
 import py7zr.callbacks
 import py7zr.exceptions
+import pytest
 
 from rubisco.config import COPY_BUFSIZE
 from rubisco.lib.archive.sevenzip import compress_7z, extract_7z
 from rubisco.lib.archive.tar import compress_tarball, extract_tarball
 from rubisco.lib.archive.zip import compress_zip, extract_zip
 from rubisco.lib.exceptions import RUValueError
 from rubisco.lib.fileutil import (
+    TemporaryObject,
     assert_rel_path,
     check_file_exists,
     rm_recursive,
@@ -48,9 +50,6 @@
 from rubisco.lib.variable import format_str
 from rubisco.shared.ktrigger import IKernelTrigger, call_ktrigger
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
 __all__ = ["compress", "extract"]
 
 
@@ -184,13 +183,14 @@ def _extract(
         raise UnsupportedArchiveTypeError
 
 
-def extract(  # pylint: disable=too-many-branches
+def extract(  # pylint: disable=R0913 # noqa: PLR0913
     file: Path,
     dest: Path,
     compress_type: str | None = None,
     password: str | None = None,
     *,
     overwrite: bool = False,
+    allow_absolute_dest: bool = False,
 ) -> None:
     """Extract compressed file to destination.
 
@@ -204,10 +204,13 @@ def extract(  # pylint: disable=too-many-branches
             Defaults to None. Tarball is not supported.
         overwrite (bool, optional): Overwrite destination if it exists.
             Defaults to False.
+        allow_absolute_dest (bool, optional): Allow absolute destination path.
+            Defaults to False.
 
     """
     compress_type = compress_type.lower().strip() if compress_type else None
-    assert_rel_path(dest)
+    if not allow_absolute_dest:
+        assert_rel_path(dest)
     try:
         if compress_type is None:
             suffix1 = file.suffix
@@ -229,13 +232,13 @@ def extract(  # pylint: disable=too-many-branches
                     ),
                     hint=_("Please specify the compression type explicitly."),
                 )
-            _extract(
-                compress_type,
-                file,
-                dest,
-                password,
-                overwrite=overwrite,
-            )
+        _extract(
+            compress_type,
+            file,
+            dest,
+            password,
+            overwrite=overwrite,
+        )
     except UnsupportedArchiveTypeError:
         logger.error(
             "Unsupported compression type: '%s'",
@@ -440,7 +443,6 @@ def _compress(  # pylint: disable=R0913, R0917 # noqa: PLR0913
         raise UnsupportedArchiveTypeError
 
 
-# We should rewrite this ugly function later.
 def compress(  # pylint: disable=R0913, R0917 # noqa: PLR0913
     src: Path,
     dest: Path,
@@ -450,6 +452,7 @@ def compress(  # pylint: disable=R0913, R0917 # noqa: PLR0913
     compress_level: int | None = None,
     *,
     overwrite: bool = False,
+    allow_absolute_dest: bool = False,
 ) -> None:
     """Compress a file or directory to destination.
 
@@ -467,10 +470,13 @@ def compress(  # pylint: disable=R0913, R0917 # noqa: PLR0913
             others.
         overwrite (bool, optional): Overwrite destination if it exists.
             Defaults to False.
+        allow_absolute_dest (bool, optional): Allow absolute destination path.
+            Defaults to False.
 
     """
     compress_type = compress_type.lower().strip() if compress_type else None
-    assert_rel_path(dest)
+    if not allow_absolute_dest:
+        assert_rel_path(dest)
     try:
         if compress_type is None:
             suffix1 = dest.suffix
@@ -534,3 +540,147 @@ def compress(  # pylint: disable=R0913, R0917 # noqa: PLR0913
                 fmt={"src": str(src), "dest": str(dest), "exc": str(exc)},
             ),
         ) from exc
+
+
+class TestArchive:
+    """Test archive module."""
+
+    def _check_extract_archive(self, path: Path) -> None:
+        """Check extract archive.
+
+        Args:
+            path (Path): Extract destination.
+
+        """
+        if not path.is_dir():
+            raise AssertionError
+        if not (path / "dir1").is_dir():
+            raise AssertionError
+        if not (path / "dir1" / "file1").is_file():
+            raise AssertionError
+        if (path / "dir1" / "file1").stat().st_size != 0:
+            raise AssertionError
+        with (path / "file2").open(encoding="utf-8") as f:
+            if f.read() != "Test\n":
+                raise AssertionError
+        if not (path / "file3").is_file():
+            raise AssertionError
+
+    def _extract(self, compress_type: str, password: str = "") -> None:
+        with TemporaryObject.new_directory(suffix="test") as path:
+            archive_file = Path(f"tests/test.{compress_type}")
+            if password:
+                archive_file = Path(f"tests/test-pwd.{compress_type}")
+            extract(
+                archive_file,
+                path.path / "test",
+                password=password,
+                allow_absolute_dest=True,
+            )
+            self._check_extract_archive(path.path / "test")
+
+    def test_extract_7z(self) -> None:
+        """Test extract 7z."""
+        self._extract("7z")
+
+    def test_extract_zip(self) -> None:
+        """Test extract zip."""
+        self._extract("zip")
+
+    def test_extract_tar_gz(self) -> None:
+        """Test extract tar.gz."""
+        self._extract("tar.gz")
+
+    def test_extract_tar_xz(self) -> None:
+        """Test extract tar.xz."""
+        self._extract("tar.xz")
+
+    def test_extract_tgz(self) -> None:
+        """Test extract tgz (alias of tar.gz)."""
+        self._extract("tgz")
+
+    def test_extract_password_zip(self) -> None:
+        """Test extract zip with password."""
+        self._extract("zip", password="1234")  # noqa: S106
+
+    def test_extract_invalid_password(self) -> None:
+        """Test extract zip with invalid password."""
+        pytest.raises(
+            RuntimeError,
+            self._extract,
+            "zip",
+            password="0",  # noqa: S106
+        )
+
+    def test_extract_to_absolute(self) -> None:
+        """Test extract tgz (alias of tar.gz)."""
+        self._extract("tgz")
+
+    def _compress(self, compress_type: str) -> None:
+        with TemporaryObject.new_directory(suffix="test") as path:
+            compress(
+                Path("tests/data").absolute(),
+                path.path / f"test.{compress_type}",
+                start=Path("tests/data").absolute(),
+                compress_type=compress_type,
+                compress_level=9,
+                allow_absolute_dest=True,
+            )
+            extract(
+                path.path / f"test.{compress_type}",
+                path.path / "extract",
+                compress_type=compress_type,
+                allow_absolute_dest=True,
+            )
+            self._check_extract_archive(path.path / "extract")
+
+    def test_compress_7z(self) -> None:
+        """Test compress 7-Zip."""
+        self._compress("7z")
+
+    def test_compress_zip(self) -> None:
+        """Test compress zip."""
+        self._compress("zip")
+
+    def test_compress_tar_gz(self) -> None:
+        """Test compress tar.gz."""
+        self._compress("tar.gz")
+
+    def test_compress_tar_xz(self) -> None:
+        """Test compress tar.xz."""
+        self._compress("tar.xz")
+
+    def test_compress_tgz(self) -> None:
+        """Test compress tgz (alias of tar.gz)."""
+        self._compress("tgz")
+
+    def test_compress_to_absolute(self) -> None:
+        """Test compress to absolute path."""
+        with TemporaryObject.new_directory(suffix="test") as path:
+            pytest.raises(
+                RUValueError,
+                compress,
+                Path("tests/data").absolute(),
+                (path.path / "test.zip").absolute(),
+                start=Path("tests/data"),
+                compress_level=9,
+                allow_absolute_dest=False,
+            )
+
+    def test_compress_invalid_type(self) -> None:
+        """Test compress invalid type."""
+        with TemporaryObject.new_directory(suffix="test") as path:
+            pytest.raises(
+                RUValueError,
+                compress,
+                Path("tests/data").absolute(),
+                path.path / "test.zip",
+                start=Path("tests/data"),
+                compress_type="invalid",
+                compress_level=9,
+                allow_absolute_dest=True,
+            )
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])

rubisco/lib/archive/__init__.py

@@ -91,7 +91,7 @@ def extract_tarball(
         )
 
         for member in memembers:
-            fp.extract(member, dest)
+            fp.extract(member, dest, filter=tarfile.tar_filter)
             call_ktrigger(
                 IKernelTrigger.on_progress,
                 task_name=task_name,

rubisco/lib/archive/tar.py

@@ -48,8 +48,15 @@ def command(args: list[str] | str) -> str:
     return res_command.strip()
 
 
+def test_command_generator() -> None:
+    """Test command generator."""
+    if command(["echo", "Hello, world!"]) != 'echo "Hello, world!"':
+        raise AssertionError
+    if command("echo Hello, world!") != "echo Hello, world!":
+        raise AssertionError
+
+
 if __name__ == "__main__":
-    _CMD = command(["echo", "Hello, world!"])
-    assert _CMD == 'echo "Hello, world!"'  # noqa: S101
-    _CMD = command("echo Hello, world!")
-    assert _CMD == "echo Hello, world!"  # noqa: S101
+    import pytest
+
+    pytest.main([__file__])