Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.assertEqual(cls._meta[k], v)
self.assertFalse('collection' in Animal._meta)
self.assertFalse('collection' in Mammal._meta)
self.assertEqual(Animal._get_collection_name(), None)
self.assertEqual(Mammal._get_collection_name(), None)
self.assertEqual(Fish._get_collection_name(), 'fish')
self.assertEqual(Guppy._get_collection_name(), 'fish')
self.assertEqual(Human._get_collection_name(), 'human')
# ensure that a subclass of a non-abstract class can't be abstract
with self.assertRaises(ValueError):
class EvilHuman(Human):
evil = BooleanField(default=True)
meta = {'abstract': True}
}
workflow = PlainReferenceField(Workflow)
from_state = PlainReferenceField(State)
to_state = PlainReferenceField(State)
is_active = BooleanField(default=True)
# Event name
# Some predefined names exists:
# seen -- discovery confirms resource usage
# expired - TTL expired
event = StringField()
# Text label
label = StringField()
# Arbbitrary description
description = StringField()
# Enable manual transition
enable_manual = BooleanField(default=True)
# Handler to be called on starting transitions
# Any exception aborts transtion
handlers = ListField(StringField())
# Visual vertices
vertices = ListField(EmbeddedDocumentField(TransitionVertex))
# Integration with external NRI and TT systems
# Reference to remote system object has been imported from
remote_system = ReferenceField(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)
_bi_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
class RemoteSystem(Document):
meta = {"collection": "noc.remotesystem", "strict": False, "auto_create_index": False}
name = StringField(unique=True)
description = StringField()
handler = StringField()
# Environment variables
environment = ListField(EmbeddedDocumentField(EnvItem))
# Enable extractors/loaders
enable_admdiv = BooleanField()
enable_administrativedomain = BooleanField()
enable_authprofile = BooleanField()
enable_container = BooleanField()
enable_link = BooleanField()
enable_managedobject = BooleanField()
enable_managedobjectprofile = BooleanField()
enable_networksegment = BooleanField()
enable_networksegmentprofile = BooleanField()
enable_service = BooleanField()
enable_serviceprofile = BooleanField()
enable_subscriber = BooleanField()
enable_resourcegroup = BooleanField()
enable_ttsystem = BooleanField()
# Usage statistics
last_extract = DateTimeField()
last_successful_extract = DateTimeField()
extract_error = StringField()
last_load = DateTimeField()
last_successful_load = DateTimeField()
load_error = StringField()
_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
that has access to the teamserver API and the proper permissions. It
might also be from the internal teamserver, a C2, or an agent.
"""
meta = {
'collection': COLLECTION_LOGS,
'indexes': [
{
'fields': ['application'],
}
]
}
timestamp = FloatField(required=True, null=False)
application = StringField(required=True, null=False, max_length=MAX_STR_LEN)
level = StringField(required=True, null=False, choices=LOG_LEVELS.keys())
message = StringField(required=True, null=False, max_length=MAX_BIGSTR_LEN)
archived = BooleanField(default=False)
@staticmethod
def list_logs(include_archived=False, application=None, since=0, levels=None):
"""
Return a list of logs.
Optionally include archived logs.
Optionally filter by application.
Optionally filter by a timestamp.
Optionally filter by log level.
"""
if levels is None:
levels = LOG_LEVELS.keys()
if application is not None:
if include_archived:
return Log.objects( # pylint: disable=no-member
# File name without extension
file_name = StringField(unique=True)
# Subscription is active
is_active = BooleanField(default=True)
# email subject
subject = StringField()
# Run report as user
run_as = ForeignKeyField(User)
# Send result to notification group
# If empty, only file will be written
notification_group = ForeignKeyField(NotificationGroup)
# Predefined report id
# :
report = StringField()
#
last_status = BooleanField(default=False)
last_run = DateTimeField()
PREFIX = "var/reports"
JCLS = "noc.main.models.reportsubscription.ReportJob"
class RequestStub(object):
def __init__(self, user):
self.user = user
@classmethod
def send_reports(cls):
"""
Calculate and send all reports for today
:return:
"""
subscriptions = list(ReportSubscription.objects.filter(is_active=True))
## TS id
seq_ts = IntSequence("pm.ts")
class PMTS(Document):
"""
Time Series
"""
meta = {
"collection": "noc.pm.ts",
"allow_inheritance": False
}
ts_id = fields.IntField(primary_key=True, default=seq_ts.next)
name = fields.StringField(unique=True, unique_with="ts_id")
is_active = fields.BooleanField(default=True)
storage = PlainReferenceField(PMStorage)
check = PlainReferenceField(PMCheck)
type = fields.StringField(
default="G",
choices=[
("G", "GAUGE"),
("C", "COUNTER"),
("D", "DERIVE")
]
)
def __unicode__(self):
return self.name
def register(self, timestamp, value):
if isinstance(timestamp, datetime.datetime):
username = fields.StringField(required=True)
email = fields.EmailField()
# name is a human-readable name used to refer to user e.g. "Martin Taylor"
# longest full name registered in guinness book is 744 letters-long
name = fields.StringField()
password = fields.StringField(
max_length=128,
verbose_name=_('password'),
help_text=_("Use '[algo]$[iterations]$[salt]$[hexdigest]' or use the <a href="\"password/\"">change password form</a>.")
)
# the following DjangoORM-like fields are defined
# just like in mongoengine.django.auth.User
is_staff = fields.BooleanField(default=False)
is_active = fields.BooleanField(default=True)
is_superuser = fields.BooleanField(default=False)
last_login = fields.DateTimeField(default=timezone.now, verbose_name=_('last login'))
date_joined = fields.DateTimeField(default=timezone.now, verbose_name=_('date joined'))
user_permissions = fields.ListField(
fields.ReferenceField(auth.Permission),
verbose_name=_('user permissions'),
help_text=_('Permissions for the user.')
)
USERNAME_FIELD = 'username'
REQUIRED_FIELDS = ['email']
def __str__(self):
return self.username
IntField,
BooleanField,
DateTimeField,
ListField,
ObjectIdField,
)
import mongoengine
# DynamicDocument allows for unspecified fields to be put in as well
class User(mongoengine.DynamicDocument):
"""User Document Schema"""
username = StringField(required=True)
email = StringField(required=True)
verified = BooleanField(required=True, default=False)
trusted = BooleanField(required=True, default=False)
anon = BooleanField(required=True, default=False)
karma = IntField(required=True, default=0)
posted_tips = ListField(ObjectIdField())
date_created = DateTimeField(required=True)
pro_pic = StringField(
required=True,
default="https://pngimage.net/wp-content/uploads/2018/05/default-profile-image-png-5.png",
)
auth_server_uid = StringField(required=True)