bl_pkg/tests/test_cli.py

   1 # SPDX-FileCopyrightText: 2023 Blender Foundation
   2 #
   3 # SPDX-License-Identifier: GPL-2.0-or-later
   4
   5 """
   6 Test with command:
   7    make test_blender BLENDER_BIN=$PWD/../../../blender.bin
   8 """
   9
  10 import json
  11 import os
  12 import shutil
  13 import subprocess
  14 import sys
  15 import tempfile
  16 import unittest
  17
  18 import unittest.util
  19
  20 from typing import (
  21     Any,
  22     Sequence,
  23     Dict,
  24     NamedTuple,
  25     Optional,
  26     Set,
  27     Tuple,
  28 )
  29
  30 # For more useful output that isn't clipped.
  31 unittest.util._MAX_LENGTH = 10_000
  32
  33 IS_WIN32 = sys.platform == "win32"
  34
  35 # See the variable with the same name in `blender_ext.py`.
  36 REMOTE_REPO_HAS_JSON_IMPLIED = True
  37
  38 PKG_EXT = ".zip"
  39
  40 # PKG_REPO_LIST_FILENAME = "bl_ext_repo.json"
  41 PKG_MANIFEST_FILENAME = "bl_ext_pkg_manifest.json"
  42
  43 PKG_MANIFEST_FILENAME_TOML = "blender_manifest.toml"
  44
  45 # Use an in-memory temp, when available.
  46 TEMP_PREFIX = tempfile.gettempdir()
  47 if os.path.exists("/ramcache/tmp"):
  48     TEMP_PREFIX = "/ramcache/tmp"
  49
  50 TEMP_DIR_REMOTE = os.path.join(TEMP_PREFIX, "bl_ext_remote")
  51 TEMP_DIR_LOCAL = os.path.join(TEMP_PREFIX, "bl_ext_local")
  52
  53 if TEMP_DIR_LOCAL and not os.path.isdir(TEMP_DIR_LOCAL):
  54     os.makedirs(TEMP_DIR_LOCAL)
  55 if TEMP_DIR_REMOTE and not os.path.isdir(TEMP_DIR_REMOTE):
  56     os.makedirs(TEMP_DIR_REMOTE)
  57
  58
  59 BASE_DIR = os.path.abspath(os.path.dirname(__file__))
  60 # PYTHON_CMD = sys.executable
  61
  62 CMD = (
  63     sys.executable,
  64     os.path.normpath(os.path.join(BASE_DIR, "..", "cli", "blender_ext.py")),
  65 )
  66
  67 # Simulate communicating with a web-server.
  68 USE_HTTP = os.environ.get("USE_HTTP", "0") != "0"
  69 HTTP_PORT = 8001
  70
  71 VERBOSE = os.environ.get("VERBOSE", "0") != "0"
  72
  73 sys.path.append(os.path.join(BASE_DIR, "modules"))
  74 from http_server_context import HTTPServerContext  # noqa: E402
  75
  76 STATUS_NON_ERROR = {'STATUS', 'PROGRESS'}
  77
  78
  79 # -----------------------------------------------------------------------------
  80 # Generic Utilities
  81 #
  82
  83
  84 def rmdir_contents(directory: str) -> None:
  85     """
  86     Remove all directory contents without removing the directory.
  87     """
  88     for entry in os.scandir(directory):
  89         filepath = os.path.join(directory, entry.name)
  90         if entry.is_dir():
  91             shutil.rmtree(filepath)
  92         else:
  93             os.unlink(filepath)
  94
  95
  96 # -----------------------------------------------------------------------------
  97 # HTTP Server (simulate remote access)
  98 #
  99
 100 # -----------------------------------------------------------------------------
 101 # Generate Repository
 102 #
 103
 104
 105 def my_create_package(dirpath: str, filename: str, *, metadata: Dict[str, Any], files: Dict[str, bytes]) -> None:
 106     """
 107     Create a package using the command line interface.
 108     """
 109     assert filename.endswith(PKG_EXT)
 110     outfile = os.path.join(dirpath, filename)
 111
 112     # NOTE: use the command line packaging utility to ensure 1:1 behavior with actual packages.
 113     metadata_copy = metadata.copy()
 114
 115     with tempfile.TemporaryDirectory() as temp_dir_pkg:
 116         temp_dir_pkg_manifest_toml = os.path.join(temp_dir_pkg, PKG_MANIFEST_FILENAME_TOML)
 117         with open(temp_dir_pkg_manifest_toml, "wb") as fh:
 118             # NOTE: escaping is not supported, this is primitive TOML writing for tests.
 119             data = "".join((
 120                 """# Example\n""",
 121                 """schema_version = "{:s}"\n""".format(metadata_copy.pop("schema_version")),
 122                 """id = "{:s}"\n""".format(metadata_copy.pop("id")),
 123                 """name = "{:s}"\n""".format(metadata_copy.pop("name")),
 124                 """tagline = "{:s}"\n""".format(metadata_copy.pop("tagline")),
 125                 """version = "{:s}"\n""".format(metadata_copy.pop("version")),
 126                 """type = "{:s}"\n""".format(metadata_copy.pop("type")),
 127                 """tags = [{:s}]\n""".format(", ".join("\"{:s}\"".format(v) for v in metadata_copy.pop("tags"))),
 128                 """blender_version_min = "{:s}"\n""".format(metadata_copy.pop("blender_version_min")),
 129                 """maintainer = "{:s}"\n""".format(metadata_copy.pop("maintainer")),
 130                 """license = [{:s}]\n""".format(", ".join("\"{:s}\"".format(v) for v in metadata_copy.pop("license"))),
 131             )).encode('utf-8')
 132             fh.write(data)
 133
 134         if metadata_copy:
 135             raise Exception("Unexpected mata-data: {!r}".format(metadata_copy))
 136
 137         for filename_iter, data in files.items():
 138             with open(os.path.join(temp_dir_pkg, filename_iter), "wb") as fh:
 139                 fh.write(data)
 140
 141         output_json = command_output_from_json_0(
 142             [
 143                 "build",
 144                 "--source-dir", temp_dir_pkg,
 145                 "--output-filepath", outfile,
 146             ],
 147             exclude_types={"PROGRESS"},
 148         )
 149
 150         output_json_error = command_output_filter_exclude(
 151             output_json,
 152             exclude_types=STATUS_NON_ERROR,
 153         )
 154
 155         if output_json_error:
 156             raise Exception("Creating a package produced some error output: {!r}".format(output_json_error))
 157
 158
 159 class PkgTemplate(NamedTuple):
 160     """Data need to create a package for testing."""
 161     idname: str
 162     name: str
 163     version: str
 164
 165
 166 def my_generate_repo(
 167         dirpath: str,
 168         *,
 169         templates: Sequence[PkgTemplate],
 170 ) -> None:
 171     for template in templates:
 172         my_create_package(
 173             dirpath, template.idname + PKG_EXT,
 174             metadata={
 175                 "schema_version": "1.0.0",
 176                 "id": template.idname,
 177                 "name": template.name,
 178                 "tagline": """This package has a tagline""",
 179                 "version": template.version,
 180                 "type": "add-on",
 181                 "tags": ["UV", "Modeling"],
 182                 "blender_version_min": "0.0.0",
 183                 "maintainer": "Some Developer",
 184                 "license": ["SPDX:GPL-2.0-or-later"],
 185             },
 186             files={
 187                 "__init__.py": b"# This is a script\n",
 188             },
 189         )
 190
 191
 192 def command_output_filter_include(
 193         output_json: Sequence[Tuple[str, Any]],
 194         include_types: Set[str],
 195 ) -> Sequence[Tuple[str, Any]]:
 196     return [(a, b) for a, b in output_json if a in include_types]
 197
 198
 199 def command_output_filter_exclude(
 200         output_json: Sequence[Tuple[str, Any]],
 201         exclude_types: Set[str],
 202 ) -> Sequence[Tuple[str, Any]]:
 203     return [(a, b) for a, b in output_json if a not in exclude_types]
 204
 205
 206 def command_output(
 207         args: Sequence[str],
 208         expected_returncode: int = 0,
 209 ) -> str:
 210     proc = subprocess.run(
 211         [*CMD, *args],
 212         stdout=subprocess.PIPE,
 213         check=expected_returncode == 0,
 214     )
 215     if proc.returncode != expected_returncode:
 216         raise subprocess.CalledProcessError(proc.returncode, proc.args, output=proc.stdout, stderr=proc.stderr)
 217     result = proc.stdout.decode("utf-8")
 218     if IS_WIN32:
 219         result = result.replace("\r\n", "\n")
 220     return result
 221
 222
 223 def command_output_from_json_0(
 224         args: Sequence[str],
 225         *,
 226         exclude_types: Optional[Set[str]] = None,
 227         expected_returncode: int = 0,
 228 ) -> Sequence[Tuple[str, Any]]:
 229     result = []
 230
 231     proc = subprocess.run(
 232         [*CMD, *args, "--output-type=JSON_0"],
 233         stdout=subprocess.PIPE,
 234         check=expected_returncode == 0,
 235     )
 236     if proc.returncode != expected_returncode:
 237         raise subprocess.CalledProcessError(proc.returncode, proc.args, output=proc.stdout, stderr=proc.stderr)
 238     for json_bytes in proc.stdout.split(b'\0'):
 239         if not json_bytes:
 240             continue
 241         json_str = json_bytes.decode("utf-8")
 242         json_data = json.loads(json_str)
 243         assert len(json_data) == 2
 244         assert isinstance(json_data[0], str)
 245         if (exclude_types is not None) and (json_data[0] in exclude_types):
 246             continue
 247         result.append((json_data[0], json_data[1]))
 248
 249     return result
 250
 251
 252 class TestCLI(unittest.TestCase):
 253
 254     def test_version(self) -> None:
 255         self.assertEqual(command_output(["--version"]), "0.1\n")
 256
 257
 258 class TestCLI_WithRepo(unittest.TestCase):
 259     dirpath = ""
 260     dirpath_url = ""
 261
 262     @classmethod
 263     def setUpClass(cls) -> None:
 264         if TEMP_DIR_REMOTE:
 265             cls.dirpath = TEMP_DIR_REMOTE
 266             if os.path.isdir(cls.dirpath):
 267                 # pylint: disable-next=using-constant-test
 268                 if False:
 269                     shutil.rmtree(cls.dirpath)
 270                     os.makedirs(TEMP_DIR_REMOTE)
 271                 else:
 272                     # Empty the path without removing it,
 273                     # handy so a developer can remain in the directory.
 274                     rmdir_contents(TEMP_DIR_REMOTE)
 275             else:
 276                 os.makedirs(TEMP_DIR_REMOTE)
 277         else:
 278             cls.dirpath = tempfile.mkdtemp(prefix="bl_ext_")
 279
 280         my_generate_repo(
 281             cls.dirpath,
 282             templates=(
 283                 PkgTemplate(idname="foo_bar", name="Foo Bar", version="1.0.5"),
 284                 PkgTemplate(idname="another_package", name="Another Package", version="1.5.2"),
 285                 PkgTemplate(idname="test_package", name="Test Package", version="1.5.2"),
 286             ),
 287         )
 288
 289         if USE_HTTP:
 290             if REMOTE_REPO_HAS_JSON_IMPLIED:
 291                 cls.dirpath_url = "http://localhost:{:d}/bl_ext_repo.json".format(HTTP_PORT)
 292             else:
 293                 cls.dirpath_url = "http://localhost:{:d}".format(HTTP_PORT)
 294         else:
 295             cls.dirpath_url = cls.dirpath
 296
 297     @classmethod
 298     def tearDownClass(cls) -> None:
 299         if not TEMP_DIR_REMOTE:
 300             shutil.rmtree(cls.dirpath)
 301         del cls.dirpath
 302         del cls.dirpath_url
 303
 304     def test_version(self) -> None:
 305         self.assertEqual(command_output(["--version"]), "0.1\n")
 306
 307     def test_server_generate(self) -> None:
 308         output = command_output(["server-generate", "--repo-dir", self.dirpath])
 309         self.assertEqual(output, "found 3 packages.\n")
 310
 311     def test_client_list(self) -> None:
 312         # TODO: only run once.
 313         self.test_server_generate()
 314
 315         output = command_output(["list", "--remote-url", self.dirpath_url, "--local-dir", ""])
 316         self.assertEqual(
 317             output, (
 318                 "another_package(1.5.2): Another Package\n"
 319                 "foo_bar(1.0.5): Foo Bar\n"
 320                 "test_package(1.5.2): Test Package\n"
 321             )
 322         )
 323         del output
 324
 325         # TODO, figure out how to split JSON & TEXT output tests, this test just checks JSON is working at all.
 326         output_json = command_output_from_json_0(
 327             ["list", "--remote-url", self.dirpath_url, "--local-dir", ""],
 328             exclude_types={"PROGRESS"},
 329         )
 330         self.assertEqual(
 331             output_json, [
 332                 ("STATUS", "another_package(1.5.2): Another Package"),
 333                 ("STATUS", "foo_bar(1.0.5): Foo Bar"),
 334                 ("STATUS", "test_package(1.5.2): Test Package"),
 335             ]
 336         )
 337
 338     def test_client_install_and_uninstall(self) -> None:
 339         with tempfile.TemporaryDirectory(dir=TEMP_DIR_LOCAL) as temp_dir_local:
 340             # TODO: only run once.
 341             self.test_server_generate()
 342
 343             output_json = command_output_from_json_0([
 344                 "sync",
 345                 "--remote-url", self.dirpath_url,
 346                 "--local-dir", temp_dir_local,
 347             ], exclude_types={"PROGRESS"})
 348             self.assertEqual(
 349                 output_json, [
 350                     ('STATUS', 'Sync repo: ' + self.dirpath_url),
 351                     ('STATUS', 'Sync downloading remote data'),
 352                     ('STATUS', 'Sync complete: ' + self.dirpath_url),
 353                 ]
 354             )
 355
 356             # Install.
 357             output_json = command_output_from_json_0(
 358                 [
 359                     "install", "another_package",
 360                     "--remote-url", self.dirpath_url,
 361                     "--local-dir", temp_dir_local,
 362                 ],
 363                 exclude_types={"PROGRESS"},
 364             )
 365             self.assertEqual(
 366                 output_json, [
 367                     ("STATUS", "Installed \"another_package\"")
 368                 ]
 369             )
 370             self.assertTrue(os.path.isdir(os.path.join(temp_dir_local, "another_package")))
 371
 372             # Re-Install.
 373             output_json = command_output_from_json_0(
 374                 [
 375                     "install", "another_package",
 376                     "--remote-url", self.dirpath_url,
 377                     "--local-dir", temp_dir_local,
 378                 ],
 379                 exclude_types={"PROGRESS"},
 380             )
 381             self.assertEqual(
 382                 output_json, [
 383                     ("STATUS", "Re-Installed \"another_package\"")
 384                 ]
 385             )
 386             self.assertTrue(os.path.isdir(os.path.join(temp_dir_local, "another_package")))
 387
 388             # Uninstall (not found).
 389             output_json = command_output_from_json_0(
 390                 [
 391                     "uninstall", "another_package_",
 392                     "--local-dir", temp_dir_local,
 393                 ],
 394                 expected_returncode=1,
 395             )
 396             self.assertEqual(
 397                 output_json, [
 398                     ("ERROR", "Package not found \"another_package_\"")
 399                 ]
 400             )
 401
 402             # Uninstall.
 403             output_json = command_output_from_json_0([
 404                 "uninstall", "another_package",
 405                 "--local-dir", temp_dir_local,
 406             ])
 407             self.assertEqual(
 408                 output_json, [
 409                     ("STATUS", "Removed \"another_package\"")
 410                 ]
 411             )
 412             self.assertFalse(os.path.isdir(os.path.join(temp_dir_local, "another_package")))
 413
 414
 415 if __name__ == "__main__":
 416     if USE_HTTP:
 417         with HTTPServerContext(directory=TEMP_DIR_REMOTE, port=HTTP_PORT):
 418             unittest.main()
 419     else:
 420         unittest.main()