tests for plugin upload from registry

core/cat/mad_hatter/plugin.py

@@ -169,7 +169,7 @@ def _load_hooks_and_tools(self):
         for py_file in self.py_files:
             py_filename = py_file.replace("/", ".").replace(".py", "")  # this is UGLY I know. I'm sorry
 
-            log(f"Import module {py_filename}", "WARNING")
+            log(f"Import module {py_filename}", "INFO")
 
             # save a reference to decorated functions
             try:

core/tests/mad_hatter/test_mad_hatter.py

@@ -99,10 +99,6 @@ def test_plugin_install(mad_hatter: MadHatter, plugin_is_flat):
     assert "core_plugin" in active_plugins
     assert "mock_plugin" in active_plugins
 
-    # remove plugin files (both zip and extracted)
-    os.remove(new_plugin_zip_path)
-    shutil.rmtree(os.path.join(mad_hatter.ccat.get_plugin_path(), "mock_plugin"))
-
 
 def test_plugin_uninstall_non_existent(mad_hatter: MadHatter):
     # should not throw error
@@ -141,6 +137,3 @@ def test_plugin_uninstall(mad_hatter: MadHatter, plugin_is_flat):
     active_plugins = mad_hatter.load_active_plugins_from_db()
     assert len(active_plugins) == 1
     assert active_plugins[0] == "core_plugin"
-
-    # remove also original zip file
-    os.remove(new_plugin_zip_path)

core/tests/routes/plugins/test_plugins_install_uninstall.py

@@ -1,6 +1,5 @@
 import os
 import time
-import pytest
 import shutil
 from tests.utils import get_embedded_tools
 from fixture_just_installed_plugin import just_installed_plugin
@@ -15,6 +14,13 @@ def test_plugin_install_from_zip(client, just_installed_plugin):
 
     #### PLUGIN IS ALREADY ACTIVE
 
+    # GET plugin endpoint responds
+    response = client.get(f"/plugins/mock_plugin")
+    assert response.status_code == 200
+    json = response.json()
+    assert json["data"]["id"] == "mock_plugin"
+    assert json["data"]["active"] == True
+
     # GET plugins endpoint lists the plugin
     response = client.get("/plugins")
     installed_plugins = response.json()["installed"]

core/tests/routes/plugins/test_plugins_registry.py

@@ -1,5 +1,6 @@
 import os
-from utils import get_embedded_tools
+import shutil
+from tests.utils import get_embedded_tools
 
 # TODO: registry responses here should be mocked, at the moment we are actually calling the service
 
@@ -39,65 +40,48 @@ def test_list_registry_plugins_by_query(client):
         assert params["query"] in plugin_text # verify searched text
 
 
-# TOOD: these tests are to be activated when also search by tag and author is activated in core
-'''
-def test_list_registry_plugins_by_author(client):
-
-    params = {
-        "author": "Nicola Corbellini"
-    }
-    response = client.get("/plugins", params=params)
-    json = response.json()
-
-    assert response.status_code == 200
-    assert json["filters"]["author"] == params["query"]
-    assert len(json["registry"]) > 0 # found registry plugins with author
-    for plugin in json["registry"]:
-        assert params["author"] in plugin["author_name"] # verify author
-
-
-def test_list_registry_plugins_by_tag(client):
-
-    params = {
-        "tag": "llm"
-    }
-    response = client.get("/plugins", params=params)
-    json = response.json()
-
-    assert response.status_code == 200
-    assert json["filters"]["tag"] == params["tag"]
-    assert len(json["registry"]) > 0 # found registry plugins with tag
-    for plugin in json["registry"]:
-        plugin_tags = plugin["tags"].split(", ")
-        assert params["tag"] in plugin_tags # verify tag
-'''
-
-
 def test_plugin_install_from_registry(client):
 
+    new_plugin_id = "ccat_summarization"
     # during tests, the cat uses a different folder for plugins
-    mock_plugin_final_folder = "tests/mocks/mock_plugin_folder/mock_plugin"
-
+    new_plugin_final_folder = f"tests/mocks/mock_plugin_folder/{new_plugin_id}"
+    if os.path.exists(new_plugin_final_folder):
+        shutil.rmtree(new_plugin_final_folder)
+    assert not os.path.exists(new_plugin_final_folder)
+
+    # install plugin from registry
+    payload = {
+        "url": "https://github.com/nicola-corbellini/ccat_summarization"
+    }
+    response = client.post("/plugins/upload/registry", json=payload)
+    assert response.status_code == 200
+    assert response.json()["url"] == payload["url"]
+    assert response.json()["info"] == "Plugin is being installed asynchronously"
 
+    # GET plugin endpoint responds
+    response = client.get(f"/plugins/{new_plugin_id}")
+    assert response.status_code == 200
+    json = response.json()
+    assert json["data"]["id"] == new_plugin_id
+    assert json["data"]["active"] == True
 
     # GET plugins endpoint lists the plugin
     response = client.get("/plugins")
+    assert response.status_code == 200
     installed_plugins = response.json()["installed"]
     installed_plugins_names = list(map(lambda p: p["id"], installed_plugins))
-    assert "mock_plugin" in installed_plugins_names
-    # both core_plugin and mock_plugin are active
+    assert new_plugin_id in installed_plugins_names
+    # both core_plugin and new_plugin are active
     for p in installed_plugins:
         assert p["active"] == True
 
     # plugin has been actually extracted in (mock) plugins folder
-    assert os.path.exists(mock_plugin_final_folder)
+    assert os.path.exists(new_plugin_final_folder)
 
-    # check whether new tool has been embedded
-    tools = get_embedded_tools(client)
-    assert len(tools) == 2
-    tool_names = list(map(lambda t: t["metadata"]["name"], tools))
-    assert "mock_tool" in tool_names
-    assert "get_the_time" in tool_names # from core_plugin
+    # TODO: check for tools and hooks creation
+
+    # cleanup
+    shutil.rmtree(new_plugin_final_folder)
 
 
 # take away from the list of availbale registry plugins, the ones that are already installed
@@ -117,3 +101,37 @@ def test_list_registry_plugins_without_duplicating_installed_plugins(client):
     assert response.status_code == 200
     # TODO plugin compares in installed!!!
     # TODO plugin does not appear in registry!!!
+
+
+# TOOD: these tests are to be activated when also search by tag and author is activated in core
+'''
+def test_list_registry_plugins_by_author(client):
+
+    params = {
+        "author": "Nicola Corbellini"
+    }
+    response = client.get("/plugins", params=params)
+    json = response.json()
+
+    assert response.status_code == 200
+    assert json["filters"]["author"] == params["query"]
+    assert len(json["registry"]) > 0 # found registry plugins with author
+    for plugin in json["registry"]:
+        assert params["author"] in plugin["author_name"] # verify author
+
+
+def test_list_registry_plugins_by_tag(client):
+
+    params = {
+        "tag": "llm"
+    }
+    response = client.get("/plugins", params=params)
+    json = response.json()
+
+    assert response.status_code == 200
+    assert json["filters"]["tag"] == params["tag"]
+    assert len(json["registry"]) > 0 # found registry plugins with tag
+    for plugin in json["registry"]:
+        plugin_tags = plugin["tags"].split(", ")
+        assert params["tag"] in plugin_tags # verify tag
+'''