前后端全面升级

This commit is contained in:
pycook
2023-07-10 17:42:15 +08:00
parent c444fed436
commit db5ff60aff
629 changed files with 97789 additions and 23995 deletions

View File

@@ -1 +1 @@
# -*- coding: utf-8 -*-
"""Tests for the app."""

View File

@@ -1,55 +1,16 @@
# -*- coding: utf-8 -*-
"""Defines fixtures available to all tests."""
import jwt
import pytest
from webtest import TestApp
from flask import Response, json
from flask.testing import FlaskClient
from werkzeug.datastructures import Headers
from api.app import create_app
from api.extensions import db, cache
from api.models.acl import User
class CMDBTestClient(FlaskClient):
TEST_APP_SECRET = "test"
def open(self, *args, **kwargs):
headers = kwargs.pop("headers", Headers())
headers.setdefault("User-Agent", "py.test")
kwargs["headers"] = headers
json_data = kwargs.pop("json", None)
if json_data is not None:
kwargs["data"] = json.dumps(json_data)
if not kwargs.get("content_type"):
kwargs["content_type"] = "application/json"
auth = kwargs.pop("auth", (
"Access-Token",
jwt.encode({"sub": "test@xx.com"}, key=self.TEST_APP_SECRET)
))
kwargs["headers"][auth[0]] = auth[1]
return super(CMDBTestClient, self).open(*args, **kwargs)
class CMDBTestResponse(Response):
@property
def json(self):
return json.loads(self.data)
@pytest.fixture(scope="session")
@pytest.fixture
def app():
"""Create application for the tests."""
_app = create_app("tests.settings")
_app.config['SECRET_KEY'] = CMDBTestClient.TEST_APP_SECRET
_app.test_client_class = CMDBTestClient
_app.response_class = CMDBTestResponse
ctx = _app.test_request_context()
ctx.push()
yield _app
@@ -57,66 +18,6 @@ def app():
ctx.pop()
@pytest.fixture(scope="session")
def client(app):
with app.test_client(use_cookies=False) as c:
yield c
@pytest.fixture(scope="session")
def database(app):
"""Clean database after each case finished"""
setup_db()
yield db
teardown_db()
@pytest.fixture(scope="function")
def session(database, app):
with app.app_context():
clean_db()
clean_cache()
yield database.session
database.session.rollback()
def setup_db():
teardown_db()
db.create_all()
# create test user
def teardown_db():
db.session.remove()
db.drop_all()
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):
if table.fullname in ["users"]:
continue
db.session.execute(table.delete())
db.session.commit()
if not User.get_by(email="test@xx.com"):
u = User.create(
flush=True,
username="test",
nickname="测试",
email="test@xx.com",
key="",
secret=""
)
u._set_password("123456")
u.save()
@pytest.fixture
def testapp(app):
"""Create Webtest app."""

View File

@@ -1,59 +1 @@
# -*- coding: utf-8 -*-
from api.models.cmdb import Attribute
from tests.sample import init_attributes
def test_create_attribute(session, client):
url = "/api/v0.1/attributes"
payload = {
"name": "region",
"alias": "区域",
"value_type": "2"
}
resp = client.post(url, json=payload)
# check resp status code and content
assert resp.status_code == 200
assert resp.json["attr_id"]
# check there is a attribute in database
attr_id = resp.json["attr_id"]
attr_ins = Attribute.get_by_id(attr_id)
assert attr_ins.id == attr_id
assert attr_ins.name == "region"
assert attr_ins.alias == "区域"
def test_update_attribute(session, client):
attr_ins = init_attributes(1)[0]
url = "/api/v0.1/attributes/" + str(attr_ins.id)
payload = {
"name": "update",
}
resp = client.put(url, json=payload)
# check resp status code and content
assert resp.status_code == 200
assert resp.json["attr_id"] == attr_ins.id
# check attribute updated in database
attr_ins = Attribute.get_by_id(attr_ins.id)
assert attr_ins.name == "update"
def test_delete_attribute(session, client):
attr_ins = init_attributes(1)[0]
url = "/api/v0.1/attributes/" + str(attr_ins.id)
resp = client.delete(url)
assert resp.status_code == 200
# attr should be soft delete
attr_ins = Attribute.get_by_id(attr_ins.id)
assert attr_ins.deleted is True
assert attr_ins.deleted_at

View File

@@ -1,89 +1,10 @@
# -*- coding: utf-8 -*-
from .sample import init_ci_types, fake_attr_value, init_ci
from api.lib.cmdb.ci_type import CITypeAttributeManager, CITypeManager
from api.lib.cmdb.ci import CIManager
from api.models.cmdb import CI
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']]
def test_update_ci(session, client):
ci = init_ci(1)[0]
ci_id = ci.get("ci_id")
ci_type = CITypeManager.get_ci_types(ci.get("ci_type"))[0]
attrs = CITypeAttributeManager.get_attributes_by_type_id(ci_type["id"])
url = "/api/v0.1/ci/{}".format(ci_id)
fake_value = fake_attr_value(attrs[0])
payload = {**fake_value}
resp = client.put(url, json=payload)
assert resp.status_code == 200
assert resp.json["ci_id"] == ci_id
ci = CIManager().get_ci_by_id_from_db(ci_id)
assert ci[attrs[0]['name']] == fake_value[attrs[0]['name']]
def test_delete_ci(session, client):
ci = init_ci(1)[0]
ci_id = ci.get("ci_id")
url = "/api/v0.1/ci/{}".format(ci_id)
resp = client.delete(url)
assert resp.status_code == 200
ci_from_db = CI.query.filter_by(id=ci_id).first()
assert ci_from_db is None
def test_get_ci_by_types(session, client):
ci = init_ci(1)[0]
ci_type = CITypeManager.get_ci_types(ci.get("ci_type"))[0]
url = "/api/v0.1/ci/type/{}".format(ci_type["id"])
resp = client.get(url)
assert resp.status_code == 200
assert resp.json['cis'][0]['ci_id'] == ci['ci_id']
def test_get_ci_by_id(session, client):
ci = init_ci(1)[0]
url = "/api/v0.1/ci/{}".format(ci["ci_id"])
resp = client.get(url)
assert resp.status_code == 200
assert resp.json['ci_id'] == ci['ci_id']
def test_get_ci_detail_by_id(session, client):
ci = init_ci(1)[0]
url = "/api/v0.1/ci/{}/detail".format(ci["ci_id"])
resp = client.get(url)
assert resp.status_code == 200
assert resp.json['id'] == ci['ci_id']
class TestCI:
def test_ci_search_only_type_query(self, app):
with app.test_client() as c:
rv = c.get('/api/v0.1/ci/s?q=_type:server', json={})
json_data = rv.get_json()
assert type(json_data.get("result")) is list

View File

@@ -1,39 +1 @@
# -*- coding: utf-8 -*-
from .sample import init_ci_with_type, init_ci_type_relation, init_ci_relation
from api.lib.cmdb.ci import CIRelationManager
from api.models.cmdb import CIRelation, CIType, CI
def test_create_ci_relation(session, client):
init_ci_type_relation(1)
ci_types = CIType.query.all()
cis = init_ci_with_type(ci_types)
url = "/api/v0.1/ci_relations/{}/{}".format(cis[0]['ci_id'], cis[1]['ci_id'])
resp = client.post(url)
assert resp.status_code == 200
cr_id = resp.json['cr_id']
cr = CIRelation.get_by_id(cr_id)
assert cr is not None
def test_delte_ci_relation_by_ci_id(session, client):
cr_id = init_ci_relation()
cis = CI.query.all()
url = "/api/v0.1/ci_relations/{}/{}".format(cis[0].id, cis[1].id)
resp = client.delete(url)
assert resp.status_code == 200
cr = CIRelation.get_by_id(cr_id)
assert cr is None
def test_delete_ci_relation_by_id(session, client):
cr_id = init_ci_relation()
url = "/api/v0.1/ci_relations/{cr_id}".format(cr_id=cr_id)
resp = client.delete(url)
assert resp.status_code == 200
cr = CIRelation.get_by_id(cr_id)
assert cr is None

View File

@@ -1,167 +1 @@
# -*- coding: utf-8 -*-
from api.models.cmdb import (
CIType, CITypeAttribute,
Attribute, CITypeAttributeGroup,
CITypeAttributeGroupItem)
from tests.sample import (
init_attributes, init_ci_types,
init_attribute_groups)
def test_create_ci_type(session, client):
attr = init_attributes(1)[0]
url = "/api/v0.1/ci_types"
payload = {
"name": "test",
"alias": "测试",
"unique_key": attr.id
}
resp = client.post(url, json=payload)
# check resp status code and content
assert resp.status_code == 200
assert resp.json["type_id"]
# check there is a attribute in database
type_id = resp.json["type_id"]
ci_type_ins = CIType.get_by_id(type_id)
assert ci_type_ins.id == type_id
assert ci_type_ins.name == "test"
assert ci_type_ins.alias == "测试"
assert ci_type_ins.unique_id == attr.id
def test_update_ci_type(session, client):
ci_type_ins = init_ci_types(1)[0]
url = "/api/v0.1/ci_types/" + str(ci_type_ins.id)
payload = {
"name": "update",
}
resp = client.put(url, json=payload)
# check resp status code and content
assert resp.status_code == 200
assert resp.json["type_id"] == ci_type_ins.id
# check ci_type updated in database
ci_type_ins = CIType.get_by_id(ci_type_ins.id)
assert ci_type_ins.name == "update"
def test_delete_ci_type(session, client):
ci_type_ins = init_ci_types(1)[0]
url = "/api/v0.1/ci_types/" + str(ci_type_ins.id)
resp = client.delete(url)
assert resp.status_code == 200
# attr should be soft delete
ci_type_ins = CIType.get_by_id(ci_type_ins.id)
assert ci_type_ins.deleted is True
assert ci_type_ins.deleted_at
def test_bind_attributes_ci_type(session, client):
attrs = init_attributes(3)
ci_type = init_ci_types(1)[0]
url = "/api/v0.1/ci_types/{}/attributes".format(ci_type.id)
payload = {
"attr_id": [str(x.id) for x in attrs]
}
resp = client.post(url, json=payload)
# check resp status code and content
assert resp.status_code == 200
assert len(resp.json["attributes"]) == len(attrs)
# check ci_type has 4 attributes
ci_type_attribute_ids = [x.attr_id for x in CITypeAttribute.query.filter_by(type_id=ci_type.id).all()]
for attr in attrs:
assert attr.id in ci_type_attribute_ids
def test_get_attributes_ci_type(session, client):
ci_type = init_ci_types(1)[0]
url = "/api/v0.1/ci_types/{}/attributes".format(ci_type.name)
resp = client.get(url)
assert resp.status_code == 200
assert len(resp.json["attributes"]) == 1
def test_update_attributes_ci_type(session, client):
ci_type = init_ci_types(1)[0]
attr = Attribute.query.first()
url = "/api/v0.1/ci_types/{}/attributes".format(ci_type.id)
payload = {
"attributes": [
{"attr_id": attr.id, "default_show": False, "is_required": True}
]
}
resp = client.put(url, json=payload)
assert resp.status_code == 200
ci_type_attr_ins = CITypeAttribute.query.filter_by(type_id=ci_type.id).first()
assert ci_type_attr_ins
assert ci_type_attr_ins.is_required is True
assert ci_type_attr_ins.default_show is False
def test_create_attribute_group_ci_type(session, client):
ci_type = init_ci_types(1)[0]
url = "/api/v0.1/ci_types/{}/attribute_groups".format(ci_type.id)
payload = {
"name": "A",
"order": 100,
}
resp = client.post(url, json=payload)
# check resp status code and content
assert resp.status_code == 200
assert resp.json["group_id"]
ins = CITypeAttributeGroup.query.filter_by(type_id=ci_type.id).first()
assert ins
assert ins.id == resp.json["group_id"]
assert ins.name == "A"
assert ins.order == 100
def test_update_attribute_group_ci_type(session, client):
attribute_groups = init_attribute_groups(1)[0]
url = "/api/v0.1/ci_types/attribute_groups/{}".format(attribute_groups.id)
payload = {
"attributes": [x.id for x in Attribute.query.all()]
}
resp = client.put(url, json=payload)
assert resp.status_code == 200
assert resp.json["group_id"]
ag_items = CITypeAttributeGroupItem.query.filter_by(group_id=attribute_groups.id).all()
for a in Attribute.query.all():
assert a.id in [x.attr_id for x in ag_items]
def test_delete_attribute_group_ci_type(session, client):
attribute_groups = init_attribute_groups(1)[0]
url = "/api/v0.1/ci_types/attribute_groups/{}".format(attribute_groups.id)
resp = client.delete(url)
assert resp.status_code == 200
attribute_group_ins = CITypeAttributeGroup.query.filter_by(id=attribute_groups.id).first()
assert attribute_group_ins.deleted is True
assert attribute_group_ins.deleted_at

View File

@@ -1,86 +1 @@
# -*- 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

View File

@@ -1,80 +1 @@
# -*- 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