Merge pull request #34 from OhBonsai/develop

add test cases

cmdb-api/api/lib/perm/auth.py

@@ -53,6 +53,7 @@ def _auth_with_token():
             return False
 
         login_user(user)
+        g.user = user
         return True
     except jwt.ExpiredSignatureError:
         return False

cmdb-api/api/models/cmdb.py

@@ -266,8 +266,8 @@ class OperationRecord(Model):
     __tablename__ = "c_records"
 
     uid = db.Column(db.Integer, index=True, nullable=False)
-    origin = db.Column(db.String(32), nullable=False)
+    origin = db.Column(db.String(32), nullable=True)
-    ticket_id = db.Column(db.String(32), nullable=False)
+    ticket_id = db.Column(db.String(32), nullable=True)
     reason = db.Column(db.Text)
 
 

cmdb-api/tests/conftest.py

@@ -9,7 +9,7 @@ from flask.testing import FlaskClient
 from werkzeug.datastructures import Headers
 
 from api.app import create_app
-from api.extensions import db
+from api.extensions import db, cache
 from api.models.acl import User
 
 
@@ -75,6 +75,7 @@ def database(app):
 def session(database, app):
     with app.app_context():
         clean_db()
+        clean_cache()
         yield database.session
         database.session.rollback()
 
@@ -91,6 +92,10 @@ def teardown_db():
     db.session.bind.dispose()
 
 
+def clean_cache():
+    cache.clear()
+
+
 def clean_db():
     """clean all data but not drop table"""
     for table in reversed(db.metadata.sorted_tables):

cmdb-api/tests/sample.py

@@ -3,7 +3,14 @@
 import uuid
 import random
 
-from api.models.cmdb import Attribute, CIType, CITypeAttributeGroup, CITypeAttribute
+from api.models.cmdb import (
+    Attribute,
+    CIType,
+    CITypeAttributeGroup,
+    CITypeAttribute,
+    CITypeRelation,
+    RelationType
+)
 
 
 def init_attributes(num=1):
@@ -12,7 +19,7 @@ def init_attributes(num=1):
         attrs.append(Attribute.create(
             name=uuid.uuid4().hex[:8],
             alias=uuid.uuid4().hex[:8],
-            value_type=str(random.randint(0, 100) % 7)
+            value_type=str(random.randint(0, 100) % 3)
         ))
     return attrs
 
@@ -47,3 +54,37 @@ def init_attribute_groups(num=1):
             order=i
         ))
     return ags
+
+
+def init_relation_type(num=1):
+    result = []
+    for i in range(num):
+        result.append(RelationType.create(
+            name=uuid.uuid4().hex[:8],
+        ))
+    return result
+
+
+def init_ci_type_relation(num=1):
+    result = []
+    ci_types = init_ci_types(num+1)
+    relation_types = init_relation_type(num)
+    for i in range(num):
+        result.append(CITypeRelation.create(
+            parent_id=ci_types[i].id,
+            child_id=ci_types[i+1].id,
+            relation_type_id=relation_types[i].id
+        ))
+    return result
+
+
+def fake_attr_value(attr_dict):
+    attr_type = attr_dict["value_type"]
+    attr_name = attr_dict["name"]
+
+    if attr_type == "0":
+        return {attr_name: random.randint(0, 1000)}
+    elif attr_type == "1":
+        return {attr_name: random.randint(0, 1000) / 3.0}
+    elif attr_type == "2":
+        return {attr_name: uuid.uuid4().hex[:8]}

cmdb-api/tests/test_cmdb_ci.py

@@ -1,3 +1,27 @@
 # -*- coding: utf-8 -*-
+from .sample import init_ci_types, fake_attr_value
+from api.lib.cmdb.ci_type import CITypeAttributeManager
+from api.lib.cmdb.ci import CIManager
+
+
+def test_create_ci(session, client):
+    ci_type = init_ci_types(1)[0]
+    attrs = CITypeAttributeManager.get_attributes_by_type_id(ci_type.id)
+    url = "/api/v0.1/ci"
+
+    fake_value = fake_attr_value(attrs[0])
+
+    payload = {
+        "ci_type": ci_type.id,
+        **fake_value
+    }
+
+    resp = client.post(url, json=payload)
+    assert resp.status_code == 200
+    assert resp.json["ci_id"]
+
+    ci_id = resp.json["ci_id"]
+    ci = CIManager().get_ci_by_id_from_db(ci_id)
+    assert ci[attrs[0]["name"]] == fake_value[attrs[0]['name']]
 
 

cmdb-api/tests/test_cmdb_ci_type_relation.py

@@ -1 +1,86 @@
 # -*- coding: utf-8 -*-
+from api.models.cmdb import (
+   CITypeRelation
+)
+
+from .sample import init_relation_type, init_ci_types, init_ci_type_relation
+
+
+def test_create_ci_type_relation(session, client):
+    ci_types = init_ci_types(2)
+    relation_type = init_relation_type(1)[0]
+
+    url = "/api/v0.1/ci_type_relations/{}/{}".format(*[x.id for x in ci_types])
+    payload = {
+        "relation_type_id": relation_type.id,
+    }
+
+    resp = client.post(url, json=payload)
+
+    assert resp.status_code == 200
+    assert resp.json["ctr_id"]
+
+    ci_type_relations_id = resp.json["ctr_id"]
+    ci_type_relation = CITypeRelation.get_by_id(ci_type_relations_id)
+    assert ci_type_relation.parent_id == ci_types[0].id
+    assert ci_type_relation.child_id == ci_types[1].id
+    assert ci_type_relation.relation_type_id == relation_type.id
+
+
+def test_delete_ci_type_relation_by_ci_type_id(session, client):
+    ci_type_relation_ins = init_ci_type_relation(1)[0]
+    url = "/api/v0.1/ci_type_relations/{}/{}".format(
+        ci_type_relation_ins.parent_id, ci_type_relation_ins.child_id)
+    resp = client.delete(url)
+
+    assert resp.status_code == 200
+    # fake deleted
+    ci_type_relation_ins = CITypeRelation.query.filter_by(id=ci_type_relation_ins.id).first()
+    assert ci_type_relation_ins is not None
+
+
+def test_delete_ci_type_relation_by_id(session, client):
+    ci_type_relation_ins = init_ci_type_relation(1)[0]
+    url = "/api/v0.1/ci_type_relations/" + str(ci_type_relation_ins.id)
+    resp = client.delete(url)
+
+    assert resp.status_code == 200
+    # fake deleted
+    ci_type_relation_ins = CITypeRelation.query.filter_by(id=ci_type_relation_ins.id).first()
+    assert ci_type_relation_ins is not None
+
+
+def test_get_ci_type_relations(session, client):
+    ci_type_relations = init_ci_type_relation(2)
+    url = "/api/v0.1/ci_type_relations"
+    resp = client.get(url)
+
+    assert resp.status_code == 200
+    assert len(resp.json) == 2
+    assert resp.json[0]["id"] == ci_type_relations[0].id
+    assert resp.json[1]["id"] == ci_type_relations[1].id
+
+
+def test_get_children(session, client):
+    ci_type_relation_ins = init_ci_type_relation(1)[0]
+    url = "/api/v0.1/ci_type_relations/{parent_id}/children".format(
+        parent_id=ci_type_relation_ins.parent_id)
+    resp = client.get(url)
+
+    assert resp.status_code == 200
+    assert len(resp.json["children"]) == 1
+    assert resp.json["children"][0]["id"] == ci_type_relation_ins.child_id
+
+
+def test_get_parents(session, client):
+    ci_type_relation_ins = init_ci_type_relation(1)[0]
+    url = "/api/v0.1/ci_type_relations/{child_id}/parents".format(
+        child_id=ci_type_relation_ins.child_id)
+    resp = client.get(url)
+
+    assert resp.status_code == 200
+    assert len(resp.json["parents"]) == 1
+    assert resp.json["parents"][0]["id"] == ci_type_relation_ins.parent_id
+
+
+

cmdb-api/tests/test_cmdb_relation_type.py

@@ -1 +1,80 @@
 # -*- coding: utf-8 -*-
+from api.models.cmdb import (
+   RelationType 
+)
+
+from .sample import init_relation_type
+
+
+def test_get_relation_type(session, client):
+    relation_type_instances = init_relation_type(2)
+    url = "/api/v0.1/relation_types"
+    resp = client.get(url)
+
+    assert resp.status_code == 200
+    assert len(resp.json) == 2
+    assert resp.json[0]["id"] == relation_type_instances[0].id
+    assert resp.json[1]["id"] == relation_type_instances[1].id
+
+
+def test_create_relation_type(session, client):
+
+    url = "/api/v0.1/relation_types"
+    payload = {
+        "name": "test",
+    }
+
+    resp = client.post(url, json=payload)
+
+    assert resp.status_code == 200
+    assert resp.json["id"]
+
+    relation_types_id = resp.json["id"]
+    relation_type = RelationType.get_by_id(relation_types_id)
+    assert relation_type.id == relation_types_id
+    assert relation_type.name == "test"
+
+
+def test_create_relation_type_name_strip(session, client):
+    url = "/api/v0.1/relation_types"
+    payload = {
+        "name": "test\t   ",
+    }
+
+    resp = client.post(url, json=payload)
+
+    assert resp.status_code == 200
+    assert resp.json["id"]
+
+    relation_types_id = resp.json["id"]
+    relation_type = RelationType.get_by_id(relation_types_id)
+    assert relation_type.name == "test"
+
+
+def test_update_relation_type(session, client):
+    relation_type_ins = init_relation_type(1)[0]
+
+    url = "/api/v0.1/relation_types/" + str(relation_type_ins.id)
+    payload = {
+        "name": "update",
+    }
+
+    resp = client.put(url, json=payload)
+
+    assert resp.status_code == 200
+    assert resp.json["id"] == relation_type_ins.id
+
+    relation_type_ins = RelationType.get_by_id(relation_type_ins.id)
+    assert relation_type_ins.name == "update"
+
+
+def test_delete_relation_type(session, client):
+    relation_type_ins = init_relation_type(1)[0]
+    url = "/api/v0.1/relation_types/" + str(relation_type_ins.id)
+    resp = client.delete(url)
+
+    assert resp.status_code == 200
+    relation_type_ins = RelationType.query.filter_by(id=relation_type_ins.id).first()
+    assert relation_type_ins.deleted is True
+    assert relation_type_ins.deleted_at
+