68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.urls import reverse
|
|
from rest_framework.test import APIClient
|
|
|
|
from events.tests.utils import mocked_elk_query_raise_elk_exception, mocked_elk_query_valid, mock_elk_indexes_valid
|
|
|
|
mocked_target_elk = 'events.services.elk_string_search.connect_to_elasticsearch_instance'
|
|
|
|
|
|
class TestEventsApi:
|
|
"""Test events api."""
|
|
|
|
user = None
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_test(self, add_user_with_permissions) -> None:
|
|
self.user = add_user_with_permissions(username='testuser0', password='Passwd123', is_superuser=True)
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
@pytest.mark.unit
|
|
@patch(mocked_target_elk, side_effect=mocked_elk_query_valid)
|
|
def test_elk_query_events_api_valid(self, *args):
|
|
url = reverse('elk_query_events', args=['1']) + '?from=0&size=1&sort=desc&track_total_hits=true?q=qweqweqwe: q '
|
|
response = self.client.get(url)
|
|
data = response.json()
|
|
assert response.status_code == 200
|
|
assert 'status' not in data
|
|
assert 'error_message' not in data
|
|
assert 'hits' in data
|
|
assert len(data['hits']['hits']) == 1
|
|
assert data['hits']['total']['value'] == 31
|
|
|
|
@pytest.mark.unit
|
|
@patch(mocked_target_elk, side_effect=lambda *args, **kwargs: None)
|
|
def test_elk_not_connect_to_elk(self, *args):
|
|
url = reverse('elk_query_events', args=['1']) + '?from=0&size=1&sort=desc&track_total_hits=true'
|
|
response = self.client.get(url)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert 'status' in data
|
|
assert 'error_message' in data
|
|
assert data['error_message'] == 'Cannot establish the connection to elasticsearch'
|
|
assert data['status'] == 'err'
|
|
|
|
@pytest.mark.unit
|
|
@patch(mocked_target_elk, side_effect=mocked_elk_query_raise_elk_exception)
|
|
def test_elk_raise_elk_exception(self, *args):
|
|
url = reverse('elk_query_events', args=['3']) + '?from=0&size=1&sort=desc&track_total_hits=true'
|
|
response = self.client.get(url)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert 'status' in data
|
|
assert 'error_message' in data
|
|
assert data['error_message'] == ['Test exception']
|
|
assert data['status'] == 'err'
|
|
|
|
@pytest.mark.unit
|
|
@patch(mocked_target_elk, side_effect=mock_elk_indexes_valid)
|
|
def test_get_valid_indexes_elk(self, *args):
|
|
url = reverse('elk_all_indexes')
|
|
response = self.client.get(url)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data == [{'value': 'aggregated-2022.05.20', 'label': 'aggregated-2022.05.20'},
|
|
{'value': 'aggregated-2022.05.21', 'label': 'aggregated-2022.05.21'}]
|