fix(backend): registry install command

src/backend/InvenTree/plugin/installer.py

@@ -3,8 +3,10 @@
 import re
 import subprocess
 import sys
+from typing import Optional
 
 from django.conf import settings
+from django.contrib.auth.models import User
 from django.core.exceptions import ValidationError
 from django.utils.translation import gettext_lazy as _
 
@@ -158,18 +160,12 @@ def install_plugins_file():
     return True
 
 
-def update_plugins_file(install_name, full_package=None, version=None, remove=False):
+def update_plugins_file(package_reference: str, remove: bool = False):
     """Add a plugin to the plugins file."""
     if remove:
-        logger.info('Removing plugin from plugins file: %s', install_name)
+        logger.info('Removing plugin from plugins file: %s', package_reference)
     else:
-        logger.info('Adding plugin to plugins file: %s', install_name)
-
-    # If a full package name is provided, use that instead
-    if full_package and full_package != install_name:
-        new_value = full_package
-    else:
-        new_value = f'{install_name}=={version}' if version else install_name
+        logger.info('Adding plugin to plugins file: %s', package_reference)
 
     pf = settings.PLUGIN_FILE
 
@@ -179,7 +175,7 @@ def update_plugins_file(install_name, full_package=None, version=None, remove=Fa
 
     def compare_line(line: str):
         """Check if a line in the file matches the installname."""
-        return re.match(rf'^{install_name}[\s=@]', line.strip())
+        return re.match(rf'^{package_reference}[\s=@]', line.strip())
 
     # First, read in existing plugin file
     try:
@@ -206,13 +202,13 @@ def compare_line(line: str):
             found = True
             if not remove:
                 # Replace line with new install name
-                output.append(new_value)
+                output.append(package_reference)
         else:
             output.append(line)
 
     # Append plugin to file
     if not found and not remove:
-        output.append(new_value)
+        output.append(package_reference)
 
     # Write file back to disk
     try:
@@ -227,13 +223,18 @@ def compare_line(line: str):
         log_error('update_plugins_file', scope='plugins')
 
 
-def install_plugin(url=None, packagename=None, user=None, version=None):
+def install_plugin(
+    user: Optional[User] = None,
+    url: Optional[str] = None,
+    packagename: Optional[str] = None,
+    version: Optional[str] = None,
+):
     """Install a plugin into the python virtual environment.
 
     Args:
+        user: user performing the installation
         packagename: Optional package name to install
         url: Optional URL to install from
-        user: Optional user performing the installation
         version: Optional version specifier
     """
     if user and not user.is_superuser:
@@ -245,44 +246,45 @@ def install_plugin(url=None, packagename=None, user=None, version=None):
     logger.info('install_plugin: %s, %s', url, packagename)
 
     # build up the command
-    install_name = ['install', '-U', '--disable-pip-version-check']
-
-    full_pkg = ''
+    package_ref: Optional[str] = None
+    index_url = None
 
     if url:
-        # use custom registration / VCS
-        if True in [
-            identifier in url for identifier in ['git+https', 'hg+https', 'svn+svn']
-        ]:
+        # VCS based install - this can just be a VCS reference
+        if url.startswith(('git+https://', 'hg+https://', 'svn+svn://')):
             # using a VCS provider
-            full_pkg = f'{packagename}@{url}' if packagename else url
-        elif url:
-            install_name.append('-i')
-            full_pkg = url
+            package_ref = f'{packagename}@{url}' if packagename else url
+        # http based index reference
+        elif url.startswith(('http://', 'https://')) and packagename:
+            package_ref = packagename
+            index_url = url
+        # Ignore url and just use default index
         elif packagename:
-            full_pkg = packagename
+            package_ref = packagename
+        else:
+            raise ValidationError(_('Invalid URL and no package name provided'))
 
     elif packagename:
-        # use pypi
-        full_pkg = packagename
+        # use default index - most often pypi
+        package_ref = packagename
 
         if version:
-            full_pkg = f'{full_pkg}=={version}'
-
-    if not full_pkg:
+            package_ref = f'{packagename}=={version}'
+    else:
         raise ValidationError(_('No package name or URL provided for installation'))
 
     # Sanitize the package name for installation
-    if any(c in full_pkg for c in ';&|`$()'):
+    if any(c in package_ref for c in ';&|`$()'):
         raise ValidationError(_('Invalid characters in package name or URL'))
 
-    install_name.append(full_pkg)
+    # Execute installation via pip
+    cmd: list[str] = ['install', '-U', '--disable-pip-version-check']
+    if index_url:
+        cmd += ['-i', index_url]
 
     ret = {}
-
-    # Execute installation via pip
     try:
-        result = pip_command(*install_name)
+        result = pip_command(*cmd, package_ref)
 
         ret['result'] = ret['success'] = _('Installed plugin successfully')
         ret['output'] = str(result, 'utf-8')
@@ -299,7 +301,7 @@ def install_plugin(url=None, packagename=None, user=None, version=None):
 
     if version := ret.get('version'):
         # Save plugin to plugins file
-        update_plugins_file(packagename, full_package=full_pkg, version=version)
+        update_plugins_file(package_reference=package_ref)
 
         # Reload the plugin registry, to discover the new plugin
         from plugin.registry import registry
@@ -389,7 +391,7 @@ def uninstall_plugin(cfg: plugin.models.PluginConfig, user=None, delete_config=T
         raise ValidationError(_('Plugin installation not found'))
 
     # Update the plugins file
-    update_plugins_file(package_name, remove=True)
+    update_plugins_file(package_reference=package_name, remove=True)
 
     if delete_config:
         logger.info('Deleting plugin configuration from database: %s', cfg.key)

src/backend/InvenTree/plugin/test_api.py

@@ -126,6 +126,15 @@ def test_plugin_install(self):
         ).data
         self.assertEqual(data['success'], 'Installed plugin successfully')
 
+        # valid (kindoff) - Pypi and onsense uri
+        data = self.post(
+            url,
+            {'confirm': True, 'packagename': self.PKG_NAME, 'url': 'lol://example.com'},
+            expected_code=201,
+            max_query_time=30,
+            max_query_count=450,
+        ).data
+
         # invalid tries
         # no input
         data = self.post(url, {}, expected_code=400).data
@@ -163,6 +172,17 @@ def test_plugin_install(self):
                 str(response.data['non_field_errors']),
             )
 
+        # plugin - normal user should not be able to install plugins
+        self.user.is_staff = False
+        self.user.save()
+        self.post(
+            url,
+            {'confirm': True, 'packagename': self.PKG_NAME},
+            expected_code=400,
+            max_query_time=30,
+            max_query_count=450,
+        )
+
     def test_plugin_deactivate_mandatory(self):
         """Test deactivating a mandatory plugin."""
         self.user.is_superuser = True
@@ -673,6 +693,33 @@ def test_full_process(self):
             with self.assertRaises(PluginConfig.DoesNotExist):
                 PluginConfig.objects.get(key=slug)
 
+    @override_settings(PLUGIN_TESTING_SETUP=True)
+    def test_registry(self):
+        """Test install with a custom registry."""
+        plrg_name = 'inventree-dummy-app-plugin'
+
+        # install - python repository url and package name
+        data = self.post(
+            reverse('api-plugin-install'),
+            {
+                'confirm': True,
+                'url': 'https://git.invenhost.com/api/packages/invenhost-c1/pypi/simple/',
+                'packagename': plrg_name,
+            },
+            expected_code=201,
+            max_query_count=450,
+            max_query_time=30,
+        ).data
+        self.assertEqual(data['success'], 'Installed plugin successfully')
+
+        # # and uninstall it again to clean up
+        # response = self.patch(
+        #     reverse('api-plugin-uninstall', kwargs={'plugin': plrg_name}),
+        #     data={'delete_config': True},
+        #     max_query_count=350,
+        # )
+        # self.assertEqual(response.status_code, 200)
+
 
 class PluginLockedSettingsTest(PluginMixin, InvenTreeAPITestCase):
     """Tests for locked plugin settings (overridden via configuration).