[BUG] Plugin Schedule do not update when beeing changed + plugin testing (#4243)

* Add test for api_call

* Add coverage for LocateMixin

* no cov for no url

* make sure changed details get updated

* restructure code

* Test that changes in schedules are reflected
Fixes #4239
This commit is contained in:
Matthias Mair 2023-01-24 23:41:20 +01:00 committed by GitHub
parent 428e939b1d
commit af0bc90e48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 45 additions and 31 deletions

View File

@ -153,43 +153,34 @@ class ScheduleMixin:
for key, task in self.scheduled_tasks.items():
task_name = self.get_task_name(key)
if Schedule.objects.filter(name=task_name).exists():
# Scheduled task already exists - continue!
continue # pragma: no cover
logger.info(f"Adding scheduled task '{task_name}'")
obj = {
'name': task_name,
'schedule_type': task['schedule'],
'minutes': task.get('minutes', None),
'repeats': task.get('repeats', -1),
}
func_name = task['func'].strip()
if '.' in func_name:
"""Dotted notation indicates that we wish to run a globally defined function, from a specified Python module."""
Schedule.objects.create(
name=task_name,
func=func_name,
schedule_type=task['schedule'],
minutes=task.get('minutes', None),
repeats=task.get('repeats', -1),
)
obj['func'] = func_name
else:
"""
Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin.
This is managed by the plugin registry itself.
"""
"""Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin. This is managed by the plugin registry itself."""
slug = self.plugin_slug()
obj['func'] = registry.call_plugin_function
obj['args'] = f"'{slug}', '{func_name}'"
Schedule.objects.create(
name=task_name,
func=registry.call_plugin_function,
args=f"'{slug}', '{func_name}'",
schedule_type=task['schedule'],
minutes=task.get('minutes', None),
repeats=task.get('repeats', -1),
)
if Schedule.objects.filter(name=task_name).exists():
# Scheduled task already exists - update it!
logger.info(f"Updating scheduled task '{task_name}'")
instance = Schedule.objects.get(name=task_name)
for item in obj:
setattr(instance, item, obj[item])
instance.save()
else:
logger.info(f"Adding scheduled task '{task_name}'")
# Create a new scheduled task
Schedule.objects.create(**obj)
except (ProgrammingError, OperationalError): # pragma: no cover
# Database might not yet be ready

View File

@ -271,6 +271,11 @@ class APICallMixinTest(BaseMixinDefinition, TestCase):
self.assertTrue(result)
self.assertEqual(result['name'], 'morpheus')
# api_call with endpoint with leading slash
result = self.mixin.api_call('/orgs/inventree', simple_response=False)
self.assertTrue(result)
self.assertEqual(result.reason, 'OK')
# api_call with filter
result = self.mixin.api_call('repos/inventree/InvenTree/stargazers', url_args={'page': '2'})
self.assertTrue(result)

View File

@ -55,7 +55,7 @@ class InvenTreeInternalBarcodePlugin(BarcodeMixin, InvenTreePlugin):
url = instance.get_absolute_url()
data['web_url'] = url
else:
url = None
url = None # pragma: no cover
response = {
label: data

View File

@ -31,6 +31,19 @@ class ExampleScheduledTaskPluginTests(TestCase):
scheduled_plugin_tasks = Schedule.objects.filter(name__istartswith="plugin.")
self.assertEqual(len(scheduled_plugin_tasks), 3)
# test updating the schedule
hello_schedule = Schedule.objects.get(name='plugin.schedule.hello')
self.assertEqual(hello_schedule.minutes, 45)
# change the schedule and reregister
plg.scheduled_tasks['hello']['minutes'] = 15
plg.register_tasks()
# Check that the schedule was updated
hello_schedule = Schedule.objects.get(name='plugin.schedule.hello')
scheduled_plugin_tasks = Schedule.objects.filter(name__istartswith="plugin.")
self.assertEqual(hello_schedule.minutes, 15)
self.assertEqual(len(scheduled_plugin_tasks), 3)
# delete middle task
# this is to check the system also deals with disappearing tasks
scheduled_plugin_tasks[1].delete()

View File

@ -51,9 +51,14 @@ class SampleLocatePlugintests(InvenTreeAPITestCase):
def test_mixin(self):
"""Test that MixinNotImplementedError is raised."""
# Test location locator
with self.assertRaises(MixinNotImplementedError):
class Wrong(LocateMixin, InvenTreePlugin):
pass
plugin = Wrong()
plugin.locate_stock_location(1)
# Test item locator
with self.assertRaises(MixinNotImplementedError):
plugin.locate_stock_item(1)