mirror of
https://github.com/alicevision/Meshroom.git
synced 2025-06-16 01:31:58 +02:00
98fbfae013
On some systems such as GitHub's Windows CI, the `write` operation is too fast and does not cause a change in the timestamp of the file we're reloading, hence causing the test to fail for external reasons. Adding a sleep does not change anything to the test functionally, but on the contrary ensures that we are actually testing the feature. https://stackoverflow.com/questions/19059877/python-os-path-getmtime-time-not-changing
215 lines
8.6 KiB
Python
215 lines
8.6 KiB
Python
# coding:utf-8
|
|
|
|
from meshroom.core import desc, pluginManager, loadClassesNodes
|
|
from meshroom.core.plugins import NodePluginStatus, Plugin
|
|
|
|
import os
|
|
import time
|
|
|
|
class TestPluginWithValidNodesOnly:
|
|
plugin = None
|
|
|
|
@classmethod
|
|
def setup_class(cls):
|
|
folder = os.path.join(os.path.dirname(__file__), "plugins", "meshroom")
|
|
package = "pluginA"
|
|
cls.plugin = Plugin(package, folder)
|
|
nodes = loadClassesNodes(folder, package)
|
|
for node in nodes:
|
|
cls.plugin.addNodePlugin(node)
|
|
pluginManager.addPlugin(cls.plugin)
|
|
|
|
@classmethod
|
|
def teardown_class(cls):
|
|
for node in cls.plugin.nodes.values():
|
|
pluginManager.unregisterNode(node)
|
|
cls.plugin = None
|
|
|
|
def test_loadedPlugin(self):
|
|
# Assert that there are loaded plugins, and that "pluginA" is one of them
|
|
assert len(pluginManager.getPlugins()) >= 1
|
|
plugin = pluginManager.getPlugin("pluginA")
|
|
assert plugin == self.plugin
|
|
assert str(plugin.path) == os.path.join(os.path.dirname(__file__), "plugins", "meshroom")
|
|
|
|
# Assert that the nodes of pluginA have been successfully registered
|
|
assert len(pluginManager.getRegisteredNodePlugins()) >= 2
|
|
for nodeName, nodePlugin in plugin.nodes.items():
|
|
assert nodePlugin.status == NodePluginStatus.LOADED
|
|
assert pluginManager.isRegistered(nodeName)
|
|
|
|
# Assert the template has been loaded
|
|
assert len(plugin.templates) == 1
|
|
name = list(plugin.templates.keys())[0]
|
|
assert name == "sharedTemplate"
|
|
assert plugin.templates[name] == os.path.join(str(plugin.path), "sharedTemplate.mg")
|
|
|
|
def test_unloadPlugin(self):
|
|
plugin = pluginManager.getPlugin("pluginA")
|
|
assert plugin == self.plugin
|
|
|
|
# Unload the plugin without unregistering the nodes
|
|
pluginManager.removePlugin(plugin, unregisterNodePlugins=False)
|
|
|
|
# Assert the plugin is not loaded anymore
|
|
assert pluginManager.getPlugin(plugin.name) is None
|
|
|
|
# Assert the nodes are still registered and belong to an unloaded plugin
|
|
for nodeName, nodePlugin in plugin.nodes.items():
|
|
assert nodePlugin.status == NodePluginStatus.LOADED
|
|
assert pluginManager.isRegistered(nodeName)
|
|
assert pluginManager.belongsToPlugin(nodeName) is None
|
|
|
|
# Re-add the plugin
|
|
pluginManager.addPlugin(plugin, registerNodePlugins=False)
|
|
assert pluginManager.getPlugin(plugin.name)
|
|
|
|
# Unload the plugin with a full unregistration of the nodes
|
|
pluginManager.removePlugin(plugin)
|
|
|
|
# Assert the plugin is not loaded anymore
|
|
assert pluginManager.getPlugin(plugin.name) is None
|
|
|
|
# Assert the nodes have been successfully unregistered
|
|
for nodeName, nodePlugin in plugin.nodes.items():
|
|
assert nodePlugin.status == NodePluginStatus.NOT_LOADED
|
|
assert not pluginManager.isRegistered(nodeName)
|
|
|
|
# Re-add the plugin and re-register the nodes
|
|
pluginManager.addPlugin(plugin)
|
|
assert pluginManager.getPlugin(plugin.name)
|
|
for nodeName, nodePlugin in plugin.nodes.items():
|
|
assert nodePlugin.status == NodePluginStatus.LOADED
|
|
assert pluginManager.isRegistered(nodeName)
|
|
|
|
def test_updateRegisteredNodes(self):
|
|
nbRegisteredNodes = len(pluginManager.getRegisteredNodePlugins())
|
|
plugin = pluginManager.getPlugin("pluginA")
|
|
assert plugin == self.plugin
|
|
nodeA = pluginManager.getRegisteredNodePlugin("PluginANodeA")
|
|
nodeAName = nodeA.nodeDescriptor.__name__
|
|
|
|
# Unregister a node
|
|
assert nodeA
|
|
pluginManager.unregisterNode(nodeA)
|
|
|
|
# Check that the node has been fully unregistered:
|
|
# - its status is "NOT_LOADED"
|
|
# - it is still part of pluginA
|
|
# - it is not in the list of registered plugins anymore (and returns None when requested)
|
|
assert nodeA.status == NodePluginStatus.NOT_LOADED
|
|
assert plugin.containsNodePlugin(nodeAName)
|
|
assert nodeA.plugin == plugin
|
|
|
|
assert pluginManager.getRegisteredNodePlugin(nodeAName) is None
|
|
assert nodeAName not in pluginManager.getRegisteredNodePlugins()
|
|
assert len(pluginManager.getRegisteredNodePlugins()) == nbRegisteredNodes - 1
|
|
|
|
# Re-register the node
|
|
pluginManager.registerNode(nodeA)
|
|
|
|
assert nodeA.status == NodePluginStatus.LOADED
|
|
assert pluginManager.getRegisteredNodePlugin(nodeAName)
|
|
assert len(pluginManager.getRegisteredNodePlugins()) == nbRegisteredNodes
|
|
|
|
|
|
class TestPluginWithInvalidNodes:
|
|
plugin = None
|
|
|
|
@classmethod
|
|
def setup_class(cls):
|
|
folder = os.path.join(os.path.dirname(__file__), "plugins", "meshroom")
|
|
package = "pluginB"
|
|
cls.plugin = Plugin(package, folder)
|
|
nodes = loadClassesNodes(folder, package)
|
|
for node in nodes:
|
|
cls.plugin.addNodePlugin(node)
|
|
pluginManager.addPlugin(cls.plugin)
|
|
|
|
@classmethod
|
|
def teardown_class(cls):
|
|
for node in cls.plugin.nodes.values():
|
|
pluginManager.unregisterNode(node)
|
|
cls.plugin = None
|
|
|
|
def test_loadedPlugin(self):
|
|
# Assert that there are loaded plugins, and that "pluginB" is one of them
|
|
assert len(pluginManager.getPlugins()) >= 1
|
|
plugin = pluginManager.getPlugin("pluginB")
|
|
assert plugin == self.plugin
|
|
assert str(plugin.path) == os.path.join(os.path.dirname(__file__), "plugins", "meshroom")
|
|
|
|
# Assert that PluginBNodeA is successfully registered
|
|
assert pluginManager.isRegistered("PluginBNodeA")
|
|
assert plugin.nodes["PluginBNodeA"].status == NodePluginStatus.LOADED
|
|
assert plugin.nodes["PluginBNodeA"].plugin == plugin
|
|
|
|
# Assert that PluginBNodeB has not been registered (description error)
|
|
assert not pluginManager.isRegistered("PluginBNodeB")
|
|
assert plugin.nodes["PluginBNodeB"].status == NodePluginStatus.DESC_ERROR
|
|
assert plugin.nodes["PluginBNodeB"].plugin == plugin
|
|
|
|
# Assert the template has been loaded
|
|
assert len(plugin.templates) == 1
|
|
name = list(plugin.templates.keys())[0]
|
|
assert name == "sharedTemplate"
|
|
assert plugin.templates[name] == os.path.join(str(plugin.path), "sharedTemplate.mg")
|
|
|
|
def test_reloadNodePlugin(self):
|
|
plugin = pluginManager.getPlugin("pluginB")
|
|
assert plugin == self.plugin
|
|
node = plugin.nodes["PluginBNodeB"]
|
|
nodeName = node.nodeDescriptor.__name__
|
|
|
|
# Check that the node has not been registered
|
|
assert node.status == NodePluginStatus.DESC_ERROR
|
|
assert not pluginManager.isRegistered(nodeName)
|
|
|
|
# Check that the node cannot be registered
|
|
pluginManager.registerNode(node)
|
|
assert not pluginManager.isRegistered(nodeName)
|
|
|
|
# Replace directly in the node file the line that fails the validation
|
|
# on the description with a line that will pass
|
|
originalFileContent = None
|
|
with open(node.path, "r") as f:
|
|
originalFileContent = f.read()
|
|
|
|
replaceFileContent = originalFileContent.replace('"not an integer"', '1')
|
|
with open(node.path, "w") as f:
|
|
f.write(replaceFileContent)
|
|
|
|
# Reload the node and assert it is valid
|
|
node.reload()
|
|
assert node.status == NodePluginStatus.NOT_LOADED
|
|
|
|
# Attempt to register node plugin
|
|
pluginManager.registerNode(node)
|
|
assert pluginManager.isRegistered(nodeName)
|
|
|
|
# Reload the node again without any change
|
|
node.reload()
|
|
assert pluginManager.isRegistered(nodeName)
|
|
|
|
# Hack to ensure that the timestamp of the file will be different after being rewritten
|
|
# Without it, on some systems, the operation is too fast and the timestamp does not change,
|
|
# cause the test to fail
|
|
time.sleep(0.1)
|
|
|
|
# Restore the node file to its original state (with a description error)
|
|
with open(node.path, "w") as f:
|
|
f.write(originalFileContent)
|
|
|
|
timestampOr2 = os.path.getmtime(node.path)
|
|
print(f"New timestamp: {timestampOr2}")
|
|
print(os.stat(node.path))
|
|
|
|
# Reload the node and assert it is invalid while still registered
|
|
node.reload()
|
|
assert node.status == NodePluginStatus.DESC_ERROR
|
|
assert pluginManager.isRegistered(nodeName)
|
|
|
|
# Unregister it
|
|
pluginManager.unregisterNode(node)
|
|
assert node.status == NodePluginStatus.DESC_ERROR # Not NOT_LOADED
|
|
assert not pluginManager.isRegistered(nodeName)
|