path: root/dcae-cli/dcae_cli/catalog/mock/tests/test_schema.py

# ============LICENSE_START=======================================================
# org.onap.dcae
# ================================================================================
# Copyright (c) 2017 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
#
# ECOMP is a trademark and service mark of AT&T Intellectual Property.

# -*- coding: utf-8 -*-
'''
Tests the mock catalog
'''
import pytest
import json, copy

from dcae_cli.catalog.mock.schema import validate_component, validate_format,apply_defaults_docker_config, apply_defaults
from dcae_cli.catalog.mock import schema
from dcae_cli.util.exc import DcaeException


format_test = r'''
{
  "self": {
    "name": "asimov.format.integerClassification",
    "version": "1.0.0",
    "description": "Represents a single classification from a machine learning model - just a test version"
  },
  "dataformatversion": "1.0.0",
  "jsonschema": {
      "$schema": "http://json-schema.org/draft-04/schema#",
      "type": "object",
      "properties": {
          "classification": {
              "type": "string"
          }
      },
      "additionalProperties": false
   }
}
'''


component_test = r'''
{
  "self": {
    "version": "1.0.0",
    "name": "asimov.component.kpi_anomaly",
    "description": "Classifies VNF KPI data as anomalous",
    "component_type": "docker"
  },
  "streams": {
    "subscribes": [
      {
        "format": "dcae.vnf.kpi",
        "version": "1.0.0",
        "route": "/data",
        "type": "http"
      },
      {
        "format":"std.format_one",
        "version":"1.0.0",
        "config_key":"sub2",
        "type": "message router"
      }
    ],
    "publishes": [
      {
        "format": "asimov.format.integerClassification",
        "version": "1.0.0",
        "config_key": "prediction",
        "type": "http"
      },
      {
        "format":"std.format_one",
        "version":"1.0.0",
        "config_key":"pub2",
        "type": "message router"
      }
    ]
  },
  "services": {
    "calls": [],
    "provides": [
      {
        "route": "/score-vnf",
        "request": {
          "format": "dcae.vnf.kpi",
          "version": "1.0.0"
        },
        "response": {
          "format": "asimov.format.integerClassification",
          "version": "1.0.0"
        }
      }
    ]
  },
  "parameters": [
    {
      "name": "threshold",
      "value": 0.75,
      "description": "Probability threshold to exceed to be anomalous"
    }
  ],
  "artifacts": [
    {
      "uri": "somedockercontainerpath",
      "type": "docker image"
    }
    ],
  "auxilary": {
    "healthcheck": {
      "type": "http",
      "endpoint": "/health"
    }
  }
}
'''

cdap_component_test = r'''
{  
   "self":{  
      "name":"std.cdap_comp",
      "version":"0.0.0",
      "description":"cdap test component",
      "component_type":"cdap"
   },
   "streams":{  
      "publishes":[  
         {  
            "format":"std.format_one",
            "version":"1.0.0",
            "config_key":"pub1",
            "type": "http"
         },
         {  
            "format":"std.format_one",
            "version":"1.0.0",
            "config_key":"pub2",
            "type": "message router"
         }
      ],
      "subscribes":[  
         {  
            "format":"std.format_two",
            "version":"1.5.0",
            "route":"/sub1",
            "type": "http"
         },
         {  
            "format":"std.format_one",
            "version":"1.0.0",
            "config_key":"sub2",
            "type": "message router"
         }
      ]
   },
   "services":{  
      "calls":[  

      ],
      "provides":[  
         {  
            "request":{  
               "format":"std.format_one",
               "version":"1.0.0"
            },
            "response":{  
               "format":"std.format_two",
               "version":"1.5.0"
            },
            "service_name":"baphomet",
            "service_endpoint":"rises",
            "verb":"GET"
         }
      ]
   },
   "parameters":[  

   ],
   "artifacts": [
     {
       "uri": "somecdapjarurl",
       "type": "jar"
     }
     ],
   "auxilary": {
     "streamname":"who",
     "artifact_name" : "HelloWorld",
     "artifact_version" : "3.4.3",
     "programs" : [
                    {"program_type" : "flows", "program_id" : "WhoFlow"},
                    {"program_type" : "services", "program_id" : "Greeting"}
                  ],
     "namespace" : "hw"
   }
}
'''


def test_basic():
    validate_component(json.loads(component_test))
    validate_format(json.loads(format_test))
    validate_component(json.loads(cdap_component_test))

    # Test with DR publishes for cdap
    dr_publishes = { "format":"std.format_one", "version":"1.0.0",
            "config_key":"pub3", "type": "data router" }
    cdap_valid = json.loads(cdap_component_test)
    cdap_valid["streams"]["publishes"].append(dr_publishes)

    # Test with DR subscribes for cdap
    cdap_invalid = json.loads(cdap_component_test)
    ss = cdap_invalid["streams"]["subscribes"][0]
    ss["type"] = "data_router"
    ss["config_key"] = "nada"
    cdap_invalid["streams"]["subscribes"][0] = ss

    with pytest.raises(DcaeException):
        validate_component(cdap_invalid)



def test_validate_docker_config():

    def compose_spec(config):
        spec = json.loads(component_test)
        spec["auxilary"] = config
        return spec

    good_docker_configs = [
            {
                "healthcheck": {
                    "type": "http",
                    "endpoint": "/health",
                    "interval": "15s",
                    "timeout": "1s"
                }
            },
            {
                "healthcheck": {
                    "type": "script",
                    "script": "curl something"
                    }
            }]

    for good_config in good_docker_configs:
        spec = compose_spec(good_config)
        assert validate_component(spec) == None

    bad_docker_configs = [
            #{},
            {
                "healthcheck": {}
            },
            {
                "healthcheck": {
                    "type": "http"
                }
            },
            {
                "healthcheck": {
                    "type": "http",
                    "script": "huh"
                }
            }]

    for bad_config in bad_docker_configs:
        with pytest.raises(DcaeException):
            spec = compose_spec(bad_config)
            validate_component(spec)


def test_validate_cdap_config():

    def compose_spec(config):
        spec = json.loads(cdap_component_test)
        spec["auxilary"] = config
        return spec

    good_cdap_configs = [
       {
           "streamname":"streamname",
           "artifact_version":"6.6.6",
           "artifact_name" : "testname",
           "programs" : [],
       },
       {
           "streamname":"streamname",
           "artifact_version":"6.6.6",
           "artifact_name" : "testname",
           "programs" : [{"program_type" : "flows", "program_id" : "flow_id"}],
           "program_preferences" : [{"program_type" : "flows", "program_id" : "flow_id", "program_pref" : {"he" : "shall rise"}}],
           "namespace" : "this should be an optional field",
           "app_preferences" : {"he" : "shall rise"}
       }
    ]

    for good_config in good_cdap_configs:
        spec = compose_spec(good_config)
        assert validate_component(spec) == None

    bad_cdap_configs = [
            {},
            {"YOU HAVE" : "ALWAYS FAILED ME"}
            ]

    for bad_config in bad_cdap_configs:
        with pytest.raises(DcaeException):
            spec = compose_spec(bad_config)
            validate_component(bad_config)


def test_apply_defaults():
    definition = { "length": { "default": 10 }, "duration": { "default": "10s" } }

    # Test: Add expected properties
    properties = {}
    actual = apply_defaults(definition, properties)
    assert actual == { "length": 10, "duration": "10s" }

    # Test: Don't mess with existing values
    properties = { "length": 100, "duration": "100s" }
    actual = apply_defaults(definition, properties)
    assert actual == properties

    # Test: No defaults to apply
    definition = { "length": {}, "duration": {} }
    properties = { "width": 100 }
    actual = apply_defaults(definition, properties)
    assert actual == properties

    # Test: Nested object
    definition = { "length": { "default": 10 }, "duration": { "default": "10s" },
            "location": { "properties": { "lat": { "default": "40" },
                "long": { "default": "75" }, "alt": {} } } }
    actual = apply_defaults(definition, {})
    assert actual == {'duration': '10s', 'length': 10,
            'location': {'lat': '40', 'long': '75'}}


def test_apply_defaults_docker_config():
    # Test: Adding of missing expected properties for http
    dc = { "healthcheck": { "type": "http", "endpoint": "/foo" } }
    actual = apply_defaults_docker_config(dc)

    assert "interval" in actual["healthcheck"]
    assert "timeout" in actual["healthcheck"]

    # Test: Adding of missing expected properties for script
    dc = { "healthcheck": { "type": "script", "script": "/bin/do-something" } }
    actual = apply_defaults_docker_config(dc)

    assert "interval" in actual["healthcheck"]
    assert "timeout" in actual["healthcheck"]

    # Test: Expected properties already exist
    dc = { "healthcheck": { "type": "http", "endpoint": "/foo",
        "interval": "10000s", "timeout": "100000s" } }
    actual = apply_defaults_docker_config(dc)
    assert dc == actual

    # Test: Never should happen
    dc = { "healthcheck": { "type": "bogus" } }
    actual = apply_defaults_docker_config(dc)
    assert dc == actual


def test_validate():
    fake_schema = {
            "$schema": "http://json-schema.org/draft-04/schema#",
            "title": "Test schema",
            "type": "object",
            "properties": {
                "foo": { "type": "string" },
                "bar": { "type": "integer" }
                },
            "required": ["foo", "bar"]
            }

    good_path = "/correct_path"

    def fetch_schema(path):
        if path == good_path:
            return fake_schema
        else:
            raise schema.FetchSchemaError("Schema not found")

    # Success case

    good_instance = { "foo": "hello", "bar": 1776 }

    schema._validate(fetch_schema, good_path, good_instance)

    # Error from validating

    bad_instance = {}

    with pytest.raises(DcaeException):
        schema._validate(fetch_schema, good_path, bad_instance)

    # Error from fetching

    bad_path = "/wrong_path"

    with pytest.raises(DcaeException):
        schema._validate(fetch_schema, bad_path, good_instance)