From 1629b6055898aeee2ff065b11935fcac0e9964b6 Mon Sep 17 00:00:00 2001 From: Kirill Chernakov Date: Mon, 23 Sep 2024 21:33:19 +0400 Subject: [PATCH] tests: add tests for topology service methods and endpoints --- tests/test_topology.py | 274 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 274 insertions(+) create mode 100644 tests/test_topology.py diff --git a/tests/test_topology.py b/tests/test_topology.py new file mode 100644 index 000000000..225d97bb4 --- /dev/null +++ b/tests/test_topology.py @@ -0,0 +1,274 @@ +from datetime import datetime +import uuid +import pytest +from sqlmodel import select + +from keep.api.core.dependencies import SINGLE_TENANT_UUID +from keep.api.models.db.topology import ( + TopologyApplication, + TopologyApplicationDtoIn, + TopologyService, + TopologyServiceDependency, + TopologyServiceDtoIn, +) +from keep.topologies.topologies_service import ( + TopologiesService, + ApplicationNotFoundException, + InvalidApplicationDataException, + ServiceNotFoundException, +) +from tests.fixtures.client import setup_api_key, client, test_app + + +VALID_API_KEY = "valid_api_key" + + +def create_service(db_session, tenant_id, id): + service = TopologyService( + tenant_id=tenant_id, + service="test_service_" + id, + display_name="Test Service", + repository="test_repository", + tags=["test_tag"], + description="test_description", + team="test_team", + email="test_email", + slack="test_slack", + updated_at=datetime.now(), + ) + db_session.add(service) + db_session.commit() + return service + + +def test_get_all_topology_data(db_session): + service_1 = create_service(db_session, SINGLE_TENANT_UUID, "1") + service_2 = create_service(db_session, SINGLE_TENANT_UUID, "2") + + result = TopologiesService.get_all_topology_data(SINGLE_TENANT_UUID, db_session) + # We have no dependencies, so we should not return any services + assert len(result) == 0 + + dependency = TopologyServiceDependency( + service_id=service_1.id, + depends_on_service_id=service_2.id, + updated_at=datetime.now(), + ) + db_session.add(dependency) + db_session.commit() + + result = TopologiesService.get_all_topology_data(SINGLE_TENANT_UUID, db_session) + assert len(result) == 1 + assert result[0].service == "test_service_1" + + result = TopologiesService.get_all_topology_data( + SINGLE_TENANT_UUID, db_session, include_empty_deps=True + ) + assert len(result) == 2 + assert result[0].service == "test_service_1" + assert result[1].service == "test_service_2" + + +def test_get_applications_by_tenant_id(db_session): + service_1 = create_service(db_session, SINGLE_TENANT_UUID, "1") + service_2 = create_service(db_session, SINGLE_TENANT_UUID, "2") + application = TopologyApplication( + tenant_id=SINGLE_TENANT_UUID, + name="Test Application", + services=[service_1, service_2], + ) + db_session.add(application) + db_session.commit() + + result = TopologiesService.get_applications_by_tenant_id( + SINGLE_TENANT_UUID, db_session + ) + assert len(result) == 1 + assert result[0].name == "Test Application" + assert len(result[0].services) == 2 + + +def test_create_application_by_tenant_id(db_session): + application_dto = TopologyApplicationDtoIn(name="New Application", services=[]) + + with pytest.raises(InvalidApplicationDataException): + TopologiesService.create_application_by_tenant_id( + SINGLE_TENANT_UUID, application_dto, db_session + ) + + application_dto.services.append(TopologyServiceDtoIn(id=123)) + with pytest.raises(ServiceNotFoundException): + TopologiesService.create_application_by_tenant_id( + SINGLE_TENANT_UUID, application_dto, db_session + ) + + application_dto.services = [] + + service_1 = create_service(db_session, SINGLE_TENANT_UUID, "1") + service_2 = create_service(db_session, SINGLE_TENANT_UUID, "2") + + application_dto.services.append(TopologyServiceDtoIn(id=service_1.id)) + application_dto.services.append(TopologyServiceDtoIn(id=service_2.id)) + + result = TopologiesService.create_application_by_tenant_id( + SINGLE_TENANT_UUID, application_dto, db_session + ) + assert result.name == "New Application" + + result = TopologiesService.get_applications_by_tenant_id( + SINGLE_TENANT_UUID, db_session + ) + print(result) + assert len(result) == 1 + assert result[0].name == "New Application" + assert len(result[0].services) == 2 + assert result[0].services[0].service == "test_service_1" + assert result[0].services[1].service == "test_service_2" + + +def test_update_application_by_id(db_session): + application = TopologyApplication( + tenant_id=SINGLE_TENANT_UUID, name="Old Application" + ) + db_session.add(application) + db_session.commit() + + application_dto = TopologyApplicationDtoIn(name="Updated Application", services=[]) + + random_uuid = uuid.uuid4() + with pytest.raises(ApplicationNotFoundException): + TopologiesService.update_application_by_id( + SINGLE_TENANT_UUID, random_uuid, application_dto, db_session + ) + + result = TopologiesService.update_application_by_id( + SINGLE_TENANT_UUID, application.id, application_dto, db_session + ) + assert result.name == "Updated Application" + + +def test_delete_application_by_id(db_session): + application = TopologyApplication( + tenant_id=SINGLE_TENANT_UUID, name="Test Application" + ) + db_session.add(application) + db_session.commit() + + TopologiesService.delete_application_by_id( + SINGLE_TENANT_UUID, application.id, db_session + ) + result = db_session.exec( + select(TopologyApplication).where(TopologyApplication.id == application.id) + ).first() + assert result is None + + +@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True) +def test_get_applications(db_session, client, test_app): + setup_api_key( + db_session, VALID_API_KEY, tenant_id=SINGLE_TENANT_UUID, role="webhook" + ) + + service_1 = create_service(db_session, SINGLE_TENANT_UUID, "1") + service_2 = create_service(db_session, SINGLE_TENANT_UUID, "2") + + application = TopologyApplication( + tenant_id=SINGLE_TENANT_UUID, + name="Test Application", + services=[service_1, service_2], + ) + db_session.add(application) + db_session.commit() + + response = client.get( + "/topology/applications", headers={"x-api-key": VALID_API_KEY} + ) + assert response.status_code == 200 + assert len(response.json()) == 1 + assert response.json()[0]["name"] == "Test Application" + + +@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True) +def test_create_application(db_session, client, test_app): + setup_api_key( + db_session, VALID_API_KEY, tenant_id=SINGLE_TENANT_UUID, role="webhook" + ) + + service = create_service(db_session, SINGLE_TENANT_UUID, "1") + + application_data = {"name": "New Application", "services": [{"id": service.id}]} + + response = client.post( + "/topology/applications", + json=application_data, + headers={"x-api-key": VALID_API_KEY}, + ) + assert response.status_code == 200 + assert response.json()["name"] == "New Application" + + +@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True) +def test_update_application(db_session, client, test_app): + setup_api_key( + db_session, VALID_API_KEY, tenant_id=SINGLE_TENANT_UUID, role="webhook" + ) + + application = TopologyApplication( + tenant_id=SINGLE_TENANT_UUID, name="Old Application" + ) + db_session.add(application) + db_session.commit() + + update_data = { + "name": "Updated Application", + } + + random_uuid = uuid.uuid4() + response = client.put( + f"/topology/applications/{random_uuid}", + json=update_data, + headers={"x-api-key": VALID_API_KEY}, + ) + assert response.status_code == 404 + + response = client.put( + f"/topology/applications/{application.id}", + json=update_data, + headers={"x-api-key": VALID_API_KEY}, + ) + assert response.status_code == 200 + assert response.json()["name"] == "Updated Application" + + invalid_update_data = {"name": "Invalid Application", "services": [{"id": "123"}]} + + response = client.put( + f"/topology/applications/{application.id}", + json=invalid_update_data, + headers={"x-api-key": VALID_API_KEY}, + ) + assert response.status_code == 400 + + +@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True) +def test_delete_application(db_session, client, test_app): + setup_api_key( + db_session, VALID_API_KEY, tenant_id=SINGLE_TENANT_UUID, role="webhook" + ) + random_uuid = uuid.uuid4() + + response = client.delete( + f"/topology/applications/{random_uuid}", headers={"x-api-key": VALID_API_KEY} + ) + assert response.status_code == 404 + + application = TopologyApplication( + tenant_id=SINGLE_TENANT_UUID, name="Test Application" + ) + db_session.add(application) + db_session.commit() + + response = client.delete( + f"/topology/applications/{application.id}", headers={"x-api-key": VALID_API_KEY} + ) + assert response.status_code == 200 + assert response.json()["message"] == "Application deleted successfully"