"Fossies" - the Fresh Open Source Software Archive

Member "senlin-8.0.0/senlin/tests/unit/db/shared.py" (16 Oct 2019, 5184 Bytes) of package /linux/misc/openstack/senlin-8.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "shared.py": 4.0.0_vs_6.0.0.

    1 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 # not use this file except in compliance with the License. You may obtain
    3 # a copy of the License at
    4 #
    5 #      http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 # Unless required by applicable law or agreed to in writing, software
    8 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 # License for the specific language governing permissions and limitations
   11 # under the License.
   12 
   13 
   14 from oslo_serialization import jsonutils
   15 from oslo_utils import timeutils as tu
   16 from oslo_utils import uuidutils
   17 
   18 from senlin.db.sqlalchemy import api as db_api
   19 from senlin.engine import parser
   20 
   21 
   22 sample_profile = """
   23   name: test_profile_name
   24   type: my_test_profile_type
   25   spec:
   26     template:
   27       heat_template_version: "2013-05-23"
   28       resources:
   29         myrandom: OS::Heat::RandomString
   30       files:
   31         myfile: contents
   32 """
   33 
   34 sample_action = """
   35   name: test_cluster_create_action
   36   target: cluster_001
   37   action: create
   38   cause: User Initiate
   39   timeout: 60
   40   control: READY
   41   status: INIT
   42   status_reason: Just Initialized
   43   inputs:
   44     min_size: 1
   45     max_size: 10
   46     pause_time: PT10M
   47 """
   48 
   49 
   50 UUIDs = (UUID1, UUID2, UUID3) = sorted([uuidutils.generate_uuid()
   51                                         for x in range(3)])
   52 
   53 
   54 def create_profile(context, profile=sample_profile, **kwargs):
   55     data = parser.simple_parse(profile)
   56     data['user'] = context.user_id
   57     data['project'] = context.project_id
   58     data['domain'] = context.domain_id
   59     data.update(kwargs)
   60     return db_api.profile_create(context, data)
   61 
   62 
   63 def create_cluster(ctx, profile, **kwargs):
   64     values = {
   65         'name': 'db_test_cluster_name',
   66         'profile_id': profile.id,
   67         'user': ctx.user_id,
   68         'project': ctx.project_id,
   69         'domain': 'unknown',
   70         'parent': None,
   71         'next_index': 1,
   72         'timeout': 60,
   73         'desired_capacity': 0,
   74         'init_at': tu.utcnow(True),
   75         'status': 'INIT',
   76         'status_reason': 'Just Initialized',
   77         'meta_data': {},
   78         'dependents': {},
   79         'config': {},
   80     }
   81     values.update(kwargs)
   82     if 'project' in kwargs:
   83         values.update({'project': kwargs.get('project')})
   84     return db_api.cluster_create(ctx, values)
   85 
   86 
   87 def create_node(ctx, cluster, profile, **kwargs):
   88     if cluster:
   89         cluster_id = cluster.id
   90         index = db_api.cluster_next_index(ctx, cluster_id)
   91     else:
   92         cluster_id = ''
   93         index = -1
   94 
   95     values = {
   96         'name': 'test_node_name',
   97         'physical_id': UUID1,
   98         'cluster_id': cluster_id,
   99         'profile_id': profile.id,
  100         'project': ctx.project_id,
  101         'index': index,
  102         'role': None,
  103         'created_at': None,
  104         'updated_at': None,
  105         'status': 'ACTIVE',
  106         'status_reason': 'create complete',
  107         'meta_data': jsonutils.loads('{"foo": "123"}'),
  108         'data': jsonutils.loads('{"key1": "value1"}'),
  109         'dependents': {},
  110     }
  111     values.update(kwargs)
  112     return db_api.node_create(ctx, values)
  113 
  114 
  115 def create_webhook(ctx, obj_id, obj_type, action, **kwargs):
  116     values = {
  117         'name': 'test_webhook_name',
  118         'user': ctx.user_id,
  119         'project': ctx.project_id,
  120         'domain': ctx.domain_id,
  121         'created_at': None,
  122         'obj_id': obj_id,
  123         'obj_type': obj_type,
  124         'action': action,
  125         'credential': None,
  126         'params': None,
  127     }
  128     values.update(kwargs)
  129     return db_api.webhook_create(ctx, values)
  130 
  131 
  132 def create_action(ctx, **kwargs):
  133     values = {
  134         'context': kwargs.get('context'),
  135         'description': 'Action description',
  136         'target': kwargs.get('target'),
  137         'action': kwargs.get('action'),
  138         'cause': 'Reason for action',
  139         'owner': kwargs.get('owner'),
  140         'interval': -1,
  141         'inputs': {'key': 'value'},
  142         'outputs': {'result': 'value'},
  143         'depends_on': [],
  144         'depended_by': []
  145     }
  146     values.update(kwargs)
  147     return db_api.action_create(ctx, values)
  148 
  149 
  150 def create_policy(ctx, **kwargs):
  151     values = {
  152         'name': 'test_policy',
  153         'type': 'senlin.policy.scaling',
  154         'user': ctx.user_id,
  155         'project': ctx.project_id,
  156         'domain': ctx.domain_id,
  157         'spec': {
  158             'type': 'senlin.policy.scaling',
  159             'version': '1.0',
  160             'properties': {
  161                 'adjustment_type': 'WHATEVER',
  162                 'count': 1,
  163             }
  164         },
  165         'data': None,
  166     }
  167 
  168     values.update(kwargs)
  169     return db_api.policy_create(ctx, values)
  170 
  171 
  172 def create_event(ctx, **kwargs):
  173     values = {
  174         'timestamp': tu.utcnow(True),
  175         'obj_id': 'FAKE_ID',
  176         'obj_name': 'FAKE_NAME',
  177         'obj_type': 'CLUSTER',
  178         'cluster_id': 'FAKE_CLUSTER',
  179         'level': '20',
  180         'user': ctx.user_id,
  181         'project': ctx.project_id,
  182         'action': 'DANCE',
  183         'status': 'READY',
  184         'status_reason': 'Just created.',
  185         'meta_data': {
  186             'air': 'polluted'
  187         }
  188     }
  189 
  190     values.update(kwargs)
  191     return db_api.event_create(ctx, values)