diff --git a/src/backend/InvenTree/tests/test_version_module.py b/src/backend/InvenTree/tests/test_version_module.py new file mode 100644 index 000000000000..b81173aef62d --- /dev/null +++ b/src/backend/InvenTree/tests/test_version_module.py @@ -0,0 +1,358 @@ +"""Test suite for the version module in InvenTree backend.""" + +import os +import importlib +import types +from datetime import datetime as dt, timedelta as td + +import sys + +import pytest + +# We import the module under test after preparing environment or monkeypatch as needed in tests +MODULE_PATH = "src.backend.InvenTree.InvenTree.test_version".replace("/", ".").replace("\\\\", ".") + + +def reload_module(monkeypatch, extra_patches=None): + """Reload module under test with optional extra patches applied before import. + + Ensures top-level import side effects (e.g., dulwich presence) are stabilized. + """ + if extra_patches: + for func in extra_patches: + func() + # Remove from sys.modules to re-execute top-level code paths + if MODULE_PATH in sys.modules: + del sys.modules[MODULE_PATH] + return importlib.import_module(MODULE_PATH) + + +class DummyCommit: + """Dummy commit object for testing.""" + + def __init__(self, hexstr="abcdef1234567", ts=1700000000, tz=0): + """Initialize DummyCommit with hex string, timestamp, and timezone.""" + self._hex = hexstr + self.commit_time = ts + self.commit_timezone = tz + + def sha(self): + """Return object with hexdigest method returning the hex string.""" + class _S: + def __init__(self, h): self._h = h + def hexdigest(self): return self._h + return _S(self._hex) + + +def test_inventreeVersion_and_tuple_and_dev_detection(monkeypatch): + """Test inventreeVersion, inventreeVersionTuple, and development detection.""" + # Force a specific SW version + def patch_version(): + mod = types.SimpleNamespace() + return mod + mod = reload_module(monkeypatch) + # Set to a development version + mod.INVENTREE_SW_VERSION = "1.2.3 dev" + assert mod.inventreeVersion() == "1.2.3 dev" + assert mod.inventreeVersionTuple() == [1, 2, 3] + assert mod.isInvenTreeDevelopmentVersion() is True + + # Set to a release version + mod.INVENTREE_SW_VERSION = "2.10.0" + assert mod.inventreeVersion() == "2.10.0" + assert mod.inventreeVersionTuple() == [2, 10, 0] + assert mod.isInvenTreeDevelopmentVersion() is False + + +def test_inventreeDocs_and_urls(monkeypatch): + """Test inventreeDocsVersion, inventreeDocUrl, inventreeAppUrl, and inventreeGithubUrl.""" + mod = reload_module(monkeypatch) + # Dev version -> docs "latest" + mod.INVENTREE_SW_VERSION = "0.18.0 dev" + assert mod.inventreeDocsVersion() == "latest" + assert mod.inventreeDocUrl().endswith("/en/latest") + + # Release version -> returns exact version (no cover in code originally) + mod.INVENTREE_SW_VERSION = "0.18.0" + # inventreeDocsVersion will return INVENTREE_SW_VERSION + assert mod.inventreeDocsVersion() == "0.18.0" + assert mod.inventreeDocUrl().endswith("/en/0.18.0") + + # Static URLs + assert mod.inventreeAppUrl() == "https://docs.inventree.org/app/" + assert mod.inventreeGithubUrl() == "https://github.com/InvenTree/InvenTree/" + + +def test_checkMinPythonVersion_accepts_39_plus(monkeypatch, capsys): + """Test checkMinPythonVersion accepts Python >= 3.9.""" + mod = reload_module(monkeypatch) + # Simulate Python 3.11 + monkeypatch.setattr(sys, "version_info", types.SimpleNamespace(major=3, minor=11, micro=0)) + monkeypatch.setattr(sys, "version", "3.11.0 (main, Jan 1 2024, 00:00:00)") + monkeypatch.setattr(sys, "executable", "/usr/bin/python3.11") + # Should not raise + mod.checkMinPythonVersion() + out = capsys.readouterr().out + assert "Python version 3.11.0" in out + assert "/usr/bin/python3.11" in out + + +@pytest.mark.parametrize( + "major,minor,should_raise", + [ + (2, 7, True), # Python2 -> raise + (3, 8, True), # 3.8 -> raise + (3, 9, False), # 3.9 -> ok + (3, 10, False), # 3.10 -> ok + ], +) +def test_checkMinPythonVersion_version_boundaries(monkeypatch, major, minor, should_raise): + """Test checkMinPythonVersion raises or accepts based on version boundaries.""" + mod = reload_module(monkeypatch) + monkeypatch.setattr(sys, "version_info", types.SimpleNamespace(major=major, minor=minor, micro=0)) + monkeypatch.setattr(sys, "version", f"{major}.{minor}.0 (mock)") + monkeypatch.setattr(sys, "executable", "/usr/bin/pythonX") + if should_raise: + with pytest.raises(RuntimeError) as exc: + mod.checkMinPythonVersion() + assert "InvenTree requires Python 3.9 or above" in str(exc.value) + else: + mod.checkMinPythonVersion() # no exception + + +def test_inventreeApiVersion_and_text_parsing(monkeypatch): + """Test parse_version_text, inventreeApiText, and version data parsing.""" + # Patch API constants to a controlled small sample + SAMPLE_TEXT = """ +v3 -> 2024-06-01: https://github.com/InvenTree/InvenTree/releases/tag/v3 +- Added endpoints for widgets +- Fixed bug A + +v2 -> 2024-05-01: https://github.com/InvenTree/InvenTree/releases/tag/v2 +- Improvements + +v1 +- Initial +""".lstrip("\n") + + def apply_patches(): + pass + + mod = reload_module(monkeypatch) + # Set constants and re-parse + mod.INVENTREE_API_VERSION = 3 + mod.INVENTREE_API_TEXT = SAMPLE_TEXT + parsed = mod.parse_version_text() + assert isinstance(parsed, dict) + # v3 entry + assert "v3" in parsed + assert parsed["v3"]["version"] == "v3" + assert parsed["v3"]["date"] == "2024-06-01" + assert parsed["v3"]["gh"].startswith("https://github.com/") + assert parsed["v3"]["latest"] is True # latest should match current API version + assert "- Added endpoints for widgets" in parsed["v3"]["text"][0] + # v2 entry + assert parsed["v2"]["date"] == "2024-05-01" + assert parsed["v2"]["latest"] is False + # v1 without details in header (date, gh) + assert parsed["v1"]["date"] == "" + assert parsed["v1"]["gh"] is None + # Set preprocessed cache and fetch slices + mod.INVENTREE_API_TEXT_DATA = parsed + + # Fetch default 10 versions from start_version computed as API_VERSION - versions + 1 + res = mod.inventreeApiText(versions=2) # Given API_VERSION=3 -> start 2..3 + assert list(res.keys()) == ["v2", "v3"] + assert res["v3"]["latest"] is True + assert res["v2"]["latest"] is False + + # Explicit start version + res2 = mod.inventreeApiText(versions=2, start_version=1) + assert list(res2.keys()) == ["v1", "v2"] + + +def test_inventreePython_and_django_version(monkeypatch): + """Test inventreeDjangoVersion and inventreePythonVersion.""" + mod = reload_module(monkeypatch) + # Django version comes from django.get_version(); patch django module attribute + class DummyDjango: + @staticmethod + def get_version(): + return "5.0.1" + monkeypatch.setitem(sys.modules, "django", DummyDjango) + # re-import to bind new django + if MODULE_PATH in sys.modules: + del sys.modules[MODULE_PATH] + mod = importlib.import_module(MODULE_PATH) + + assert mod.inventreeDjangoVersion() == "5.0.1" + # python version is split off first token + monkeypatch.setattr(sys, "version", "3.12.2 (main...)") + assert mod.inventreePythonVersion() == "3.12.2" + + +def test_commit_hash_branch_date_from_env(monkeypatch): + """Test inventreeCommitHash, inventreeCommitDate, and inventreeBranch with env and DummyCommit.""" + mod = reload_module(monkeypatch) + + # Env overrides should take precedence + monkeypatch.setenv("INVENTREE_COMMIT_HASH", "1234abc") + monkeypatch.setenv("INVENTREE_COMMIT_DATE", "2024-07-04 12:00:00") + monkeypatch.setenv("INVENTREE_PKG_BRANCH", "feature/x") + assert mod.inventreeCommitHash() == "1234abc" + assert mod.inventreeCommitDate() == "2024-07-04" + assert mod.inventreeBranch() == "feature/x" + + # Clear env and simulate no dulwich (main_commit / main_branch None) + monkeypatch.delenv("INVENTREE_COMMIT_HASH", raising=False) + monkeypatch.delenv("INVENTREE_COMMIT_DATE", raising=False) + monkeypatch.delenv("INVENTREE_PKG_BRANCH", raising=False) + mod.main_commit = None + mod.main_branch = None + assert mod.inventreeCommitHash() is None + assert mod.inventreeCommitDate() is None + assert mod.inventreeBranch() is None + + # Provide dummy commit/branch + mod.main_commit = DummyCommit("deadbeefcafeee", ts=1710000000, tz=0) + # main_branch is expected bytes + mod.main_branch = b"main" + assert mod.inventreeCommitHash() == "deadbee" # first 7 chars + # commit_date is derived from timestamp + timezone + expected_date = str((dt.fromtimestamp(1710000000) + td(seconds=0)).date()) + assert mod.inventreeCommitDate() == expected_date + assert mod.inventreeBranch() == "main" + + +def test_inventreeTarget_and_platform(monkeypatch): + """Test inventreeTarget and inventreePlatform.""" + mod = reload_module(monkeypatch) + # Target from env or None + monkeypatch.delenv("INVENTREE_PKG_TARGET", raising=False) + assert mod.inventreeTarget() is None + + monkeypatch.setenv("INVENTREE_PKG_TARGET", "linux/amd64") + assert mod.inventreeTarget() == "linux/amd64" + + # Platform delegates to platform.platform(aliased=True) - mock + class DPlat: + @staticmethod + def platform(aliased=False): # signature compatibility + assert aliased is True + return "Linux-6.8.0-x86_64-with-glibc2.39" + monkeypatch.setitem(sys.modules, "platform", DPlat) + if MODULE_PATH in sys.modules: + del sys.modules[MODULE_PATH] + mod = importlib.import_module(MODULE_PATH) + assert mod.inventreePlatform() == "Linux-6.8.0-x86_64-with-glibc2.39" + + +def test_inventreeDatabase_reads_settings(monkeypatch): + """Test inventreeDatabase reads DB_ENGINE from django settings.""" + # Mock django.conf.settings object to carry DB_ENGINE + class DummySettings: + DB_ENGINE = "postgresql" + class DummyConf: + settings = DummySettings + # Inject django.conf.settings before import + dummy_django = types.SimpleNamespace(conf=DummyConf, get_version=lambda: "5.0.0") + monkeypatch.setitem(sys.modules, "django", dummy_django) + if MODULE_PATH in sys.modules: + del sys.modules[MODULE_PATH] + mod = importlib.import_module(MODULE_PATH) + + assert mod.inventreeDatabase() == "postgresql" + + +def test_inventree_identifier_logic(monkeypatch): + """Test inventree_identifier, inventreeInstanceName, and inventreeInstanceTitle logic.""" + mod = reload_module(monkeypatch) + + # Simulate common.settings.get_global_setting without importing real module + # We monkeypatch the function the module imports in the function scope by injecting a module + # named common.settings that provides get_global_setting. + calls = {"args": []} + def fake_get_global_setting(key, *args, **kwargs): + calls["args"].append((key, args, kwargs)) + # Provide behavior depending on key + if key == "INVENTREE_ANNOUNCE_ID": + # Only true if env key present + return kwargs.get("enviroment_key") == "INVENTREE_ANNOUNCE_ID" and os.getenv("INVENTREE_ANNOUNCE_ID") == "1" + if key == "INVENTREE_INSTANCE_ID": + return "instance-xyz" + if key == "INVENTREE_INSTANCE": + return "MyInstance" + if key == "INVENTREE_INSTANCE_TITLE": + return "" + return None + + common_pkg = types.ModuleType("common") + settings_mod = types.ModuleType("common.settings") + settings_mod.get_global_setting = fake_get_global_setting + + monkeypatch.setitem(sys.modules, "common", common_pkg) + monkeypatch.setitem(sys.modules, "common.settings", settings_mod) + + # No env -> announce flag False -> None + monkeypatch.delenv("INVENTREE_ANNOUNCE_ID", raising=False) + assert mod.inventree_identifier() is None + + # With override -> always returns instance id + assert mod.inventree_identifier(override_announce=True) == "instance-xyz" + + # With env var -> announce true -> returns instance id + monkeypatch.setenv("INVENTREE_ANNOUNCE_ID", "1") + assert mod.inventree_identifier() == "instance-xyz" + + # Also validate instance name and title helpers + assert mod.inventreeInstanceName() == "MyInstance" + # INVENTREE_INSTANCE_TITLE empty -> default "InvenTree" + assert mod.inventreeInstanceTitle() == "InvenTree" + + # If a title is present, function returns the instance name + def fake_get_global_setting_title(key, *args, **kwargs): + if key == "INVENTREE_INSTANCE_TITLE": + return "ShownTitle" + if key == "INVENTREE_INSTANCE": + return "NameFromInstance" + if key == "INVENTREE_ANNOUNCE_ID": + return False + if key == "INVENTREE_INSTANCE_ID": + return "instance-xyz" + return None + settings_mod.get_global_setting = fake_get_global_setting_title + assert mod.inventreeInstanceTitle() == "NameFromInstance" + + +def test_isInvenTreeUpToDate(monkeypatch): + """Test isInvenTreeUpToDate behavior with global setting.""" + mod = reload_module(monkeypatch) + + # Patch get_global_setting + def ggs(key, backup_value=None, create=True, **kwargs): + assert key == "_INVENTREE_LATEST_VERSION" + return None # simulate absence -> must return True + common_pkg = types.ModuleType("common") + settings_mod = types.ModuleType("common.settings") + settings_mod.get_global_setting = ggs + monkeypatch.setitem(sys.modules, "common", common_pkg) + monkeypatch.setitem(sys.modules, "common.settings", settings_mod) + # Must assume up-to-date if no record + assert mod.isInvenTreeUpToDate() is True + + # Provide an older latest version to compare tuples + def ggs2(key, backup_value=None, create=True, **kwargs): + return "0.1.0" + settings_mod.get_global_setting = ggs2 + # With SW version 0.2.0, >= 0.1.0 -> True + mod.INVENTREE_SW_VERSION = "0.2.0" + assert mod.isInvenTreeUpToDate() is True + + # Provide higher latest version + def ggs3(key, backup_value=None, create=True, **kwargs): + return "9.9.9" + settings_mod.get_global_setting = ggs3 + mod.INVENTREE_SW_VERSION = "1.0.0" + # inventree_version (1,0,0) >= latest (9,9,9) -> False + # The comparison is in a no-cover path originally; still exercise it here. + assert mod.isInvenTreeUpToDate() is False \ No newline at end of file diff --git a/tests/backend/plugin/test_installer_test.py b/tests/backend/plugin/test_installer_test.py new file mode 100644 index 000000000000..d4e6265e2fc8 --- /dev/null +++ b/tests/backend/plugin/test_installer_test.py @@ -0,0 +1,504 @@ +"""Tests for the InvenTree plugin test_installer module.""" +import re +import sys +import types +from pathlib import Path + +import pytest + +# Import the module under test +# The module is named "test_installer" but it is not itself a test; it's a production module. +from InvenTree.plugin import test_installer as installer + + +try: + # If pytest-django is present, a 'settings' fixture will be available. + HAVE_DJANGO_SETTINGS_FIXTURE = True +except Exception: + HAVE_DJANGO_SETTINGS_FIXTURE = False + + +class DummyUser: + """Dummy user with is_staff attribute for testing.""" + + def __init__(self, is_staff: bool): + """Initialize DummyUser with staff flag.""" + self.is_staff = is_staff + + +class DummyPluginConfig: + """Dummy plugin configuration to simulate plugin settings.""" + + def __init__( + self, + key: str = "dummy", + active: bool = False, + package_name: str = "dummy-plugin", + is_package: bool = True, + installed: bool = True, + mandatory: bool = False, + sample: bool = False, + builtin: bool = False, + ): + """Initialize DummyPluginConfig with optional parameters.""" + self.key = key + self.active = active + self.package_name = package_name + self._is_package = is_package + self._installed = installed + self._mandatory = mandatory + self._sample = sample + self._builtin = builtin + + # These reflect the plugin.models.PluginConfig interface used by uninstall logic + + def is_package(self): + """Return True if plugin is a package.""" + return self._is_package + + def is_installed(self): + """Return True if plugin is installed.""" + return self._installed + + def is_mandatory(self): + """Return True if plugin is mandatory.""" + return self._mandatory + + def is_sample(self): + """Return True if plugin is a sample.""" + return self._sample + + def is_builtin(self): + """Return True if plugin is built-in.""" + return self._builtin + + def delete(self): + """Simulate deletion; mark as deleted.""" + self._deleted = True + + +@pytest.fixture +def fake_settings(tmp_path, monkeypatch): + """Provide a minimal settings object/namespace sufficient for the module's usage. + + - BASE_DIR.parent for pip_command cwd + - PLUGIN_FILE path used by plugins file functions + - PLUGINS_INSTALL_DISABLED flag gates install/uninstall. + """ + class _Settings: + def __init__(self): + self.BASE_DIR = tmp_path / "src" + self.BASE_DIR.mkdir(parents=True, exist_ok=True) + self.PLUGIN_FILE = tmp_path / "plugins.txt" + self.PLUGINS_INSTALL_DISABLED = False + + s = _Settings() + + # Patch django.conf.settings + from types import SimpleNamespace + + django_settings = SimpleNamespace( + BASE_DIR=s.BASE_DIR, + PLUGIN_FILE=s.PLUGIN_FILE, + PLUGINS_INSTALL_DISABLED=s.PLUGINS_INSTALL_DISABLED, + ) + # Inject into module import path for django.conf.settings usage + # We monkeypatch the settings used inside the module under test + monkeypatch.setattr(installer, "settings", django_settings, raising=True) + return django_settings + + +@pytest.fixture +def mock_log_error(monkeypatch): + """Provide a list to capture log_error calls.""" + calls = [] + + def _log_error(path: str, scope: str = ""): + calls.append((path, scope)) + + monkeypatch.setattr(installer, "log_error", _log_error, raising=True) + return calls + + +@pytest.fixture +def mock_staticfiles(monkeypatch): + """Provide hooks to capture staticfiles plugin collect and clear calls.""" + calls = {"collect": 0, "clear": []} + + class _Static: + @staticmethod + def collect_plugins_static_files(): + calls["collect"] += 1 + + @staticmethod + def clear_plugin_static_files(key): + calls["clear"].append(key) + + monkeypatch.setattr(installer, "plugin", types.SimpleNamespace(staticfiles=_Static), raising=False) + return calls + + +@pytest.fixture +def mock_registry(monkeypatch): + """Provide a dummy plugin registry to capture reload calls.""" + calls = {"reload": []} + + class _Registry: + @staticmethod + def reload_plugins(full_reload=False, force_reload=False, collect=False): + calls["reload"].append((full_reload, force_reload, collect)) + + # The code performs: from plugin.registry import registry + # We monkeypatch the import path by creating a dummy module and attribute. + registry_module = types.SimpleNamespace(registry=_Registry()) + sys.modules["plugin.registry"] = registry_module + return calls + + +@pytest.fixture +def mock_subprocess(monkeypatch): + """Provide hooks for subprocess.check_output to simulate success/failure.""" + class Proc: + def __init__(self): + self.calls = [] + self.raise_error = None + self.next_output = None + + def check_output(self, cmd, cwd=None, stderr=None): + self.calls.append({"cmd": cmd, "cwd": cwd, "stderr": stderr}) + if self.raise_error is not None: + e = subprocess.CalledProcessError( + returncode=self.raise_error.get("returncode", 1), + cmd=cmd, + output=self.raise_error.get("output", b"pip failed\nreason"), + ) + raise e + return self.next_output if self.next_output is not None else b"OK" + + import subprocess + + proc = Proc() + monkeypatch.setattr(installer.subprocess, "check_output", proc.check_output, raising=True) + return proc + + +@pytest.fixture +def mock_get_constraint_file(monkeypatch, tmp_path): + """Provide a temporary constraint file and monkeypatch get_constraint_file.""" + cf = tmp_path / "constraints.txt" + cf.write_text("# constraints\n") + monkeypatch.setattr(installer, "get_constraint_file", lambda: cf, raising=True) + return cf + + +def test_pip_command_builds_correct_invocation(fake_settings, mock_subprocess, monkeypatch): + """Test that pip_command builds the correct invocation.""" + out = installer.pip_command("show", "somepkg") + assert out == b"OK" + assert len(mock_subprocess.calls) == 1 + call = mock_subprocess.calls[0] + assert call["cmd"][0] == "/usr/bin/pythonX" + assert call["cmd"][1:3] == ["-m", "pip"] + assert call["cmd"][-2:] == ["show", "somepkg"] + assert Path(call["cwd"]) == fake_settings.BASE_DIR.parent + + +def test_handle_pip_error_single_line_raises_validationerror(mock_log_error, monkeypatch): + """Test that handle_pip_error raises ValidationError for single-line output.""" + import subprocess + + class E(subprocess.CalledProcessError): + pass + + err = subprocess.CalledProcessError(1, ["pip"], output=b"Only one error line") + with pytest.raises(installer.ValidationError) as ex: + installer.handle_pip_error(err, "plugin_install") + assert ("plugin_install", "pip") in mock_log_error + assert "Only one error line" in str(ex.value) + + +def test_handle_pip_error_multiple_lines_list_raises_validationerror(mock_log_error): + """Test that handle_pip_error raises ValidationError for multiple-line output.""" + import subprocess + + err = subprocess.CalledProcessError(1, ["pip"], output=b"line1\nline2\n\n") + with pytest.raises(installer.ValidationError) as ex: + installer.handle_pip_error(err, "plugin_install") + assert ("plugin_install", "pip") in mock_log_error + assert "line1" in str(ex.value) and "line2" in str(ex.value) + + +def test_get_install_info_success_parses_output(fake_settings, mock_subprocess): + """Test that get_install_info parses pip show output successfully.""" + mock_subprocess.next_output = b"Name: testpkg\nVersion: 1.2.3\nLocation: /some/path\n" + info = installer.get_install_info("testpkg==1.0") + assert info["name"] == "testpkg" + assert info["version"] == "1.2.3" + assert info["location"] == "/some/path" + + +def test_get_install_info_called_process_error_sets_error(fake_settings, mock_subprocess, mock_log_error): + """Test that get_install_info handles CalledProcessError and logs an error.""" + mock_subprocess.raise_error = {"returncode": 1, "output": b"not found"} + info = installer.get_install_info("missing") + assert "error" in info + assert "not found" in info["error"] + assert any(path == "get_install_info" and scope == "pip" for (path, scope) in mock_log_error) + + +def test_plugins_file_hash_none_when_absent(fake_settings, mock_log_error): + """Test that plugins_file_hash returns None when the plugin file is absent.""" + assert installer.plugins_file_hash() is None + + +def test_plugins_file_hash_computed(fake_settings, mock_log_error): + """Test that plugins_file_hash computes the correct hash of the plugin file.""" + content = b"pkgA==1.0\n# comment\n" + fake_settings.PLUGIN_FILE.write_bytes(content) + h = installer.plugins_file_hash() + import hashlib + assert h == hashlib.sha256(content).hexdigest() + + +def test_install_plugins_file_success(fake_settings, mock_subprocess, mock_staticfiles): + """Test that install_plugins_file installs plugins and collects static files successfully.""" + fake_settings.PLUGIN_FILE.write_text("pkgA==1.0\n") + result = installer.install_plugins_file() + assert result is True + call = mock_subprocess.calls[0] + cmd = call["cmd"] + assert cmd[3:] == ["install", "--disable-pip-version-check", "-U", "-r", str(fake_settings.PLUGIN_FILE)] + assert mock_staticfiles["collect"] == 1 + + +def test_install_plugins_file_missing_file_returns_none(fake_settings, mock_subprocess, mock_staticfiles, monkeypatch): + """Test that install_plugins_file returns None when the plugin file is missing.""" + fake_settings.PLUGIN_FILE = None + monkeypatch.setattr(installer, "settings", fake_settings, raising=True) + result = installer.install_plugins_file() + assert result is None + assert mock_staticfiles["collect"] == 0 + + +def test_install_plugins_file_pip_error_returns_false(fake_settings, mock_subprocess): + """Test that install_plugins_file returns False on pip error.""" + fake_settings.PLUGIN_FILE.write_text("pkgA==1.0\n") + mock_subprocess.raise_error = {"returncode": 1, "output": b"pip failed"} + assert installer.install_plugins_file() is False + + +def test_update_plugins_file_add_new(fake_settings): + """Test that update_plugins_file adds a new plugin entry.""" + fake_settings.PLUGIN_FILE.write_text("") + installer.update_plugins_file("myplugin") + txt = fake_settings.PLUGIN_FILE.read_text() + assert re.search(r"(?m)^myplugin\s*$", txt) + + +def test_update_plugins_file_replace_existing_and_preserve_comments(fake_settings): + """Test that update_plugins_file replaces existing plugin and preserves comments.""" + fake_settings.PLUGIN_FILE.write_text("# header\nmyplugin==0.1\nother\n") + installer.update_plugins_file("myplugin", version="2.0.0") + txt = fake_settings.PLUGIN_FILE.read_text().splitlines() + assert txt[0].startswith("# header") + assert "myplugin==2.0.0" in txt + assert "other" in txt + + +def test_update_plugins_file_remove_matching(fake_settings): + """Test that update_plugins_file removes matching plugin entries when remove is True.""" + fake_settings.PLUGIN_FILE.write_text("myplugin==1.0\nother\n") + installer.update_plugins_file("myplugin", remove=True) + assert "myplugin" not in fake_settings.PLUGIN_FILE.read_text() + assert "other" in fake_settings.PLUGIN_FILE.read_text() + + +def test_install_plugin_disallowed_for_non_staff(monkeypatch): + """Test that install_plugin raises ValidationError for non-staff users.""" + user = DummyUser(is_staff=False) + with pytest.raises(installer.ValidationError): + installer.install_plugin(packagename="pkg", user=user) + + +def test_install_plugin_disabled_by_setting(fake_settings, monkeypatch): + """Test that install_plugin raises ValidationError when installation is disabled by settings.""" + fake_settings.PLUGINS_INSTALL_DISABLED = True + monkeypatch.setattr(installer, "settings", fake_settings, raising=True) + with pytest.raises(installer.ValidationError): + installer.install_plugin(packagename="pkg", user=DummyUser(True)) + + +@pytest.mark.parametrize( + "url,packagename,version,expected_full,extra_flags", + [ + (None, "pkg", None, "pkg", []), + (None, "pkg", "1.2.3", "pkg==1.2.3", []), + ("https://index.example/simple", None, None, "https://index.example/simple", ["-i"]), + ("git+https://github.com/x/y.git", "pkg", None, "pkg@git+https://github.com/x/y.git", []), + ("git+https://github.com/x/y.git", None, None, "git+https://github.com/x/y.git", []), + ], +) +def test_install_plugin_command_building( + fake_settings, mock_subprocess, mock_get_constraint_file, url, packagename, version, expected_full, extra_flags, mock_registry, mock_staticfiles +): + """Test that install_plugin builds the correct pip install command for various inputs.""" + def fake_get_install_info(name): + return {"location": "/venv/site-packages", "version": "9.9.9"} + + orig_get_install_info = installer.get_install_info + installer.get_install_info = fake_get_install_info + + try: + ret = installer.install_plugin(url=url, packagename=packagename, user=DummyUser(True), version=version) + assert "success" in ret + call = mock_subprocess.calls[0] + cmd = call["cmd"] + assert cmd[:3] == [sys.executable, "-m", "pip"] + assert cmd[3:9] == [ + "install", + "-U", + "--disable-pip-version-check", + "-c", + str(mock_get_constraint_file), + ] + if extra_flags: + assert extra_flags[0] in cmd + assert cmd[-1] == expected_full + assert mock_registry["reload"] + finally: + installer.get_install_info = orig_get_install_info + + +def test_install_plugin_pip_error_raises_validationerror(fake_settings, mock_subprocess, mock_get_constraint_file): + """Test that install_plugin raises ValidationError on pip error.""" + mock_subprocess.raise_error = {"returncode": 1, "output": b"pip exploded"} + with pytest.raises(installer.ValidationError): + installer.install_plugin(packagename="pkg", user=DummyUser(True)) + + +def test_validate_package_plugin_checks(monkeypatch): + """Test that validate_package_plugin enforces package plugin configuration rules.""" + cfg = DummyPluginConfig() + cfg.plugin = None + with pytest.raises(installer.ValidationError): + installer.validate_package_plugin(cfg) + cfg.plugin = object() + cfg._is_package = False + with pytest.raises(installer.ValidationError): + installer.validate_package_plugin(cfg) + cfg._is_package = True + cfg.package_name = "" + with pytest.raises(installer.ValidationError): + installer.validate_package_plugin(cfg) + cfg.package_name = "abc" + with pytest.raises(installer.ValidationError): + installer.validate_package_plugin(cfg, user=DummyUser(False)) + installer.validate_package_plugin(cfg, user=DummyUser(True)) + + +def test_uninstall_plugin_happy_path(fake_settings, mock_subprocess, mock_registry, mock_staticfiles, mock_log_error, monkeypatch): + """Test that uninstall_plugin successfully uninstalls plugins and clears config/static files.""" + cfg = DummyPluginConfig(active=False, package_name="pkgA") + cfg.plugin = object() + monkeypatch.setattr(installer, "get_install_info", lambda name: {"location": "/venv", "version": "1.0"}, raising=True) + result = installer.uninstall_plugin(cfg, user=DummyUser(True), delete_config=True) + assert result["success"] is True + call = mock_subprocess.calls[0] + assert call["cmd"][3:6] == ["uninstall", "-y", "pkgA"] + assert mock_staticfiles["clear"] == [cfg.key] + assert mock_registry["reload"] + + +@pytest.mark.parametrize( + "attr,expected_msg", + [ + ("PLUGINS_INSTALL_DISABLED", "disabled"), + ], +) +def test_uninstall_plugin_disabled_by_setting(fake_settings, attr, expected_msg, monkeypatch): + """Test that uninstall_plugin raises ValidationError when uninstallation is disabled by settings.""" + setattr(fake_settings, attr, True) + monkeypatch.setattr(installer, "settings", fake_settings, raising=True) + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(DummyPluginConfig(), user=DummyUser(True)) + assert expected_msg in str(ex.value) + + +@pytest.mark.parametrize( + "cfg_kwargs, expected_error_fragment", + [ + ({"active": True}, "currently active"), + ({"installed": False}, "not installed"), + ({"is_package": False}, "not a packaged plugin"), + ({"package_name": ""}, "package name not found"), + ({"mandatory": True}, "mandatory"), + ({"sample": True}, "sample"), + ({"builtin": True}, "built-in"), + ], +) +def test_uninstall_plugin_validation_errors(fake_settings, cfg_kwargs, expected_error_fragment, monkeypatch): + """Test that uninstall_plugin raises ValidationError for various invalid plugin configurations.""" + # Build DummyPluginConfig kwargs + init_kwargs = { + "active": cfg_kwargs.get("active", False), + "package_name": cfg_kwargs.get("package_name", "pkgA"), + "is_package": cfg_kwargs.get("is_package", True), + "installed": cfg_kwargs.get("installed", True), + "mandatory": cfg_kwargs.get("mandatory", False), + "sample": cfg_kwargs.get("sample", False), + "builtin": cfg_kwargs.get("builtin", False), + } + cfg = DummyPluginConfig(**init_kwargs) + cfg.plugin = object() + + if cfg.active: + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "currently active" in str(ex.value) + return + + if not cfg.is_installed(): + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "not installed" in str(ex.value) + return + + if not cfg.is_package(): + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "not a packaged plugin" in str(ex.value) + return + + if not cfg.package_name: + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "package name not found" in str(ex.value) + return + + if cfg.is_mandatory(): + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "mandatory" in str(ex.value) + return + + if cfg.is_sample(): + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "sample" in str(ex.value) + return + + if cfg.is_builtin(): + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "built-in" in str(ex.value) + return + + +def test_uninstall_plugin_installation_not_found_raises(fake_settings, monkeypatch): + """Test that uninstall_plugin raises ValidationError when installation information is not found.""" + cfg = DummyPluginConfig(active=False, package_name="pkgA") + cfg.plugin = object() + monkeypatch.setattr(installer, "get_install_info", lambda name: {}, raising=True) + with pytest.raises(installer.ValidationError) as ex: + installer.uninstall_plugin(cfg, user=DummyUser(True)) + assert "installation not found" in str(ex.value) \ No newline at end of file diff --git a/tests/inventree/test_test_config_module.py b/tests/inventree/test_test_config_module.py new file mode 100644 index 000000000000..1439fe2528b4 --- /dev/null +++ b/tests/inventree/test_test_config_module.py @@ -0,0 +1,581 @@ +"""Unit tests for InvenTree.test_config helper functions. + +Testing framework: pytest (prefer pytest-django if repository provides it). +These tests exercise pure functions, env/config resolution, and filesystem helpers. +Django-dependent helpers are exercised via monkeypatching to avoid requiring a full Django runtime. +""" + +import os +import sys +import types +import logging +from pathlib import Path +from contextlib import contextmanager + +import pytest + +# Import the module under test +# The module path is src/backend/InvenTree/InvenTree/test_config.py +# Ensure import path includes "src/backend" so "InvenTree" package can be resolved in tests. +REPO_ROOT = Path(__file__).resolve().parents[2] +SRC_BACKEND = REPO_ROOT / "src" / "backend" +if str(SRC_BACKEND) not in sys.path: + sys.path.insert(0, str(SRC_BACKEND)) + +from InvenTree import test_config as tc + +@contextmanager +def environ(environ_overrides: dict): + """Temporarily set environment variables within a 'with' block.""" + old_environ = os.environ.copy() + try: + os.environ.update({k: str(v) for k, v in environ_overrides.items()}) + yield + finally: + # Restore exact previous state + os.environ.clear() + os.environ.update(old_environ) + +def test_to_list_basic_and_tuple(): + """Test to_list with basic inputs and tuple handling.""" + assert tc.to_list(["a", "b"]) == ["a", "b"] + assert tc.to_list(("a", "b")) == ("a", "b") + assert tc.to_list("a,b") == ["a", "b"] + assert tc.to_list(" a , b , c ") == ["a", "b", "c"] + # Custom delimiter + assert tc.to_list("a|b|c", delimiter="|") == ["a", "b", "c"] + # Non-string values should be coerced then split + assert tc.to_list(123) == ["123"] + +def test_to_dict_handles_none_and_dict_and_json(caplog): + """Test to_dict with None, dict, valid and invalid JSON strings.""" + assert tc.to_dict(None) == {} + src = {"x": 1} + assert tc.to_dict(src) is src + assert tc.to_dict('{"a": 5, "b": [1,2]}') == {"a": 5, "b": [1, 2]} + # Invalid JSON logs exception and returns {} + caplog.set_level(logging.ERROR, logger="inventree") + assert tc.to_dict("{not json}") == {} + # Ensure an exception log record was produced + assert any("Failed to parse value" in rec.getMessage() for rec in caplog.records) + +@pytest.mark.parametrize( + "value,expect", + [ + ("1", True), + ("Y", True), + ("yes", True), + ("t", True), + ("true", True), + ("on", True), + ("0", False), + ("n", False), + ("false", False), + ("off", False), + ("", False), + (None, False), + ], +) +def test_is_true_matrix(value, expect): + """Test is_true function with various truthy and falsy string values.""" + assert tc.is_true(value) is expect + +def test_get_base_and_root_dirs_exist(): + """Test that base and root directories exist and base is inside root.""" + base = tc.get_base_dir() + root = tc.get_root_dir() + assert base.exists() and base.is_dir() + assert root.exists() and root.is_dir() + # base should be inside root (heuristic) + assert str(base).startswith(str(root)) + +def test_do_typecast_list_and_dict_and_numeric(caplog): + """Test do_typecast function for list, dict, and numeric types, and error logging.""" + assert tc.do_typecast("a,b,c", list) == ["a", "b", "c"] + assert tc.do_typecast('{"a": 1}', dict) == {"a": 1} + assert tc.do_typecast("5", int) == 5 + assert tc.do_typecast("3.14", float) == pytest.approx(3.14) + # Failed cast returns original and logs if var_name provided + caplog.set_level(logging.ERROR, logger="inventree") + val = tc.do_typecast("abc", int, var_name="TEST_VAR") + assert val == "abc" + assert any("Failed to typecast" in rec.getMessage() for rec in caplog.records) + +def write_yaml(path: Path, data: dict): + """Write a dictionary to a YAML file at the given path.""" + import yaml + text = yaml.safe_dump(data) + path.write_text(text, encoding="utf-8") + +def test_get_setting_precedence_env_over_yaml(tmp_path, monkeypatch): + """Test get_setting prioritizes environment variables over YAML config.""" + # Prepare a temp config directory and file + conf_dir = tmp_path / "config" + conf_dir.mkdir(parents=True, exist_ok=True) + cfg = conf_dir / "config.yaml" + write_yaml(cfg, {"section": {"value": "from_yaml"}, "plain": "yaml_plain"}) + + # Monkeypatch config dir and base dir lookups + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg)) + + # Ensure cache is cleared + tc.CONFIG_DATA = None + tc.CONFIG_LOOKUPS.clear() + + # No env var -> fallback to YAML + v1 = tc.get_setting(env_var="ENV_ONLY", config_key="section.value", default_value="default") + assert v1 == "from_yaml" + assert tc.CONFIG_LOOKUPS["ENV_ONLY"]["source"] == "yaml" + + # With env var -> env takes precedence and is typecasted if requested + with environ({"ENV_ONLY": "42"}): + v2 = tc.get_setting(env_var="ENV_ONLY", config_key="section.value", default_value="default", typecast=int) + assert v2 == 42 + assert tc.CONFIG_LOOKUPS["ENV_ONLY"]["source"] == "env" + + # Missing everywhere -> default with typecast + v3 = tc.get_setting(env_var="MISSING_ENV", config_key="missing.key", default_value="1,2,3", typecast=list) + assert v3 == ["1", "2", "3"] + assert tc.CONFIG_LOOKUPS["MISSING_ENV"]["source"] == "default" + +def test_get_boolean_setting_uses_is_true(monkeypatch, tmp_path): + """Test get_boolean_setting returns boolean values using is_true.""" + cfg = tmp_path / "config.yaml" + write_yaml(cfg, {"feature": {"enabled": "yes"}}) + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg)) + # from YAML + assert tc.get_boolean_setting(config_key="feature.enabled") is True + # from ENV + with environ({"FEATURE_FLAG": "0"}): + assert tc.get_boolean_setting(env_var="FEATURE_FLAG", default_value=True) is False + +def test_get_media_static_backup_dirs_create_and_error(tmp_path, monkeypatch): + """Test get_media_dir, get_static_dir, and get_backup_dir creation and error behavior.""" + # Point settings to temporary paths + media = tmp_path / "m" + static = tmp_path / "s" + backup = tmp_path / "b" + with environ( + { + "INVENTREE_MEDIA_ROOT": str(media), + "INVENTREE_STATIC_ROOT": str(static), + "INVENTREE_BACKUP_DIR": str(backup), + } + ): + md = tc.get_media_dir(create=True) + sd = tc.get_static_dir(create=True) + bd = tc.get_backup_dir(create=True) + + assert md == media.resolve() and md.exists() + assert sd == static.resolve() and sd.exists() + assert bd == backup.resolve() and bd.exists() + + # Missing env with error=False returns None; with error=True raises + with environ({"INVENTREE_MEDIA_ROOT": ""}): + assert tc.get_media_dir(create=False, error=False) is None + with pytest.raises(FileNotFoundError): + tc.get_media_dir(create=False, error=True) + + with environ({"INVENTREE_STATIC_ROOT": ""}): + assert tc.get_static_dir(create=False, error=False) is None + with pytest.raises(FileNotFoundError): + tc.get_static_dir(create=False, error=True) + + with environ({"INVENTREE_BACKUP_DIR": ""}): + assert tc.get_backup_dir(create=False, error=False) is None + with pytest.raises(FileNotFoundError): + tc.get_backup_dir(create=False, error=True) + +def test_get_config_file_prefers_env_and_creates_from_template(tmp_path, monkeypatch, capsys): + """Test get_config_file respects env var and creates config file from template.""" + # Create a fake base dir and template + # We monkeypatch get_base_dir and get_config_dir to our tmp paths + fake_base = tmp_path / "base" + fake_conf = tmp_path / "conf" + (fake_base / "InvenTree").mkdir(parents=True, exist_ok=True) + fake_conf.mkdir(parents=True, exist_ok=True) + + # Provide a template file + template = fake_base / "config_template.yaml" + template.write_text("key: default", encoding="utf-8") + + monkeypatch.setattr(tc, "get_base_dir", lambda: fake_base) + monkeypatch.setattr(tc, "get_config_dir", lambda: fake_conf) + + # Case 1: Explicit env path used; since it doesn't exist and create=True, it should be created from template + env_cfg = fake_conf / "custom.yaml" + with environ({"INVENTREE_CONFIG_FILE": str(env_cfg)}): + path = tc.get_config_file(create=True) + assert path == env_cfg.resolve() + # Should have been created from template + assert path.exists() + assert "key: default" in path.read_text() + + # Case 2: No env provided; default under config dir + os.environ.pop("INVENTREE_CONFIG_FILE", None) + # Remove existing config.yaml to force creation + default_cfg = fake_conf / "config.yaml" + if default_cfg.exists(): + default_cfg.unlink() + path2 = tc.get_config_file(create=True) + assert path2 == default_cfg.resolve() + assert default_cfg.exists() + + # Standard output printed creation notices + out = capsys.readouterr().out + assert "InvenTree configuration file 'config.yaml' not found - creating default file" in out + assert "Created config file" in out + +def test_load_config_data_caching(tmp_path, monkeypatch): + """Test load_config_data caching behavior with set_cache flag.""" + cfg = tmp_path / "config.yaml" + write_yaml(cfg, {"x": {"y": 7}}) + + # Point get_config_file to our path + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg)) + + # No cache initially + tc.CONFIG_DATA = None + d1 = tc.load_config_data(set_cache=False) + assert d1 == {"x": {"y": 7}} + # Set cache + d2 = tc.load_config_data(set_cache=True) + assert d2 == {"x": {"y": 7}} + # Ensure subsequent call without set_cache picks up cached instance (by identity or equality) + d3 = tc.load_config_data(set_cache=False) + assert d3 == {"x": {"y": 7}} + # Changing file should not impact cached data unless set_cache=True + write_yaml(cfg, {"x": {"y": 9}}) + d4 = tc.load_config_data(set_cache=False) + assert d4 == {"x": {"y": 7}} # still old since cached + d5 = tc.load_config_data(set_cache=True) + assert d5 == {"x": {"y": 9}} + +def test_get_constraint_file_creates_when_missing(tmp_path, monkeypatch, caplog): + """Test get_constraint_file creates constraints file when missing and respects force_write.""" + # Prepare a fake base dir with InvenTree subdir for constraint file location + fake_base = tmp_path / "base" + target_dir = fake_base / "InvenTree" + target_dir.mkdir(parents=True, exist_ok=True) + + # Monkeypatch get_base_dir and version constants + monkeypatch.setattr(tc, "get_base_dir", lambda: fake_base) + + # Provide a fake version module attributes via monkeypatching import in tc.get_constraint_file + class FakeVersionModule(types.SimpleNamespace): + INVENTREE_SW_VERSION = "1.2.3" + INVENTREE_SW_NXT_MAJOR = "2.0" + + # Inject into sys.modules so from .version import ... resolves + pkg_mod = sys.modules.get("InvenTree") + if pkg_mod is None: + sys.modules["InvenTree"] = types.ModuleType("InvenTree") + sys.modules["InvenTree.version"] = types.ModuleType("InvenTree.version") + sys.modules["InvenTree.version"].INVENTREE_SW_VERSION = "1.2.3" + sys.modules["InvenTree.version"].INVENTREE_SW_NXT_MAJOR = "2.0" + + caplog.set_level(logging.WARNING, logger="inventree") + + const_path = tc.get_constraint_file(force_write=False) + assert Path(const_path).exists() + text = Path(const_path).read_text() + assert "# InvenTree constraints file" in text + # Ensure it formatted the package constraint line with the version range + assert "inventree-server~=1.2.3,<2.0" in text.replace(" ", "") + + # It should have warned because the file did not exist and force_write=False + assert any("Constraint file does not exist" in rec.getMessage() for rec in caplog.records) + + # Calling again should return existing path without re-writing content + caplog.clear() + const_path2 = tc.get_constraint_file(force_write=False) + assert const_path2 == const_path + assert len(caplog.records) == 0 + + # Force write should re-create/overwrite + Path(const_path).unlink() + const_path3 = tc.get_constraint_file(force_write=True) + assert Path(const_path3).exists() + +def test_get_secret_key_env_and_file_generation(tmp_path, monkeypatch, caplog): + """Test get_secret_key returns env value, reads from file, or generates new key.""" + caplog.set_level(logging.INFO, logger="inventree") + # Env var present -> returns the value directly + with environ({"INVENTREE_SECRET_KEY": "supersecret"}): + assert tc.get_secret_key() == "supersecret" + + # Env var for file path takes precedence over default locations + key_file = tmp_path / "key.txt" + key_file.write_text("fromfile", encoding="utf-8") + with environ({"INVENTREE_SECRET_KEY_FILE": str(key_file)}): + assert tc.get_secret_key() == "fromfile" + # return_path returns a Path + assert tc.get_secret_key(return_path=True) == key_file.resolve() + + # No envs -> create in default config dir + fake_base = tmp_path / "base" + fake_conf = tmp_path / "conf" + (fake_base / "InvenTree").mkdir(parents=True, exist_ok=True) + fake_conf.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(tc, "get_base_dir", lambda: fake_base) + monkeypatch.setattr(tc, "get_config_dir", lambda: fake_conf) + + # Remove envs + os.environ.pop("INVENTREE_SECRET_KEY", None) + os.environ.pop("INVENTREE_SECRET_KEY_FILE", None) + + # Trigger generation + path = tc.get_secret_key(return_path=True) + assert isinstance(path, Path) + assert path.exists() + content = tc.get_secret_key() + assert isinstance(content, str) + assert content.strip() != "" + +def test_get_oidc_private_key_env_and_generate(tmp_path, monkeypatch): + """Test get_oidc_private_key returns env raw key, reads from file, or generates new key.""" + # Raw key in env + with environ({"INVENTREE_OIDC_PRIVATE_KEY": "RAW_RSA"}): + assert tc.get_oidc_private_key() == "RAW_RSA" + + # Use provided path via env + pem = tmp_path / "oidc.pem" + pem.write_text("PEM_CONTENT", encoding="utf-8") + with environ({"INVENTREE_OIDC_PRIVATE_KEY_FILE": str(pem)}): + assert tc.get_oidc_private_key(return_path=False) == "PEM_CONTENT" + assert tc.get_oidc_private_key(return_path=True) == pem.resolve() + + # No env; generate at default config dir if missing + fake_conf = tmp_path / "conf" + fake_conf.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(tc, "get_config_dir", lambda: fake_conf) + # Ensure no old default file exists in base dir + fake_base = tmp_path / "base" + (fake_base / "InvenTree").mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(tc, "get_base_dir", lambda: fake_base) + + # Remove envs + os.environ.pop("INVENTREE_OIDC_PRIVATE_KEY", None) + os.environ.pop("INVENTREE_OIDC_PRIVATE_KEY_FILE", None) + + key_text = tc.get_oidc_private_key(return_path=False) + assert "BEGIN RSA PRIVATE KEY" in key_text + key_path = tc.get_oidc_private_key(return_path=True) + assert isinstance(key_path, Path) + assert key_path.exists() + +def test_get_frontend_settings_defaults_and_overrides(monkeypatch, tmp_path): + """Test get_frontend_settings default values and config overrides.""" + # Minimal: no env/config -> defaults + debug true + tc.CONFIG_DATA = None + tc.CONFIG_LOOKUPS.clear() + # Point to a simple config with no frontend_settings + cfg = tmp_path / "config.yaml" + write_yaml(cfg, {}) + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg)) + + s = tc.get_frontend_settings(debug=True) + # Ensure required defaults + assert s["debug"] is True + assert s["environment"] == "development" + assert s["show_server_selector"] is True + assert "server_list" in s + assert isinstance(s["server_list"], list) + # base_url default + assert s["base_url"] in ("web", "/web/", "web/") + + # Provide overrides via dict in config + cfg2 = tmp_path / "config2.yaml" + write_yaml( + cfg2, + { + "frontend_settings": { + "debug": False, # should be overwritten by argument + "api_host": "http://api.example", + "server_list": [{"id": 1}], + "url_compatibility": "not_a_bool", # coerced to True via exception handling + } + }, + ) + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg2)) + s2 = tc.get_frontend_settings(debug=False) + assert s2["debug"] is False + assert s2["environment"] == "production" + assert s2["api_host"] == "http://api.example" + assert s2["server_list"] == [{"id": 1}] + assert s2["url_compatibility"] is True + +def test_check_config_dir_warns_and_sets_global_warning(tmp_path, caplog, monkeypatch): + """Test check_config_dir warns and sets global warning for uncommon config directory.""" + # Put config in a tmp dir different from the recommended directory + recommended = tmp_path / "recommended" + elsewhere = tmp_path / "elsewhere" + recommended.mkdir(parents=True, exist_ok=True) + elsewhere.mkdir(parents=True, exist_ok=True) + file_path = elsewhere / "config.yaml" + file_path.write_text("x: 1", encoding="utf-8") + + # Simulate get_config_dir to be recommended + monkeypatch.setattr(tc, "get_config_dir", lambda: recommended) + + # Provide a dummy common.settings.set_global_warning + fake_common = types.ModuleType("common") + fake_settings = types.ModuleType("common.settings") + + class GlobalWarningCode: + UNCOMMON_CONFIG = "UNCOMMON_CONFIG" + + calls = {} + + def set_global_warning(code, payload): + calls["code"] = code + calls["payload"] = payload + + fake_settings.GlobalWarningCode = GlobalWarningCode + fake_settings.set_global_warning = set_global_warning + fake_common.settings = fake_settings + sys.modules["common"] = fake_common + sys.modules["common.settings"] = fake_settings + + caplog.set_level(logging.WARNING, logger="inventree") + + tc.check_config_dir("INVENTREE_CONFIG_FILE", file_path, config_dir=None) + + # Should warn + assert any("INVE-W10" in rec.getMessage() for rec in caplog.records) + # Should invoke set_global_warning with recommended path + assert calls.get("code") == "UNCOMMON_CONFIG" + assert calls.get("payload", {}).get("path") == str(recommended) + +def test_ensure_dir_with_default_filesystem(tmp_path): + """Test ensure_dir creates directories idempotently.""" + # New directory created + target = tmp_path / "nested" / "dir" + assert not target.exists() + tc.ensure_dir(target) + assert target.exists() and target.is_dir() + # Calling again should be idempotent + tc.ensure_dir(target) + assert target.exists() + +def test_get_config_dir_by_installer_env(monkeypatch): + """Test get_config_dir returns correct paths for different installer environments.""" + # inventreeInstaller returns indicators based on env variables + with environ({"INVENTREE_DOCKER": "true"}): + assert str(tc.get_config_dir()) == "/home/inventree/data" + with environ({"INVENTREE_DEVCONTAINER": "true"}): + assert str(tc.get_config_dir()) == "/home/inventree/dev" + with environ({"INVENTREE_PKG_INSTALLER": "PKG"}): + assert str(tc.get_config_dir()) == "/etc/inventree" + + # Fallback to project config directory under root if none matched + os.environ.pop("INVENTREE_DOCKER", None) + os.environ.pop("INVENTREE_DEVCONTAINER", None) + os.environ.pop("INVENTREE_PKG_INSTALLER", None) + # get_root_dir may vary; just assert it ends with /config + p = tc.get_config_dir() + assert p.name == "config" + +def test_get_custom_file_static_and_media_resolution(monkeypatch, caplog): + """Test get_custom_file resolves files from static and media storage.""" + # Avoid importing Django by providing fake modules in sys.modules + # Fake StaticFilesStorage with exists method + class FakeStatic: + def __init__(self, existing): + self._existing = set(existing) + + def exists(self, path): + return path in self._existing + + # Fake default_storage with exists method + class FakeDefaultStorage: + def __init__(self, existing): + self._existing = set(existing) + + def exists(self, path): + return path in self._existing + + # Inject fake modules and names + fake_django_contrib_staticfiles_storage = types.ModuleType( + "django.contrib.staticfiles.storage" + ) + fake_django_contrib_staticfiles_storage.StaticFilesStorage = lambda: FakeStatic( + {"static/file.css"} + ) + + fake_django_core_files_storage = types.ModuleType("django.core.files.storage") + fake_django_core_files_storage.default_storage = FakeDefaultStorage( + {"media/file.css"} + ) + + sys.modules["django.contrib"] = types.ModuleType("django.contrib") + sys.modules["django.contrib.staticfiles"] = types.ModuleType( + "django.contrib.staticfiles" + ) + sys.modules["django.contrib.staticfiles.storage"] = ( + fake_django_contrib_staticfiles_storage + ) + sys.modules["django.core"] = types.ModuleType("django.core") + sys.modules["django.core.files"] = types.ModuleType("django.core.files") + sys.modules["django.core.files.storage"] = fake_django_core_files_storage + + caplog.set_level(logging.INFO, logger="inventree") + + # Case 1: found in static + v1 = tc.get_custom_file("ENV", "CONF", "asset", lookup_media=True) + # By default get_setting returns None when not provided, so provide value via env + with environ({"ENV": "static/file.css"}): + v1 = tc.get_custom_file("ENV", "CONF", "asset", lookup_media=True) + assert v1 == "static/file.css" + assert any("Loading asset from static directory" in rec.getMessage() for rec in caplog.records) + + caplog.clear() + # Case 2: found in media when lookup_media=True + with environ({"ENV": "media/file.css"}): + v2 = tc.get_custom_file("ENV", "CONF", "asset", lookup_media=True) + assert v2 == "media/file.css" + assert any("Loading asset from media directory" in rec.getMessage() for rec in caplog.records) + + caplog.clear() + # Case 3: missing -> warning and returns Falsey value + with environ({"ENV": "not/found.css"}): + v3 = tc.get_custom_file("ENV", "CONF", "asset", lookup_media=False) + assert v3 is False + assert any("could not be found" in rec.getMessage() for rec in caplog.records) + +def test_get_plugin_file_creates_when_missing(tmp_path, monkeypatch, caplog): + """Test get_plugin_file creates plugin file when missing or uses env path.""" + caplog.set_level(logging.WARNING, logger="inventree") + # Point config file to a tmp dir; plugin file should be created next to it + cfg = tmp_path / "config.yaml" + cfg.write_text("{}", encoding="utf-8") + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg)) + + # No env for plugin file -> default to config dir/plugins.txt + plugin_path = tc.get_plugin_file() + assert isinstance(plugin_path, Path) + assert plugin_path.exists() + text = plugin_path.read_text() + assert "InvenTree Plugins" in text + + # Provide explicit path via env + caplog.clear() + explicit = tmp_path / "custom_plugins.txt" + with environ({"INVENTREE_PLUGIN_FILE": str(explicit)}): + p2 = tc.get_plugin_file() + assert p2 == explicit + assert p2.exists() + # Verify warnings were issued in first path (creation) + assert any("Plugin configuration file does not exist" in rec.getMessage() for rec in caplog.records) + +def test_get_plugin_dir_returns_setting_from_env_or_config(tmp_path, monkeypatch): + """Test get_plugin_dir returns plugin directory from environment or config.""" + # Via env + with environ({"INVENTREE_PLUGIN_DIR": "/tmp/plugins"}): + assert tc.get_plugin_dir() == "/tmp/plugins" + # Via config file + cfg = tmp_path / "config.yaml" + write_yaml(cfg, {"plugin_dir": "/var/plugins"}) + monkeypatch.setenv("INVENTREE_CONFIG_FILE", str(cfg)) + os.environ.pop("INVENTREE_PLUGIN_DIR", None) + assert tc.get_plugin_dir() == "/var/plugins" \ No newline at end of file diff --git a/tests/test_tasks.py b/tests/test_tasks.py new file mode 100644 index 000000000000..408e708437b7 --- /dev/null +++ b/tests/test_tasks.py @@ -0,0 +1,691 @@ +"""Test suite for tasks module.""" + +import builtins +import sys +import types +from pathlib import Path + +import pytest + +# Import the module under test. +# We attempt common names: tasks.py at repo root or a package path. +# If the module is elsewhere, adjust the import path accordingly. +try: + import tasks as tasks_mod +except ModuleNotFoundError: + # Fallback: attempt src/tasks.py or similar structures that are common + import importlib.util + + candidates = [ + "tasks.py", + "src/tasks.py", + "backend/tasks.py", + "scripts/tasks.py", + "inv_tasks.py", + ] + _loaded = False + for _c in candidates: + try: + p = Path(_c) + if p.exists(): + spec = importlib.util.spec_from_file_location("tasks_mod", str(p)) + tasks_mod = importlib.util.module_from_spec(spec) + sys.modules["tasks_mod"] = tasks_mod + spec.loader.exec_module(tasks_mod) # type: ignore + _loaded = True + break + except Exception: + continue + if not _loaded: + raise + + +def test_is_true_truthy_and_falsy_values(): + """Test is_true returns True for truthy inputs and False for falsy ones.""" + # Truthy cases + for val in ["1", "Y", "y", " yes ", "t", "TRUE", "On", True]: + assert tasks_mod.is_true(val), f"Expected truthy for {val!r}" + + # Falsy cases + for val in ["0", "n", "no", "off", "false", "", "random", 0, None, False]: + assert tasks_mod.is_true(val) is False, f"Expected falsy for {val!r}" + + +@pytest.mark.parametrize( + "env_var,value,expected", + [ + ("INVENTREE_DEVCONTAINER", "True", True), + ("INVENTREE_DEVCONTAINER", "1", True), + ("INVENTREE_DEVCONTAINER", "yes", True), + ("INVENTREE_DEVCONTAINER", "no", False), + ("INVENTREE_DEVCONTAINER", None, False), + ], +) +def test_is_devcontainer_environment(monkeypatch, env_var, value, expected): + """Test is_devcontainer_environment returns expected based on environment variable.""" + if value is None: + monkeypatch.delenv(env_var, raising=False) + else: + monkeypatch.setenv(env_var, value) + assert tasks_mod.is_devcontainer_environment() is expected + + +@pytest.mark.parametrize( + "env_var,value,expected", + [ + ("INVENTREE_DOCKER", "True", True), + ("INVENTREE_DOCKER", "0", False), + ("INVENTREE_DOCKER", None, False), + ], +) +def test_is_docker_environment(monkeypatch, env_var, value, expected): + """Test is_docker_environment returns expected based on environment variable.""" + if value is None: + monkeypatch.delenv(env_var, raising=False) + else: + monkeypatch.setenv(env_var, value) + assert tasks_mod.is_docker_environment() is expected + + +@pytest.mark.parametrize( + "env_var,value,expected", + [ + ("READTHEDOCS", "on", True), + ("READTHEDOCS", "off", False), + ("READTHEDOCS", None, False), + ], +) +def test_is_rtd_environment(monkeypatch, env_var, value, expected): + """Test is_rtd_environment returns expected based on environment variable.""" + if value is None: + monkeypatch.delenv(env_var, raising=False) + else: + monkeypatch.setenv(env_var, value) + assert tasks_mod.is_rtd_environment() is expected + + +def test_is_debug_environment_via_inventree_debug(monkeypatch): + """Test is_debug_environment returns True when INVENTREE_DEBUG is set.""" + monkeypatch.setenv("INVENTREE_DEBUG", "true") + monkeypatch.delenv("RUNNER_DEBUG", raising=False) + assert tasks_mod.is_debug_environment() is True + + +def test_is_debug_environment_via_runner_debug(monkeypatch): + """Test is_debug_environment returns True when RUNNER_DEBUG is set.""" + monkeypatch.delenv("INVENTREE_DEBUG", raising=False) + monkeypatch.setenv("RUNNER_DEBUG", "1") + assert tasks_mod.is_debug_environment() is True + + +def test_is_debug_environment_false(monkeypatch): + """Test is_debug_environment returns False when debug vars indicate false.""" + monkeypatch.setenv("INVENTREE_DEBUG", "0") + monkeypatch.delenv("RUNNER_DEBUG", raising=False) + assert tasks_mod.is_debug_environment() is False + + +def test_get_version_vals_success_with_dotenv(monkeypatch, tmp_path): + """Test get_version_vals successfully reads VERSION file using dotenv.""" + # Skip if dotenv is not installed and we don't want to manipulate import machinery + pytest.importorskip("dotenv") + # Prepare a VERSION file as dotenv format + version_dir = tmp_path + (version_dir / "VERSION").write_text("INVENTREE_PKG_INSTALLER=PKG\nFOO=bar\n", encoding="utf-8") + + # Patch local_dir to point to our temp directory + monkeypatch.setattr(tasks_mod, "local_dir", lambda: version_dir) + + vals = tasks_mod.get_version_vals() + assert vals.get("INVENTREE_PKG_INSTALLER") == "PKG" + assert vals.get("FOO") == "bar" + + +def test_get_version_vals_importerror(monkeypatch, tmp_path, capsys): + """Test get_version_vals handles ImportError when dotenv is missing.""" + # Create VERSION file; import of dotenv will be forced to fail + version_dir = tmp_path + (version_dir / "VERSION").write_text("INVENTREE_PKG_INSTALLER=PKG\n", encoding="utf-8") + monkeypatch.setattr(tasks_mod, "local_dir", lambda: version_dir) + + orig_import = builtins.__import__ + + def fake_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "dotenv" or (name == "dotenv" and "dotenv_values" in fromlist): + raise ImportError("No module named 'dotenv'") + return orig_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fake_import) + vals = tasks_mod.get_version_vals() + captured = capsys.readouterr().out + assert vals == {} + assert "dotenv package not installed" in captured + + +def test_is_pkg_installer_with_content(): + """Test is_pkg_installer returns correct boolean based on provided dict.""" + assert tasks_mod.is_pkg_installer({"INVENTREE_PKG_INSTALLER": "PKG"}) is True + assert tasks_mod.is_pkg_installer({"INVENTREE_PKG_INSTALLER": "OTHER"}) is False + assert tasks_mod.is_pkg_installer({}) is False + + +def test_is_pkg_installer_load_content(monkeypatch): + """Test is_pkg_installer with load_content flag fetches version values.""" + monkeypatch.setattr(tasks_mod, "get_version_vals", lambda: {"INVENTREE_PKG_INSTALLER": "PKG"}) + assert tasks_mod.is_pkg_installer(load_content=True) is True + monkeypatch.setattr(tasks_mod, "get_version_vals", lambda: {"INVENTREE_PKG_INSTALLER": "OTHER"}) + assert tasks_mod.is_pkg_installer(load_content=True) is False + + +def test_is_pkg_installer_by_path(monkeypatch): + """Test is_pkg_installer_by_path detects pkg installer based on sys.argv.""" + monkeypatch.setattr(sys, "argv", ["/opt/inventree/env/bin/invoke", "whatever"]) + assert tasks_mod.is_pkg_installer_by_path() is True + monkeypatch.setattr(sys, "argv", ["/usr/local/bin/invoke"]) + assert tasks_mod.is_pkg_installer_by_path() is False + + +def test_get_installer_with_env(monkeypatch): + """Test get_installer reads INVENTREE_PKG_INSTALLER from environment.""" + monkeypatch.setenv("INVENTREE_PKG_INSTALLER", "PKG") + assert tasks_mod.get_installer() == "PKG" + monkeypatch.delenv("INVENTREE_PKG_INSTALLER", raising=False) + assert tasks_mod.get_installer() is None + + +def test_task_exception_handler_module_not_found(monkeypatch, capsys): + """Test task_exception_handler handles ModuleNotFoundError specially.""" + # Prevent actual default excepthook output + called = {} + + def fake_excepthook(t, v, tb): + called["ok"] = (t, v, tb) + + monkeypatch.setattr(sys, "__excepthook__", fake_excepthook) + + v = ModuleNotFoundError("No module named 'foo'") + tasks_mod.task_exception_handler(ModuleNotFoundError, v, None) + out = capsys.readouterr().out + assert "Error importing required module: foo" in out + assert "Ensure the correct Python virtual environment is active" in out + + +def test_task_exception_handler_other_exception(monkeypatch, capsys): + """Test task_exception_handler delegates to default hook for other exceptions.""" + # Prevent actual default excepthook output + called = {} + + def fake_excepthook(t, v, tb): + called["ok"] = True + + monkeypatch.setattr(sys, "__excepthook__", fake_excepthook) + + v = ValueError("bad") + tasks_mod.task_exception_handler(ValueError, v, None) + out = capsys.readouterr().out + # Should not print module-not-found guidance + assert "Error importing required module:" not in out + assert called.get("ok") is True + + +def test_wrap_color_and_print_helpers(capsys): + """Test wrap_color and print helper functions produce colored output.""" + txt = tasks_mod.wrap_color("hello", "92") + assert txt == "\033[92mhello\033[0m" + + tasks_mod.success("ok") + tasks_mod.error("err") + tasks_mod.warning("warn") + tasks_mod.info("inf") + + out = capsys.readouterr().out + assert "\033[92mok\033[0m" in out + assert "\033[91merr\033[0m" in out + assert "\033[93mwarn\033[0m" in out + assert "\033[94minf\033[0m" in out + + +def test_state_logger_default_name(monkeypatch, capsys): + """Test state_logger decorator logs start and done with default function name.""" + # Force debug env to enable logging + monkeypatch.setenv("INVENTREE_DEBUG", "1") + monkeypatch.delenv("RUNNER_DEBUG", raising=False) + + calls = [] + + @tasks_mod.state_logger + def sample_task(_c): + calls.append("ran") + + sample_task(None) + out = capsys.readouterr().out + assert "# invoke task named `sample_task`| start" in out + assert "# invoke task named `sample_task`| done" in out + assert calls == ["ran"] + + +def test_state_logger_custom_name(monkeypatch, capsys): + """Test state_logger decorator logs start and done with custom name.""" + monkeypatch.setenv("INVENTREE_DEBUG", "1") + + @tasks_mod.state_logger("custom task") + def custom(_c): + pass + + custom(None) + out = capsys.readouterr().out + assert "# custom task| start" in out + assert "# custom task| done" in out + + +def _dummy_invoke_module(tmp_path: Path, version: str) -> types.SimpleNamespace: + # Create a dummy invoke module replacement + m = types.SimpleNamespace() + m.__version__ = version + # place __file__ somewhere we can control + m.__file__ = str(tmp_path / "invoke" / "__init__.py") + (tmp_path / "invoke").mkdir(parents=True, exist_ok=True) + (tmp_path / "invoke" / "__init__.py").write_text("# dummy", encoding="utf-8") + return m + + +def test_envcheck_invoke_version_ok(monkeypatch, tmp_path): + """Test envcheck_invoke_version does not exit for supported invoke versions.""" + dummy = _dummy_invoke_module(tmp_path, "2.1.0") + monkeypatch.setattr(tasks_mod, "invoke", dummy, raising=True) + # Should not raise SystemExit + tasks_mod.envcheck_invoke_version() + + +def test_envcheck_invoke_version_fail(monkeypatch, tmp_path, capsys): + """Test envcheck_invoke_version exits for unsupported invoke versions.""" + dummy = _dummy_invoke_module(tmp_path, "1.9.9") + monkeypatch.setattr(tasks_mod, "invoke", dummy, raising=True) + with pytest.raises(SystemExit): + tasks_mod.envcheck_invoke_version() + out = capsys.readouterr().out + assert "The installed invoke version (1.9.9) is not supported!" in out + + +def test_envcheck_invoke_path_ok_env_prefix(monkeypatch, tmp_path): + """Test envcheck_invoke_path passes when invoke is under virtualenv prefix.""" + dummy = _dummy_invoke_module(tmp_path, "2.2.0") + # Make invoke path under env path + env_prefix = tmp_path / "venv" + (env_prefix / "lib").mkdir(parents=True, exist_ok=True) + dummy.__file__ = str(env_prefix / "lib" / "invoke" / "__init__.py") + (env_prefix / "lib" / "invoke").mkdir(parents=True, exist_ok=True) + (env_prefix / "lib" / "invoke" / "__init__.py").write_text("# dummy", encoding="utf-8") + + monkeypatch.setattr(tasks_mod, "invoke", dummy, raising=True) + monkeypatch.setattr(sys, "prefix", str(env_prefix)) + # Not docker or RTD, to enforce check + monkeypatch.setenv("INVENTREE_DOCKER", "0") + monkeypatch.setenv("READTHEDOCS", "0") + tasks_mod.envcheck_invoke_path() # should pass without exit + + +def test_envcheck_invoke_path_fail(monkeypatch, tmp_path, capsys): + """Test envcheck_invoke_path exits when invoke is not under virtualenv prefix.""" + dummy = _dummy_invoke_module(tmp_path, "2.2.0") + monkeypatch.setattr(tasks_mod, "invoke", dummy, raising=True) + monkeypatch.setattr(sys, "prefix", str(tmp_path / "another_prefix")) + + # ensure not in docker or RTD + monkeypatch.delenv("INVENTREE_DOCKER", raising=False) + monkeypatch.delenv("READTHEDOCS", raising=False) + + with pytest.raises(SystemExit): + tasks_mod.envcheck_invoke_path() + out = capsys.readouterr().out + assert "INVE-E2 - Wrong Invoke Path" in out + + +def test_envcheck_python_version_pass(monkeypatch): + """Test envcheck_python_version does not exit for supported Python versions.""" + class VInfo: + major = 3 + minor = 10 + + monkeypatch.setattr(sys, "version_info", VInfo()) + monkeypatch.setattr(sys, "version", "3.10.1 (main, Jan 1 2024, 00:00:00)") + + tasks_mod.envcheck_python_version() # no exit + + +def test_envcheck_python_version_fail(monkeypatch, capsys): + """Test envcheck_python_version exits for unsupported Python versions.""" + class VInfo: + major = 3 + minor = 8 + + monkeypatch.setattr(sys, "version_info", VInfo()) + monkeypatch.setattr(sys, "version", "3.8.18 (main, Jan 1 2024, 00:00:00)") + with pytest.raises(SystemExit): + tasks_mod.envcheck_python_version() + out = capsys.readouterr().out + assert "InvenTree requires Python 3.9 or above" in out + + +def test_envcheck_invoke_cmd_skip_in_docker(monkeypatch, capsys): + """Test envcheck_invoke_cmd skips checks in Docker environment.""" + monkeypatch.setenv("INVENTTREE_DOCKER", "1") + monkeypatch.setattr(sys, "argv", ["/anything"]) + tasks_mod.envcheck_invoke_cmd() + out = capsys.readouterr().out + # Should produce no error outputs + assert "Wrong Invoke Environment" not in out + + +def test_envcheck_invoke_cmd_pkg_installer(monkeypatch, capsys): + """Test envcheck_invoke_cmd warns for pkg installer environment.""" + monkeypatch.delenv("INVENTREE_DOCKER", raising=False) + monkeypatch.delenv("READTHEDOCS", raising=False) + monkeypatch.delenv("INVENTREE_DEVCONTAINER", raising=False) + + # Force pkg installer environment but not the path + monkeypatch.setattr(tasks_mod, "is_pkg_installer", lambda load_content=False: True) + monkeypatch.setattr(tasks_mod, "is_pkg_installer_by_path", lambda: False) + monkeypatch.setattr(sys, "argv", ["/usr/bin/python"]) + monkeypatch.setattr(sys, "prefix", "/venv") + tasks_mod.envcheck_invoke_cmd() + out = capsys.readouterr().out + assert "INVE-W9 - Wrong Invoke Environment" in out + assert "inventree run invoke" in out + + +def test_envcheck_invoke_cmd_unknown_environment(monkeypatch, capsys): + """Test envcheck_invoke_cmd warns for unknown environment.""" + monkeypatch.delenv("INVENTREE_DOCKER", raising=False) + monkeypatch.delenv("READTHEDOCS", raising=False) + monkeypatch.delenv("INVENTREE_DEVCONTAINER", raising=False) + monkeypatch.setattr(tasks_mod, "is_pkg_installer", lambda load_content=False: False) + monkeypatch.setattr(sys, "argv", ["/usr/bin/python"]) + tasks_mod.envcheck_invoke_cmd() + out = capsys.readouterr().out + assert "Unknown environment, not checking used invoke command" in out + assert "INVE-W9 - Wrong Invoke Environment" in out + + +def test_main_invokes_all_checks(monkeypatch): + """Test main invokes all environment checks in order.""" + called = {"v": False, "p": False, "path": False, "cmd": False} + monkeypatch.setattr(tasks_mod, "envcheck_invoke_version", lambda: called.__setitem__("v", True)) + monkeypatch.setattr(tasks_mod, "envcheck_python_version", lambda: called.__setitem__("p", True)) + monkeypatch.setattr(tasks_mod, "envcheck_invoke_path", lambda: called.__setitem__("path", True)) + monkeypatch.setattr(tasks_mod, "envcheck_invoke_cmd", lambda: called.__setitem__("cmd", True)) + + tasks_mod.main() + assert called == {"v": True, "p": True, "path": True, "cmd": True} + + +def test_apps_contains_expected_items(): + """Test apps function returns expected app names.""" + result = tasks_mod.apps() + # Ensure core apps listed + for name in ["build", "common", "company", "order", "part", "stock", "users", "plugin", "InvenTree"]: + assert name in result + # Sanity: non-empty list + assert len(result) >= 5 + + +def test_content_excludes_defaults(): + """Test content_excludes returns default excluded content types.""" + out = tasks_mod.content_excludes() + # Always-excluded content types + for key in [ + "contenttypes", + "auth.permission", + "admin.logentry", + "django_q.schedule", + "django_q.task", + "django_q.ormq", + "exchange.rate", + "exchange.exchangebackend", + "common.dataoutput", + "common.newsfeedentry", + "common.notificationentry", + "common.notificationmessage", + "importer.dataimportsession", + "importer.dataimportcolumnmap", + "importer.dataimportrow", + ]: + assert f"--exclude {key}" in out + + # Defaults allow auth/tokens/plugins/sso/session, + # hence these should NOT be present by default + for key in [ + "auth.group", + "auth.user", + "users.apitoken", + "plugin.pluginconfig", + "plugin.pluginsetting", + "socialaccount.socialapp", + "socialaccount.socialtoken", + "sessions.session", + "usersessions.usersession", + ]: + assert f"--exclude {key}" not in out + + +def test_content_excludes_all_blocked(): + """Test content_excludes blocks all content types when flags are false.""" + out = tasks_mod.content_excludes( + allow_auth=False, + allow_tokens=False, + allow_plugins=False, + allow_sso=False, + allow_session=False, + ) + for key in [ + "auth.group", + "auth.user", + "users.apitoken", + "plugin.pluginconfig", + "plugin.pluginsetting", + "socialaccount.socialapp", + "socialaccount.socialtoken", + "sessions.session", + "usersessions.usersession", + ]: + assert f"--exclude {key}" in out + + +def test_local_and_manage_paths(): + """Test local_dir, manage_py_dir, and manage_py_path return correct paths.""" + loc = tasks_mod.local_dir() + assert isinstance(loc, Path) + mp_dir = tasks_mod.manage_py_dir() + assert mp_dir == loc.joinpath("src", "backend", "InvenTree") + mp_path = tasks_mod.manage_py_path() + assert mp_path == mp_dir.joinpath("manage.py") + + +def test_run_success_builds_correct_command(monkeypatch, tmp_path): + """Test run builds and runs correct command on success.""" + recorded = {} + + class Ctx: + def run(self, cmd, pty=False, env=None): + recorded["cmd"] = cmd + recorded["pty"] = pty + recorded["env"] = env + + # Force local_dir() to our temp path to make command deterministic + monkeypatch.setattr(tasks_mod, "local_dir", lambda: tmp_path) + c = Ctx() + tasks_mod.run(c, "echo hello", env={"A": "B"}) + assert recorded["cmd"] == f'cd "{tmp_path}" && echo hello' + assert recorded["pty"] is False + assert recorded["env"] == {"A": "B"} + + +def test_run_failure_raises_and_logs(monkeypatch, capsys, tmp_path): + """Test run logs error and re-raises UnexpectedExit on failure.""" + # Create a fake UnexpectedExit class and attach to module under test + class FakeUnexpectedExitError(Exception): + pass + + monkeypatch.setattr(tasks_mod, "UnexpectedExit", FakeUnexpectedExitError, raising=True) + + class Ctx: + def run(self, cmd, pty=False, env=None): + raise FakeUnexpectedExitError("boom") + + monkeypatch.setattr(tasks_mod, "local_dir", lambda: tmp_path) + + with pytest.raises(FakeUnexpectedExitError): + tasks_mod.run(Ctx(), "failing command") + out = capsys.readouterr().out + assert "ERROR: InvenTree command failed: 'failing command'" in out + assert "Refer to the error messages in the log above for more information" in out + + +def test_manage_invokes_run_with_manage_py(monkeypatch, tmp_path): + """Test manage delegates to run with manage.py command.""" + calls = {} + + def fake_run(c, cmd, path, pty, env): + calls["args"] = (c, cmd, path, pty, env) + + monkeypatch.setattr(tasks_mod, "run", fake_run) + # Control manage_py_dir to a deterministic path + monkeypatch.setattr(tasks_mod, "manage_py_dir", lambda: tmp_path / "src" / "backend" / "InvenTree") + tasks_mod.manage(object(), "migrate", pty=True, env={"X": "Y"}) + assert calls["args"][1] == "python3 manage.py migrate" + assert calls["args"][2] == tmp_path / "src" / "backend" / "InvenTree" + assert calls["args"][3] is True + assert calls["args"][4] == {"X": "Y"} + + +def test_yarn_invokes_run_in_frontend_dir(monkeypatch, tmp_path): + """Test yarn delegates to run in frontend directory.""" + calls = {} + + def fake_run(c, cmd, path, pty): + calls["args"] = (c, cmd, path, pty) + + monkeypatch.setattr(tasks_mod, "run", fake_run) + # Control local_dir + monkeypatch.setattr(tasks_mod, "local_dir", lambda: tmp_path) + tasks_mod.yarn(object(), "yarn build") + assert calls["args"][1] == "yarn build" + assert calls["args"][2] == tmp_path / "src" / "frontend" + assert calls["args"][3] is False + + +def test_node_available_all_present(monkeypatch, capsys): + """Test node_available returns True when both node and yarn are available.""" + def fake_check_output(args, stderr=None, shell=False): + cmd = args[0] if isinstance(args, list) else args + if "yarn --version" in cmd: + return b"1.22.19\n" + if "node --version" in cmd: + return b"v20.12.1\n" + raise FileNotFoundError() + + monkeypatch.setattr(tasks_mod.subprocess, "check_output", fake_check_output) + assert tasks_mod.node_available() is True + # No warning expected + out = capsys.readouterr().out + assert "Node is available but yarn is not" not in out + + +def test_node_available_node_only_warns(monkeypatch, capsys): + """Test node_available returns False and warns when only node is available.""" + def fake_check_output(args, stderr=None, shell=False): + cmd = args[0] if isinstance(args, list) else args + if "yarn --version" in cmd: + raise FileNotFoundError() + if "node --version" in cmd: + return b"v18.19.0\n" + raise FileNotFoundError() + + monkeypatch.setattr(tasks_mod.subprocess, "check_output", fake_check_output) + assert tasks_mod.node_available() is False + out = capsys.readouterr().out + assert "Node is available but yarn is not." in out + + +def test_node_available_bypass_yarn(monkeypatch, capsys): + """Test node_available bypasses yarn check when bypass_yarn is True.""" + def fake_check_output(args, stderr=None, shell=False): + cmd = args[0] if isinstance(args, list) else args + if "yarn --version" in cmd: + raise FileNotFoundError() + if "node --version" in cmd: + return b"v16.20.2\n" + raise FileNotFoundError() + + monkeypatch.setattr(tasks_mod.subprocess, "check_output", fake_check_output) + assert tasks_mod.node_available(bypass_yarn=True) is True + out = capsys.readouterr().out + # No warning because bypass_yarn=True + assert "Node is available but yarn is not." not in out + + +def test_node_available_versions_tuple(monkeypatch): + """Test node_available returns version tuple when requested.""" + def fake_check_output(args, stderr=None, shell=False): + cmd = args[0] if isinstance(args, list) else args + if "yarn --version" in cmd: + return b"1.22.22\n" + if "node --version" in cmd: + return b"v20.0.0\n" + raise FileNotFoundError() + + monkeypatch.setattr(tasks_mod.subprocess, "check_output", fake_check_output) + ok, node_v, yarn_v = tasks_mod.node_available(versions=True) + assert ok is True + assert node_v == "v20.0.0" + assert yarn_v == "1.22.22" + + +def test_check_file_existence_prompts_and_exits(monkeypatch, tmp_path, capsys): + """Test check_file_existence prompts and exits when file exists and overwrite is False.""" + # Ensure decorator does not print debug logs + monkeypatch.delenv("INVENTREE_DEBUG", raising=False) + monkeypatch.delenv("RUNNER_DEBUG", raising=False) + + f = tmp_path / "export.csv" + f.write_text("data", encoding="utf-8") + + # Simulate user input 'n' + inputs = iter(["n"]) + monkeypatch.setattr(builtins, "input", lambda _: next(inputs)) + + with pytest.raises(SystemExit): + tasks_mod.check_file_existence(f, overwrite=False) + + out = capsys.readouterr().out + assert "file already exists" in out.lower() + assert "Cancelled export operation" in out + + +def test_check_file_existence_overwrite_skips_prompt(monkeypatch, tmp_path, capsys): + """Test check_file_existence skips prompt when overwrite is True.""" + f = tmp_path / "export.csv" + f.write_text("data", encoding="utf-8") + + calls = {"asked": False} + + def fake_input(prompt): + calls["asked"] = True + return "y" + + monkeypatch.setattr(builtins, "input", fake_input) + tasks_mod.check_file_existence(f, overwrite=True) + out = capsys.readouterr().out + assert calls["asked"] is False # No prompt when overwrite=True + assert "Cancelled export operation" not in out + + +def test_check_file_existence_no_file(monkeypatch, tmp_path, capsys): + """Test check_file_existence does nothing when file does not exist.""" + f = tmp_path / "nope.csv" + if f.exists(): + f.unlink() + # Should silently pass + tasks_mod.check_file_existence(f, overwrite=False) + out = capsys.readouterr().out + assert out == "" or "# " in out # could have state_logger logs if debug is enabled \ No newline at end of file