Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_run_add_status(self):
out = StringIO()
self.addCleanup(out.close)
status = base.Status('Test status', 5, 'http://foo/#5')
with self.components_with_plugin(status):
management.call_command('runplugins', stdout=out)
self.assertIn("Running test Production:Fake plugin", out.getvalue())
self.assertIn("Created Fake plugin:Test status", out.getvalue())
status_model = models.Status.objects.all()[0]
self.assertEqual(status_model.title, status['title'])
self.assertEqual(status_model.link, status['link'])
self.assertEqual(status_model.description, status['description'])
self.assertEqual(status_model.defcon, status['defcon'])
self.assertFalse(status_model.override)
def test_update_plugin(self):
out = StringIO()
self.addCleanup(out.close)
management.call_command('loadplugins', stdout=out)
with mock.patch.object(FakePlugin, "name", "Updated Fake"):
management.call_command('loadplugins', stdout=out)
plugin = FakePlugin()
self.assertIn("Updated %s" % plugin.name, out.getvalue())
plugin_model = models.Plugin.objects.get(id=plugin.short_name)
self.assertEqual(plugin_model.name, plugin.name)
@admin.register(models.Component)
class ComponentAdmin(admin.ModelAdmin):
"""Admin for Component."""
list_display = ('name', 'link', 'contact')
@admin.register(models.Plugin)
class PluginAdmin(admin.ModelAdmin):
"""Admin for Plugin."""
list_display = ('name', 'link')
@admin.register(models.PluginInstance)
class PluginInstanceAdmin(admin.ModelAdmin):
"""Admin for PluginInstance."""
list_display = ('name', 'plugin', 'component', 'created_on', 'modified_on')
@admin.register(models.Status)
class StatusAdmin(admin.ModelAdmin):
"""Admin for Status."""
list_display = ('id', 'title', 'created_on', 'modified_on', 'active')
def handle(self, *args, **options):
"""Run the command."""
# TODO: add filters by components and plugins.
# Get the status that haven't expired yet to make them expire if
# they disappeared.
now = timezone.now()
existing_statuses = set(models.Status.objects.filter(
time_end__gte=now).values_list('id', flat=True))
updated_statuses = set()
for component_obj in models.Component.objects.all():
for plugin_obj in component_obj.plugins.all():
updated_statuses |= self.run_plugin(component_obj, plugin_obj)
expired_statuses = (existing_statuses - updated_statuses)
if expired_statuses:
self.stdout.write(self.style.SUCCESS('Expiring %s' % expired_statuses))
models.Status.objects.filter(id__in=expired_statuses).update(time_end=now)