blob: b53624394ef4620ea06a1a29a914885a6afb6c3e [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you 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.
# isort:skip_file
"""Unit tests for Superset"""
import json
import pytest
import prison
from sqlalchemy.sql import func
import tests.test_app
from superset import db
from superset.models.annotations import Annotation, AnnotationLayer
from tests.base_tests import SupersetTestCase
from tests.annotation_layers.fixtures import (
create_annotation_layers,
get_end_dttm,
get_start_dttm,
)
ANNOTATION_LAYERS_COUNT = 10
ANNOTATIONS_COUNT = 5
class TestAnnotationLayerApi(SupersetTestCase):
@staticmethod
def get_layer_with_annotation() -> AnnotationLayer:
return (
db.session.query(AnnotationLayer)
.filter(AnnotationLayer.name == "layer_with_annotations")
.one_or_none()
)
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_annotation_layer(self):
"""
Annotation Api: Test get annotation layer
"""
annotation_layer = (
db.session.query(AnnotationLayer)
.filter(AnnotationLayer.name == "name1")
.first()
)
self.login(username="admin")
uri = f"api/v1/annotation_layer/{annotation_layer.id}"
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 200
expected_result = {
"id": annotation_layer.id,
"name": "name1",
"descr": "descr1",
}
data = json.loads(rv.data.decode("utf-8"))
assert data["result"] == expected_result
def test_info_annotation(self):
"""
Annotation API: Test info
"""
self.login(username="admin")
uri = "api/v1/annotation_layer/_info"
rv = self.get_assert_metric(uri, "info")
assert rv.status_code == 200
def test_info_security_query(self):
"""
Annotation API: Test info security
"""
self.login(username="admin")
params = {"keys": ["permissions"]}
uri = f"api/v1/annotation_layer/_info?q={prison.dumps(params)}"
rv = self.get_assert_metric(uri, "info")
data = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 200
assert "can_read" in data["permissions"]
assert "can_write" in data["permissions"]
assert len(data["permissions"]) == 2
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_annotation_layer_not_found(self):
"""
Annotation Api: Test get annotation layer not found
"""
max_id = db.session.query(func.max(AnnotationLayer.id)).scalar()
self.login(username="admin")
uri = f"api/v1/annotation_layer/{max_id + 1}"
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_list_annotation_layer(self):
"""
Annotation Api: Test get list annotation layers
"""
self.login(username="admin")
uri = "api/v1/annotation_layer/"
rv = self.get_assert_metric(uri, "get_list")
expected_fields = [
"name",
"descr",
"created_by",
"created_on",
"changed_by",
"changed_on_delta_humanized",
"changed_on",
]
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == ANNOTATION_LAYERS_COUNT
for expected_field in expected_fields:
assert expected_field in data["result"][0]
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_list_annotation_layer_sorting(self):
"""
Annotation Api: Test sorting on get list annotation layers
"""
self.login(username="admin")
uri = "api/v1/annotation_layer/"
order_columns = [
"name",
"descr",
"created_by.first_name",
"changed_by.first_name",
"changed_on",
"changed_on_delta_humanized",
"created_on",
]
for order_column in order_columns:
arguments = {"order_column": order_column, "order_direction": "asc"}
uri = f"api/v1/annotation_layer/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_list_annotation_layer_filter(self):
"""
Annotation Api: Test filters on get list annotation layers
"""
self.login(username="admin")
arguments = {
"columns": ["name", "descr"],
"filters": [
{"col": "name", "opr": "annotation_layer_all_text", "value": "2"}
],
}
uri = f"api/v1/annotation_layer/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
expected_result = {
"name": "name2",
"descr": "descr2",
}
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
assert data["result"][0] == expected_result
arguments = {
"columns": ["name", "descr"],
"filters": [
{"col": "name", "opr": "annotation_layer_all_text", "value": "descr3"}
],
}
uri = f"api/v1/annotation_layer/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
expected_result = {
"name": "name3",
"descr": "descr3",
}
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
assert data["result"][0] == expected_result
@pytest.mark.usefixtures("create_annotation_layers")
def test_create_annotation_layer(self):
"""
Annotation Api: Test create annotation layer
"""
self.login(username="admin")
annotation_layer_data = {"name": "new3", "descr": "description"}
uri = "api/v1/annotation_layer/"
rv = self.client.post(uri, json=annotation_layer_data)
assert rv.status_code == 201
data = json.loads(rv.data.decode("utf-8"))
created_model = db.session.query(AnnotationLayer).get(data.get("id"))
assert created_model is not None
assert created_model.name == annotation_layer_data["name"]
assert created_model.descr == annotation_layer_data["descr"]
# Rollback changes
db.session.delete(created_model)
db.session.commit()
@pytest.mark.usefixtures("create_annotation_layers")
def test_create_annotation_layer_uniqueness(self):
"""
Annotation Api: Test create annotation layer uniqueness
"""
self.login(username="admin")
annotation_layer_data = {"name": "name3", "descr": "description"}
uri = "api/v1/annotation_layer/"
rv = self.client.post(uri, json=annotation_layer_data)
assert rv.status_code == 422
data = json.loads(rv.data.decode("utf-8"))
assert data == {"message": {"name": ["Name must be unique"]}}
@pytest.mark.usefixtures("create_annotation_layers")
def test_update_annotation_layer(self):
"""
Annotation Api: Test update annotation layer
"""
annotation_layer = (
db.session.query(AnnotationLayer)
.filter(AnnotationLayer.name == "name2")
.one_or_none()
)
self.login(username="admin")
annotation_layer_data = {"name": "changed_name", "descr": "changed_description"}
uri = f"api/v1/annotation_layer/{annotation_layer.id}"
rv = self.client.put(uri, json=annotation_layer_data)
assert rv.status_code == 200
updated_model = db.session.query(AnnotationLayer).get(annotation_layer.id)
assert updated_model is not None
assert updated_model.name == annotation_layer_data["name"]
assert updated_model.descr == annotation_layer_data["descr"]
@pytest.mark.usefixtures("create_annotation_layers")
def test_update_annotation_layer_uniqueness(self):
"""
Annotation Api: Test update annotation layer uniqueness
"""
annotation_layer = (
db.session.query(AnnotationLayer)
.filter(AnnotationLayer.name == "name2")
.one_or_none()
)
self.login(username="admin")
annotation_layer_data = {"name": "name3", "descr": "changed_description"}
uri = f"api/v1/annotation_layer/{annotation_layer.id}"
rv = self.client.put(uri, json=annotation_layer_data)
data = json.loads(rv.data.decode("utf-8"))
assert rv.status_code == 422
assert data == {"message": {"name": ["Name must be unique"]}}
@pytest.mark.usefixtures("create_annotation_layers")
def test_update_annotation_layer_not_found(self):
"""
Annotation Api: Test update annotation layer not found
"""
max_id = db.session.query(func.max(AnnotationLayer.id)).scalar()
self.login(username="admin")
annotation_layer_data = {"name": "changed_name", "descr": "changed_description"}
uri = f"api/v1/annotation_layer/{max_id + 1}"
rv = self.client.put(uri, json=annotation_layer_data)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_delete_annotation_layer(self):
"""
Annotation Api: Test update annotation layer
"""
annotation_layer = (
db.session.query(AnnotationLayer)
.filter(AnnotationLayer.name == "name1")
.one_or_none()
)
self.login(username="admin")
uri = f"api/v1/annotation_layer/{annotation_layer.id}"
rv = self.client.delete(uri)
assert rv.status_code == 200
updated_model = db.session.query(AnnotationLayer).get(annotation_layer.id)
assert updated_model is None
@pytest.mark.usefixtures("create_annotation_layers")
def test_delete_annotation_layer_not_found(self):
"""
Annotation Api: Test delete annotation layer not found
"""
max_id = db.session.query(func.max(AnnotationLayer.id)).scalar()
self.login(username="admin")
uri = f"api/v1/annotation_layer/{max_id + 1}"
rv = self.client.delete(uri)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_delete_annotation_layer_integrity(self):
"""
Annotation Api: Test delete annotation layer integrity error
"""
query_child_layer = db.session.query(AnnotationLayer).filter(
AnnotationLayer.name == "layer_with_annotations"
)
child_layer = query_child_layer.one_or_none()
self.login(username="admin")
uri = f"api/v1/annotation_layer/{child_layer.id}"
rv = self.client.delete(uri)
assert rv.status_code == 422
@pytest.mark.usefixtures("create_annotation_layers")
def test_bulk_delete_annotation_layer(self):
"""
Annotation Api: Test bulk delete annotation layers
"""
query_no_child_layers = db.session.query(AnnotationLayer).filter(
AnnotationLayer.name.like("name%")
)
no_child_layers = query_no_child_layers.all()
no_child_layers_ids = [
annotation_layer.id for annotation_layer in no_child_layers
]
self.login(username="admin")
uri = f"api/v1/annotation_layer/?q={prison.dumps(no_child_layers_ids)}"
rv = self.client.delete(uri)
assert rv.status_code == 200
deleted_annotation_layers = query_no_child_layers.all()
assert deleted_annotation_layers == []
response = json.loads(rv.data.decode("utf-8"))
expected_response = {
"message": f"Deleted {len(no_child_layers_ids)} annotation layers"
}
assert response == expected_response
@pytest.mark.usefixtures("create_annotation_layers")
def test_bulk_delete_annotation_layer_not_found(self):
"""
Annotation Api: Test bulk delete annotation layers not found
"""
all_annotation_layers = (
db.session.query(AnnotationLayer)
.filter(AnnotationLayer.name.like("name%"))
.all()
)
all_annotation_layers_ids = [
annotation_layer.id for annotation_layer in all_annotation_layers
]
max_id = db.session.query(func.max(AnnotationLayer.id)).scalar()
all_annotation_layers_ids.append(max_id + 1)
self.login(username="admin")
uri = f"api/v1/annotation_layer/?q={prison.dumps(all_annotation_layers_ids)}"
rv = self.client.delete(uri)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_annotation(self):
"""
Annotation API: Test get annotation
"""
annotation_id = 1
annotation = (
db.session.query(Annotation)
.filter(Annotation.short_descr == f"short_descr{annotation_id}")
.one_or_none()
)
self.login(username="admin")
uri = (
f"api/v1/annotation_layer/{annotation.layer_id}/annotation/{annotation.id}"
)
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 200
expected_result = {
"id": annotation.id,
"end_dttm": get_end_dttm(annotation_id).isoformat(),
"json_metadata": "",
"layer": {"id": annotation.layer_id, "name": "layer_with_annotations"},
"long_descr": annotation.long_descr,
"short_descr": annotation.short_descr,
"start_dttm": get_start_dttm(annotation_id).isoformat(),
}
data = json.loads(rv.data.decode("utf-8"))
assert data["result"] == expected_result
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_annotation_not_found(self):
"""
Annotation API: Test get annotation not found
"""
layer = self.get_layer_with_annotation()
max_id = db.session.query(func.max(Annotation.id)).scalar()
self.login(username="admin")
uri = f"api/v1/annotation_layer/{layer.id}/annotation/{max_id + 1}"
rv = self.get_assert_metric(uri, "get")
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_list_annotation(self):
"""
Annotation Api: Test get list of annotations
"""
layer = self.get_layer_with_annotation()
self.login(username="admin")
uri = f"api/v1/annotation_layer/{layer.id}/annotation/"
rv = self.get_assert_metric(uri, "get_list")
expected_fields = [
"short_descr",
"created_by",
"changed_by",
"start_dttm",
"end_dttm",
]
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == ANNOTATIONS_COUNT
for expected_field in expected_fields:
assert expected_field in data["result"][0]
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_list_annotation_sorting(self):
"""
Annotation Api: Test sorting on get list of annotations
"""
layer = self.get_layer_with_annotation()
self.login(username="admin")
order_columns = [
"short_descr",
"created_by.first_name",
"changed_by.first_name",
"changed_on_delta_humanized",
"start_dttm",
"end_dttm",
]
for order_column in order_columns:
arguments = {"order_column": order_column, "order_direction": "asc"}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
@pytest.mark.usefixtures("create_annotation_layers")
def test_get_list_annotation_filter(self):
"""
Annotation Api: Test filters on get list annotation layers
"""
layer = self.get_layer_with_annotation()
self.login(username="admin")
arguments = {
"filters": [
{"col": "short_descr", "opr": "annotation_all_text", "value": "2"}
]
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
arguments = {
"filters": [
{"col": "short_descr", "opr": "annotation_all_text", "value": "descr3"}
]
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/?q={prison.dumps(arguments)}"
rv = self.get_assert_metric(uri, "get_list")
assert rv.status_code == 200
data = json.loads(rv.data.decode("utf-8"))
assert data["count"] == 1
@pytest.mark.usefixtures("create_annotation_layers")
def test_create_annotation(self):
"""
Annotation Api: Test create annotation
"""
layer = self.get_layer_with_annotation()
self.login(username="admin")
annotation_data = {
"short_descr": "new",
"long_descr": "description",
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/"
rv = self.client.post(uri, json=annotation_data)
assert rv.status_code == 201
data = json.loads(rv.data.decode("utf-8"))
created_model: Annotation = db.session.query(Annotation).get(data.get("id"))
assert created_model is not None
assert created_model.short_descr == annotation_data["short_descr"]
assert created_model.long_descr == annotation_data["long_descr"]
# Rollback changes
db.session.delete(created_model)
db.session.commit()
@pytest.mark.usefixtures("create_annotation_layers")
def test_create_annotation_uniqueness(self):
"""
Annotation Api: Test create annotation uniqueness
"""
layer = self.get_layer_with_annotation()
self.login(username="admin")
annotation_data = {
"short_descr": "short_descr2",
"long_descr": "description",
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/"
rv = self.client.post(uri, json=annotation_data)
assert rv.status_code == 422
data = json.loads(rv.data.decode("utf-8"))
assert data == {
"message": {
"short_descr": ["Short description must be unique for this layer"]
}
}
@pytest.mark.usefixtures("create_annotation_layers")
def test_update_annotation(self):
"""
Annotation Api: Test update annotation
"""
layer = self.get_layer_with_annotation()
annotation = (
db.session.query(Annotation)
.filter(Annotation.short_descr == "short_descr2")
.one_or_none()
)
self.login(username="admin")
annotation_layer_data = {
"short_descr": "changed_name",
"long_descr": "changed_description",
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/{annotation.id}"
rv = self.client.put(uri, json=annotation_layer_data)
assert rv.status_code == 200
updated_model: Annotation = db.session.query(Annotation).get(annotation.id)
assert updated_model is not None
assert updated_model.short_descr == annotation_layer_data["short_descr"]
assert updated_model.long_descr == annotation_layer_data["long_descr"]
@pytest.mark.usefixtures("create_annotation_layers")
def test_update_annotation_uniqueness(self):
"""
Annotation Api: Test update annotation uniqueness
"""
layer = self.get_layer_with_annotation()
annotation = (
db.session.query(Annotation)
.filter(Annotation.short_descr == "short_descr2")
.one_or_none()
)
self.login(username="admin")
annotation_layer_data = {
"short_descr": "short_descr3",
"long_descr": "changed_description",
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/{annotation.id}"
rv = self.client.put(uri, json=annotation_layer_data)
assert rv.status_code == 422
data = json.loads(rv.data.decode("utf-8"))
assert data == {
"message": {
"short_descr": ["Short description must be unique for this layer"]
}
}
@pytest.mark.usefixtures("create_annotation_layers")
def test_update_annotation_not_found(self):
"""
Annotation Api: Test update annotation not found
"""
layer = self.get_layer_with_annotation()
max_id = db.session.query(func.max(Annotation.id)).scalar()
self.login(username="admin")
annotation_layer_data = {
"short_descr": "changed_name",
}
uri = f"api/v1/annotation_layer/{layer.id}/annotation/{max_id + 1}"
rv = self.client.put(uri, json=annotation_layer_data)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_delete_annotation(self):
"""
Annotation Api: Test update annotation
"""
layer = self.get_layer_with_annotation()
annotation = (
db.session.query(Annotation)
.filter(Annotation.short_descr == "short_descr1")
.one_or_none()
)
self.login(username="admin")
uri = f"api/v1/annotation_layer/{layer.id}/annotation/{annotation.id}"
rv = self.client.delete(uri)
assert rv.status_code == 200
updated_model = db.session.query(Annotation).get(annotation.id)
assert updated_model is None
@pytest.mark.usefixtures("create_annotation_layers")
def test_delete_annotation_not_found(self):
"""
Annotation Api: Test delete annotation not found
"""
layer = self.get_layer_with_annotation()
max_id = db.session.query(func.max(Annotation.id)).scalar()
self.login(username="admin")
uri = f"api/v1/annotation_layer/{layer.id}/annotation{max_id + 1}"
rv = self.client.delete(uri)
assert rv.status_code == 404
@pytest.mark.usefixtures("create_annotation_layers")
def test_bulk_delete_annotation(self):
"""
Annotation Api: Test bulk delete annotation
"""
layer = self.get_layer_with_annotation()
query_annotations = db.session.query(Annotation).filter(
Annotation.layer == layer
)
annotations = query_annotations.all()
annotations_ids = [annotation.id for annotation in annotations]
self.login(username="admin")
uri = f"api/v1/annotation_layer/{layer.id}/annotation/?q={prison.dumps(annotations_ids)}"
rv = self.client.delete(uri)
assert rv.status_code == 200
deleted_annotations = query_annotations.all()
assert deleted_annotations == []
response = json.loads(rv.data.decode("utf-8"))
expected_response = {"message": f"Deleted {len(annotations_ids)} annotations"}
assert response == expected_response
@pytest.mark.usefixtures("create_annotation_layers")
def test_bulk_delete_annotation_not_found(self):
"""
Annotation Api: Test bulk delete annotation not found
"""
layer = self.get_layer_with_annotation()
query_annotations = db.session.query(Annotation).filter(
Annotation.layer == layer
)
annotations = query_annotations.all()
annotations_ids = [annotation.id for annotation in annotations]
max_id = db.session.query(func.max(Annotation.id)).scalar()
annotations_ids.append(max_id + 1)
self.login(username="admin")
uri = f"api/v1/annotation_layer/{layer.id}/annotation/?q={prison.dumps(annotations_ids)}"
rv = self.client.delete(uri)
assert rv.status_code == 404