Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
""" testing compound fields list and dict """
from __future__ import unicode_literals
from django.test import TestCase
from mongoengine import Document, fields
from rest_framework_mongoengine.serializers import DocumentSerializer
from .models import DumbEmbedded
from .utils import dedent
class BasicCompoundDoc(Document):
list_field = fields.ListField()
int_list_field = fields.ListField(fields.IntField())
dict_field = fields.DictField()
int_dict_field = fields.DictField(field=fields.IntField())
int_map_field = fields.MapField(fields.IntField())
class OptionsCompoundDoc(Document):
int_list_field = fields.ListField(fields.IntField(min_value=3, max_value=7))
class NestedCompoundDoc(Document):
dict_list_field = fields.ListField(fields.DictField())
list_dict_field = fields.MapField(fields.ListField())
list_dict_list_field = fields.ListField(fields.MapField(fields.ListField()))
subject = StringField(required=True)
description = StringField()
start = DateTimeField()
stop = DateTimeField()
is_completed = BooleanField(default=False)
contacts = StringField()
suppress_alarms = BooleanField()
# Escalate TT during maintenance
escalate_managed_object = ForeignKeyField(ManagedObject)
# Time pattern when maintenance is active
# None - active all the time
time_pattern = ForeignKeyField(TimePattern)
# Objects declared to be affected by maintenance
direct_objects = ListField(EmbeddedDocumentField(MaintenanceObject))
# Segments declared to be affected by maintenance
direct_segments = ListField(EmbeddedDocumentField(MaintenanceSegment))
# All objects affected by maintenance
affected_objects = ListField(EmbeddedDocumentField(MaintenanceObject))
# Escalated TT ID in form
# :
escalation_tt = StringField(required=False)
# @todo: Attachments
_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
@classmethod
@cachetools.cachedmethod(operator.attrgetter("_id_cache"), lock=lambda _: id_lock)
def get_by_id(cls, id):
return Maintenance.objects.filter(id=id).first()
def save(self, *args, **kwargs):
if self.direct_objects:
class AttackTechnique(Document):
tactics = ListField(ReferenceField(AttackTactic))
technique_id = StringField(unique=True)
name = StringField()
description = StringField()
url = URLField()
class TechniqueMapping(EmbeddedDocument):
LOW = 'Low'
MODERATE = 'Moderate'
HIGH = 'High'
technique = ReferenceField(AttackTechnique)
tactics = ListField(ReferenceField(AttackTactic))
# Likely will ignore this field
level = StringField(choices=[LOW, MODERATE, HIGH])
# tactics should be a SET, but we are not strictly enforcing that for now
class TacticSet(Document):
tactics = ListField(ReferenceField(AttackTactic))
def intersect(self, other, new=False):
if isinstance(other, TacticSet):
tactics = other.tactics
else:
tactics = other
result = {tactic for tactic in self.tactics if tactic in tactics}
"strict": False,
"auto_create_index": False,
"indexes": ["managed_object"],
}
managed_object = ForeignKeyField(ManagedObject)
type = StringField(
choices=[(x, x) for x in ("table", "bridge", "vrf", "vll", "vpls", "evpn", "vxlan")],
default="table",
)
virtual_router = StringField(required=False)
name = StringField()
# VRF/VPLS
vpn_id = StringField(required=False)
rd = StringField(required=False)
rt_export = ListField(StringField(required=False))
rt_import = ListField(StringField(required=False))
def __str__(self):
return "%s: %s" % (self.managed_object.name, self.name if self.name else "default")
def iter_changed_datastream(self, changed_fields=None):
if config.datastream.enable_managedobject:
yield "managedobject", self.managed_object.id
def delete(self, *args, **kwargs):
# Delete subinterfaces
for si in self.subinterface_set.all():
si.delete()
# Delete forwarding instance
super(ForwardingInstance, self).delete(*args, **kwargs)
@property
class AddressProfile(Document):
meta = {"collection": "addressprofiles", "strict": False, "auto_create_index": False}
name = StringField(unique=True)
description = StringField()
# Address workflow
workflow = PlainReferenceField(Workflow)
style = ForeignKeyField(Style)
# Template.subject to render Address.name
name_template = ForeignKeyField(Template)
# Template.subject to render Address.fqdn
fqdn_template = ForeignKeyField(Template)
# Send seen event to prefix
seen_propagation_policy = StringField(choices=[("E", "Enable"), ("D", "Disable")], default="D")
#
tags = ListField(StringField())
# Integration with external NRI and TT systems
# Reference to remote system object has been imported from
remote_system = PlainReferenceField(RemoteSystem)
# Object id in remote system
remote_id = StringField()
# Object id in BI
bi_id = LongField(unique=True)
_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
_name_cache = cachetools.TTLCache(maxsize=100, ttl=60)
_bi_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
def __str__(self):
return self.name
@classmethod
@bi_sync
@on_delete_check(check=[("vc.VPN", "parent"), ("vc.VLAN", "vpn")])
@workflow
class VPN(Document):
meta = {"collection": "vpns", "strict": False, "auto_create_index": False}
name = StringField(unique=True)
profile = PlainReferenceField(VPNProfile)
description = StringField()
state = PlainReferenceField(State)
# Link to parent overlay
parent = PlainReferenceField("self")
project = ForeignKeyField(Project)
route_target = ListField(EmbeddedDocumentField(RouteTargetItem))
tags = ListField(StringField())
# Integration with external NRI and TT systems
# Reference to remote system object has been imported from
remote_system = PlainReferenceField(RemoteSystem)
# Object id in remote system
remote_id = StringField()
# Object id in BI
bi_id = LongField(unique=True)
# @todo: last_seen
# @todo: expired
_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
_bi_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
def __str__(self):
return self.name
("alarm_class", "rca_neighbors"),
],
}
status = "A"
timestamp = DateTimeField(required=True)
last_update = DateTimeField(required=True)
managed_object = ForeignKeyField(ManagedObject)
alarm_class = PlainReferenceField(AlarmClass)
severity = IntField(required=True)
vars = DictField()
# Calculated alarm discriminator
# Has meaning only for alarms with is_unique flag set
# Calculated as sha1("value1\x00....\x00valueN").hexdigest()
discriminator = StringField(required=False)
log = ListField(EmbeddedDocumentField(AlarmLog))
# Manual acknowledgement timestamp
ack_ts = DateTimeField(required=False)
# Manual acknowledgement user name
ack_user = StringField(required=False)
#
opening_event = ObjectIdField(required=False)
closing_event = ObjectIdField(required=False)
# List of subscribers
subscribers = ListField(ForeignKeyField(User))
#
custom_subject = StringField(required=False)
custom_style = ForeignKeyField(Style, required=False)
#
reopens = IntField(required=False)
# RCA
# Reference to root cause (Active Alarm or Archived Alarm instance)
# Flap detection
flap_condition = StringField(
required=False, choices=[("none", "none"), ("count", "count")], default="none"
)
flap_window = IntField(required=False, default=0)
flap_threshold = FloatField(required=False, default=0)
# RCA
root_cause = ListField(EmbeddedDocumentField(AlarmRootCauseCondition))
topology_rca = BooleanField(default=False)
# List of handlers to be called on alarm raising
handlers = ListField(StringField())
# List of handlers to be called on alarm clear
clear_handlers = ListField(StringField())
# Plugin settings
plugins = ListField(EmbeddedDocumentField(AlarmPlugin))
# Time in seconds to delay alarm risen notification
notification_delay = IntField(required=False)
# Control time to reopen alarm instead of creating new
control_time0 = IntField(required=False)
# Control time to reopen alarm after 1 reopen
control_time1 = IntField(required=False)
# Control time to reopen alarm after >1 reopen
control_timeN = IntField(required=False)
# Consequence recover time
# Root cause will be detached if consequence alarm
# will not clear itself in *recover_time*
recover_time = IntField(required=False, default=300)
#
bi_id = LongField(unique=True)
#
category = ObjectIdField()
class L3Link(Document):
"""
Network L3 links.
Always contains a list of subinterface references
"""
meta = {
"collection": "noc.links",
"strict": False,
"auto_create_index": False,
"indexes": ["subinterfaces", "linked_objects"],
}
subinterfaces = PlainReferenceListField("inv.SubInterface")
# List of linked objects
linked_objects = ListField(IntField())
# Name of discovery method or "manual"
discovery_method = StringField()
# Timestamp of first discovery
first_discovered = DateTimeField(default=datetime.datetime.now)
# Timestamp of last confirmation
last_seen = DateTimeField()
# L3 path cost
l3_cost = IntField(default=1)
def __str__(self):
return "(%s)" % ", ".join([smart_text(i) for i in self.subinterfaces])
def clean(self):
self.linked_objects = sorted(set(i.managed_object.id for i in self.subinterfaces))
super(L3Link, self).clean()