[Jenkins]Jenkins

Code Coverage

Cobertura Coverage Report > osm_policy_module.core >

database.py

Trend

Classes100%
 
Lines85%
   
Conditionals100%
 

File Coverage summary

NameClassesLinesConditionals
database.py
100%
1/1
85%
116/137
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
database.py
85%
116/137
N/A

Source

osm_policy_module/core/database.py

1

# -*- coding: utf-8 -*-




2






3

# Copyright 2018 Whitestack, LLC




4

# *************************************************************




5






6

# This file is part of OSM Monitoring module




7

# All Rights Reserved to Whitestack, LLC




8






9

# Licensed under the Apache License, Version 2.0 (the "License"); you may




10

# not use this file except in compliance with the License. You may obtain




11

# a copy of the License at




12






13

#         http://www.apache.org/licenses/LICENSE-2.0




14






15

# Unless required by applicable law or agreed to in writing, software




16

# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT




17

# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the




18

# License for the specific language governing permissions and limitations




19

# under the License.




20






21

# For those usages not covered by the Apache License, Version 2.0 please




22

# contact: bdiaz@whitestack.com or glavado@whitestack.com




23

##




24
1
import datetime




25
1
import logging




26
1
import os




27
1
from typing import Iterable, List




28






29
1
from peewee import CharField, IntegerField, ForeignKeyField, Model, TextField, AutoField, DateTimeField, Proxy, \




30

    BooleanField




31
1
from peewee_migrate import Router




32
1
from playhouse.db_url import connect




33






34
1
from osm_policy_module import migrations




35
1
from osm_policy_module.core.config import Config




36






37
1
log = logging.getLogger(__name__)




38






39
1
db = Proxy()




40






41






42
1
class BaseModel(Model):




43
1
    id = AutoField(primary_key=True)




44






45
1
    class Meta:




46
1
        database = db




47






48






49
1
class ScalingGroup(BaseModel):




50
1
    nsr_id = CharField()




51
1
    vnf_member_index = CharField()




52
1
    name = CharField()




53
1
    content = TextField()




54






55






56
1
class ScalingPolicy(BaseModel):




57
1
    name = CharField()




58
1
    cooldown_time = IntegerField()




59
1
    scale_in_operation = CharField(default='AND')




60
1
    scale_out_operation = CharField(default='OR')




61
1
    enabled = BooleanField(default=True)




62
1
    last_scale = DateTimeField(default=datetime.datetime.now)




63
1
    scaling_group = ForeignKeyField(ScalingGroup, related_name='scaling_policies', on_delete='CASCADE')




64






65






66
1
class ScalingCriteria(BaseModel):




67
1
    name = CharField()




68
1
    scaling_policy = ForeignKeyField(ScalingPolicy, related_name='scaling_criterias', on_delete='CASCADE')




69






70






71
1
class ScalingAlarm(BaseModel):




72
1
    alarm_uuid = CharField(unique=True)




73
1
    action = CharField()




74
1
    vnf_member_index = CharField()




75
1
    vdu_name = CharField()




76
1
    scaling_criteria = ForeignKeyField(ScalingCriteria, related_name='scaling_alarms', on_delete='CASCADE')




77
1
    last_status = CharField(default='insufficient-data')




78






79






80
1
class VnfAlarm(BaseModel):




81
1
    alarm_id = CharField()




82
1
    alarm_uuid = CharField(unique=True)




83
1
    nsr_id = CharField()




84
1
    vnf_member_index = CharField()




85
1
    vdu_name = CharField()




86






87






88
1
class AlarmAction(BaseModel):




89
1
    type = CharField()




90
1
    url = TextField()




91
1
    alarm = ForeignKeyField(VnfAlarm, related_name='actions', on_delete='CASCADE')




92






93






94
1
class DatabaseManager:




95
1
    def __init__(self, config: Config):




96
0
        db.initialize(connect(config.get('sql', 'database_uri')))




97






98
1
    def create_tables(self) -> None:




99
0
        db.connect()




100
0
        with db.atomic():




101
0
            router = Router(db, os.path.dirname(migrations.__file__))




102
0
            router.run()




103
0
        db.close()




104






105






106
1
class ScalingAlarmRepository:




107






108
1
    @staticmethod




109
1
    def list(*expressions) -> Iterable[ScalingAlarm]:




110
0
        return ScalingAlarm.select().where(*expressions)




111






112
1
    @staticmethod




113
1
    def get(*expressions, join_classes: List = None) -> ScalingAlarm:




114
1
        query = ScalingAlarm.select()




115
1
        if join_classes:




116
1
            for join_class in join_classes:




117
1
                query = query.join(join_class)




118
1
        return query.where(*expressions).get()




119






120
1
    @staticmethod




121
1
    def create(**query) -> ScalingAlarm:




122
1
        return ScalingAlarm.create(**query)




123






124






125
1
class ScalingGroupRepository:




126






127
1
    @staticmethod




128
1
    def list(*expressions) -> Iterable[ScalingGroup]:




129
0
        return ScalingGroup.select().where(*expressions)




130






131
1
    @staticmethod




132
1
    def get(*expressions) -> ScalingGroup:




133
1
        return ScalingGroup.select().where(*expressions).get()




134






135
1
    @staticmethod




136
1
    def create(**query) -> ScalingGroup:




137
1
        return ScalingGroup.create(**query)




138






139






140
1
class ScalingPolicyRepository:




141






142
1
    @staticmethod




143
1
    def list(*expressions, join_classes: List = None) -> Iterable[ScalingPolicy]:




144
0
        query = ScalingPolicy.select()




145
0
        if join_classes:




146
0
            for join_class in join_classes:




147
0
                query = query.join(join_class)




148
0
        return query.where(*expressions)




149






150
1
    @staticmethod




151
1
    def get(*expressions, join_classes: List = None) -> ScalingPolicy:




152
1
        query = ScalingPolicy.select()




153
1
        if join_classes:




154
1
            for join_class in join_classes:




155
1
                query = query.join(join_class)




156
1
        return query.where(*expressions).get()




157






158
1
    @staticmethod




159
1
    def create(**query) -> ScalingPolicy:




160
1
        return ScalingPolicy.create(**query)




161






162






163
1
class ScalingCriteriaRepository:




164






165
1
    @staticmethod




166
1
    def list(*expressions, join_classes: List = None) -> Iterable[ScalingCriteria]:




167
0
        query = ScalingCriteria.select()




168
0
        if join_classes:




169
0
            for join_class in join_classes:




170
0
                query = query.join(join_class)




171
0
        return query.where(*expressions)




172






173
1
    @staticmethod




174
1
    def get(*expressions, join_classes: List = None) -> ScalingCriteria:




175
1
        query = ScalingCriteria.select()




176
1
        if join_classes:




177
1
            for join_class in join_classes:




178
1
                query = query.join(join_class)




179
1
        return query.where(*expressions).get()




180






181
1
    @staticmethod




182
1
    def create(**query) -> ScalingCriteria:




183
1
        return ScalingCriteria.create(**query)




184






185






186
1
class VnfAlarmRepository:




187






188
1
    @staticmethod




189
1
    def list(*expressions) -> Iterable[VnfAlarm]:




190
0
        return VnfAlarm.select().where(*expressions)




191






192
1
    @staticmethod




193
1
    def get(*expressions) -> VnfAlarm:




194
1
        return VnfAlarm.select().where(*expressions).get()




195






196
1
    @staticmethod




197
1
    def create(**query) -> VnfAlarm:




198
1
        return VnfAlarm.create(**query)




199






200






201
1
class AlarmActionRepository:




202






203
1
    @staticmethod




204
1
    def list(*expressions) -> Iterable[AlarmAction]:




205
0
        return AlarmAction.select().where(*expressions)




206






207
1
    @staticmethod




208
1
    def get(*expressions) -> AlarmAction:




209
0
        return AlarmAction.select().where(*expressions).get()




210






211
1
    @staticmethod




212
1
    def create(**query) -> AlarmAction:




213
1
        return AlarmAction.create(**query)