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_update_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)
status.description = 'status description'
management.call_command('runplugins', stdout=out)
self.assertIn("Running test Production:Fake plugin", out.getvalue())
self.assertIn("Updated Fake plugin:Test status", out.getvalue())
status_model = models.Status.objects.all()[0]
self.assertEqual(status_model.description, status['description'])
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
serializer_class = serializers.PluginSerializer
class PluginInstanceViewSet(viewsets.ModelViewSet):
"""API view for PluginInstance."""
queryset = models.PluginInstance.objects.all()
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
serializer_class = serializers.PluginInstanceSerializer
class StatusViewSet(viewsets.ModelViewSet):
"""API view for Status."""
queryset = models.Status.objects.all()
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
serializer_class = serializers.StatusSerializer
@an_decorators.render_to('index.html')
def index(request):
"""Show the list of components."""
components = models.Component.objects.all()
return {'components': components}
@an_decorators.render_to('status.html')
def status(request, component_id):
"""Display a specific component."""
component = shortcuts.get_object_or_404(models.Component, pk=component_id)
base_url = "{0}://{1}".format(request.scheme, request.get_host())
@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 save(self, *args, **kwargs):
"""Save the object."""
if self.time_end is None:
self.time_end = _default_time_end()
super(Status, self).save(*args, **kwargs)
class PluginInstance(models.Model):
"""Instance of a plugin (including settings)."""
name = models.CharField(max_length=64)
description = models.TextField(max_length=1024, blank=True, null=True)
plugin = models.ForeignKey(Plugin, on_delete=models.CASCADE)
statuses = models.ManyToManyField(Status, blank=True)
config = picklefield.PickledObjectField(null=True)
created_on = models.DateTimeField(auto_now_add=True)
modified_on = models.DateTimeField(auto_now=True)
success = models.PositiveIntegerField(default=0)
failure = models.PositiveIntegerField(default=0)
success_on = models.DateTimeField(null=True, blank=True)
failure_on = models.DateTimeField(null=True, blank=True)
@property
def component(self):
"""Return the component associated with this instance."""
components = self.component_set.all()
if not components:
return None
return components[0]
"""Serializer for PluginInstance."""
class Meta:
"""Configuration."""
model = models.PluginInstance
fields = '__all__'
class StatusSerializer(serializers.HyperlinkedModelSerializer):
"""Serializer for Status."""
class Meta:
"""Configuration."""
model = models.Status
fields = '__all__'
# Bellow are serializers for the simple /defcon/ API.
class PluginInstanceFullSerializer(serializers.HyperlinkedModelSerializer):
"""Serializer for PluginInstance with expanded content."""
statuses = StatusSerializer(many=True)
class Meta:
"""Configuration."""
model = models.PluginInstance
fields = '__all__'
def _default_time_end():
return timezone.now() + Status.DEFAULT_ACTIVE_DURATION
def _save_status(self, plugin_obj, status_id, status):
"""Save a status."""
try:
del status['id']
status_obj, created = models.Status.objects.update_or_create(
id=status_id, defaults=status)
# Because statuses can be shared between plugins, we always add them
# to the plugin, not only when they are created.
plugin_obj.statuses.add(status_obj)
except Exception:
msg = "Failed to save status with id #%s" % status_id
logging.exception(msg)
self.stderr.write(self.style.ERROR(msg))
else:
action = 'Created' if created else 'Updated'
self.stdout.write(self.style.SUCCESS(
'%s %s:%s config (%s)' % (action, plugin_obj.plugin.name,
status_obj.title,
status_obj.defcon)))
def save(self, *args, **kwargs):
"""Save the object."""
if self.time_end is None:
self.time_end = _default_time_end()
super(Status, self).save(*args, **kwargs)
# 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)