starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 00:08:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: Summary PyTest in super cluster (#5938)

Browse Source
This commit is contained in:
Shyam Panjiyar
2025-02-13 12:54:27 +05:30
committed by GitHub
parent bc850b131d
commit 221821dbb9
3 changed files with 558 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/api-testing.yml
+1
View File
@@ -13,6 +13,7 @@ env:
ZO_ROOT_USER_EMAIL: root@example.com
ZO_ROOT_USER_PASSWORD: Complexpass#123
ZO_BASE_URL: http://localhost:5080/
ZO_BASE_URL_SC: http://localhost:5080/
jobs:
api_integration_tests:
.github/workflows/playwright.yml
+6 -1
View File
@@ -17,6 +17,7 @@ env:
ZO_ROOT_USER_EMAIL: root@example.com
ZO_ROOT_USER_PASSWORD: Complexpass#123
ZO_BASE_URL: http://localhost:5080
WS_ZO_BASE_URL: ws://localhost:5080
ZO_BASE_URL_SC: http://localhost:5080
INGESTION_URL: http://localhost:5080
ORGNAME: default
@@ -29,6 +30,7 @@ env:
ZO_USAGE_PUBLISH_INTERVAL: 2
ZO_USAGE_REPORTING_ENABLED: true
ZO_MIN_AUTO_REFRESH_INTERVAL: 5
ZO_WEBSOCKET_ENABLED: true
jobs:
build_binary:
@@ -150,6 +152,7 @@ jobs:
echo "ZO_ROOT_USER_EMAIL=${ZO_ROOT_USER_EMAIL}" >> .env
echo "ZO_ROOT_USER_PASSWORD=${ZO_ROOT_USER_PASSWORD}" >> .env
echo "ZO_BASE_URL=${ZO_BASE_URL}" >> .env
echo "WS_ZO_BASE_URL=${WS_ZO_BASE_URL}" >> .env
echo "ZO_BASE_URL_SC=${ZO_BASE_URL_SC}" >> .env
echo "INGESTION_URL=${INGESTION_URL}" >> .env
echo "ORGNAME=${ORGNAME}" >> .env
@@ -220,4 +223,6 @@ jobs:
with:
name: coverage-report
path: tests/ui-testing/coverage-report
retention-days: 30
retention-days: 7
tests/api-testing/tests/test_summary.py
+551
View File
@@ -0,0 +1,551 @@
import os # For environment variables
from requests.auth import HTTPBasicAuth
from datetime import datetime, timezone, timedelta # For timestamp conversion
import time # For sleep
from pathlib import Path # For file paths
import pytest # For fixtures
import random
import uuid
import string
import requests # or your HTTP client of choice
# Constants for WebSocket URL and user credentials
ZO_BASE_URL_SC = os.environ.get("ZO_BASE_URL_SC") # Use environment variable
ZO_ROOT_USER_EMAIL = os.environ.get("ZO_ROOT_USER_EMAIL") # Use environment variable
ZO_ROOT_USER_PASSWORD = os.environ.get("ZO_ROOT_USER_PASSWORD") # Use environment variable
# Directory for test data
root_dir = Path(__file__).parent.parent.parent
def generate_random_string(length=5):
"""Generate a random string of lowercase letters."""
characters = string.ascii_lowercase # Only lowercase letters
return ''.join(random.choice(characters) for _ in range(length))
# Generate a random stream name
random_string = generate_random_string()
# Organization ID
org_id = random_string
stream_name = "tdef" + random_string
print("Random Org ID:", org_id)
print("Random Stream:", stream_name)
@pytest.fixture
def create_session():
"""Create a session for API requests."""
session = requests.Session() # Create a new session
# Optionally, set up any headers or authentication here
yield session # This will make the session available to tests
session.close() # Clean up after tests
def test_data(create_session, base_url):
"""Ingest data into the OpenObserve running instance."""
session = create_session
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD)
# Open the JSON data file and read it
with open(root_dir / "test-data/logs_data.json") as f:
data = f.read()
url_ing = f"{base_url}api/{org_id}/{stream_name}/_json"
print("Ingested URL", url_ing) # Add this for debugging
resp_ing = session.post(url_ing, data=data, headers={"Content-Type": "application/json"})
print("Data ingested successfully for Summary, status code:", resp_ing.status_code)
print("Response content:", resp_ing.text)
assert resp_ing.status_code == 200, f"Data ingestion failed, status code: {resp_ing.status_code}"
# Wait for a few seconds to allow the data to be ingested
time.sleep(5) # Increase this time if necessary
# Payload for pipeline creation
pipeline_payload = {
"name": "p12345",
"description": "",
"source": {"source_type": "realtime"},
"nodes": [
{
"id": "1e1fa129-cfbd-4082-8d39-c4dc418a670f",
"type": "input",
"data": {
"node_type": "stream",
"stream_name": stream_name,
"stream_type": "logs",
"org_id": org_id,
},
"position": {"x": 169.83333, "y": 55},
"io_type": "input",
},
{
"id": "8e0ed123-7737-4801-9466-bd909c425c72",
"type": "output",
"data": {
"node_type": "stream",
"stream_name": stream_name,
"stream_type": "logs",
"org_id": org_id,
},
"position": {"x": 305.33334, "y": 276},
"io_type": "output",
},
],
"edges": [
{
"id": "e1e1fa129-cfbd-4082-8d39-c4dc418a670f-8e0ed123-7737-4801-9466-bd909c425c72",
"source": "1e1fa129-cfbd-4082-8d39-c4dc418a670f",
"target": "8e0ed123-7737-4801-9466-bd909c425c72",
}
],
"org": org_id,
}
# Create a pipeline
resp_create_pipeline = session.post(f"{base_url}api/{org_id}/pipelines", json=pipeline_payload)
assert resp_create_pipeline.status_code == 200, f"Expected status code 200 but got {resp_create_pipeline.status_code}"
print("Create pipeline response:", resp_create_pipeline.json())
# Wait for a few seconds to allow the pipeline to be created
time.sleep(5) # Increase this time if necessary
# Create new alert
alert_name = f"newalert_{random.randint(1000, 9999)}" # Make the name unique
folder_alert = f"newfloder_{random.randint(1000, 9999)}" # Make the name unique
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
# create folder
url=base_url
payload_folder = {"description": "newfoldernvp", "name": folder_alert}
resp_create_folder = session.post(
f"{url}api/v2/{org_id}/folders/alerts", json=payload_folder)
print(resp_create_folder.content)
folder_id = resp_create_folder.json()["folderId"]
assert (
resp_create_folder.status_code == 200
), f"Expected 200, but got {resp_create_folder.status_code} {resp_create_folder.content}"
# createtemplate
template_alert = f"newtemp_{random.randint(10000, 99999)}" # Make the name unique
destination_alert = f"newdest_{random.randint(10000, 99999)}" # Make the name unique
payload_temp_alert = {
"name":template_alert,
"body":"{\n \"text\": \"For stream {stream_name} of organization {org_name} alert {alert_name} of type {alert_type} is active\"\n}",
"type":"http",
"title":""
}
# Create template for alerts
resp_create_templates_alert = session.post(
f"{url}api/{org_id}/alerts/templates", json=payload_temp_alert
)
print(resp_create_templates_alert.content)
assert (
resp_create_templates_alert.status_code == 200
), f"Create template 200, but got {resp_create_templates_alert.status_code} {resp_create_templates_alert.content}"
skip_tls_verify_value = False # Define the skip_tls_verify_value
payload_dest_alert = {
"url":"https://jsonplaceholder.typicode.com/todos",
"method":"get",
"skip_tls_verify":skip_tls_verify_value,
"template":template_alert,
"headers":{},
"name":destination_alert
}
# Create destination
resp_create_destinations_alert = session.post(
f"{url}api/{org_id}/alerts/destinations",
json=payload_dest_alert,
)
print(resp_create_destinations_alert.content)
assert (
resp_create_destinations_alert.status_code == 200
), f"Failed to create destination: {resp_create_destinations_alert.status_code} {resp_create_destinations_alert.content}"
# Get destination
resp_get_destinations_alert = session.get(
f"{url}api/{org_id}/alerts/destinations/{destination_alert}"
)
print(resp_get_destinations_alert.content)
assert (
resp_get_destinations_alert.status_code == 200
), f"Get all destinations list 200, but got {resp_get_destinations_alert.status_code} {resp_get_destinations_alert.content}"
# Check if the destination name matches
assert (
resp_get_destinations_alert.json()["name"] == destination_alert
), f"Destination {destination_alert} not found in the response"
time.sleep(5)
# Create logs
payload_logs = [
{
"Athlete": "newtemp",
"City": "Athens",
"Country": "HUN",
"Discipline": "Swimming",
"Sport": "Aquatics",
"Year": 1896,
},
{
"Athlete": "HERSCHMANN",
"City": "Athens",
"Country": "CHN",
"Discipline": "Swimming",
"Sport": "Aquatics",
"Year": 1896,
},
]
resp_create_logstream = session.post(
f"{url}api/{org_id}/{stream_name}/_json", json=payload_logs
)
print(resp_create_logstream.content)
assert (
resp_create_logstream.status_code == 200
), f"Get all logs list 200, but got {resp_create_logstream.status_code} {resp_create_logstream.content}"
time.sleep(5)
payload_alert = {
"name": alert_name,
"row_template": template_alert,
"stream_type": "logs",
"stream_name": stream_name,
"is_real_time": False,
"context_attributes": {},
"query_condition": {
"conditions": [ {
"column": "log",
"operator": "=",
"value": "200",
"type": None, # Set it to None or remove it
"id": str(uuid.uuid4())
} ],
"search_event_type": "ui",
"sql": "",
"promql": "",
"type": "custom",
"promql_condition": None,
"vrl_function": None,
"multi_time_range": []
},
"trigger_condition": {
"period": 10,
"operator": ">=",
"frequency": 1,
"cron": "",
"threshold": 3,
"silence": 10,
"frequency_type": "minutes",
"timezone": "UTC",
"tolerance_in_secs": 0
},
"org_id": "default",
"destinations": [destination_alert],
"enabled": True,
"description": "test",
"folderId": folder_id
}
resp_post_alertnew = session.post(
f"{url}api/v2/{org_id}/alerts",
json=payload_alert,
headers=headers,
)
print(resp_post_alertnew.content)
assert ( resp_post_alertnew.status_code == 200
), f"Post alert expected 200, but got {resp_post_alertnew.status_code} {resp_post_alertnew.content}"
print(f"Alert {alert_name} created successfully")
time.sleep(5) # Increase this time if necessary
# Make the request to create function
payload_function = {
"function": ".a=190025552",
"name": "pytestfunction",
"params": "row",
"transType": 0,
}
resp_create_function = session.post(
f"{base_url}api/{org_id}/functions", json=payload_function
)
print(resp_create_function.content)
assert (
resp_create_function.status_code == 200
), f"Expected 200, but got {resp_create_function.status_code} {resp_create_function.content}"
time.sleep(5) # Increase this time if necessary
# Make the request to create dashboard
payload_dashboard = {
"description": "pydashboards",
"title": "pytestdashboard",
"owner": ZO_ROOT_USER_EMAIL,
}
resp_create_dashboard = session.post(
f"{base_url}api/{org_id}/dashboards", json=payload_dashboard
)
print(resp_create_dashboard.content)
assert (
resp_create_dashboard.status_code == 200
), f"Expected 200, but got {resp_create_dashboard.status_code} {resp_create_dashboard.content}"
print("Dashboard created successfully")
@pytest.fixture
def base_url_sc():
"""Provide the base URL for the API of Super Cluster."""
return ZO_BASE_URL_SC
def test_summary(create_session, base_url_sc):
"""Run an E2E test for summary mode."""
session = create_session
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD) # Add this line
url_sc = base_url_sc
print("URL for super cluster:", url_sc)
time.sleep(5) # Increase this time if necessary
res_summary_sc = session.get(f"{url_sc}api/{org_id}/summary")
print("Summary SC response:", res_summary_sc.content) # Add this for debugging
assert res_summary_sc.status_code == 200, f"Summary failed, status code: {res_summary_sc.status_code}"
res_data_summary_sc = res_summary_sc.json()
print("Summary SC data:", res_data_summary_sc) # Add this for debugging
# Validate the number of streams
actual_num_streams_sc= res_data_summary_sc["streams"]["num_streams"]
expected_num_streams_sc = 1 # Adjust based on our expectations
assert actual_num_streams_sc == expected_num_streams_sc, f"Expected to be {expected_num_streams_sc}, but got {actual_num_streams_sc}"
# Validate the number of pipelines
pipelines_data_sc = res_data_summary_sc.get("pipelines", {})
actual_num_pipelines_sc = pipelines_data_sc.get("num_realtime", 0) # Default to 0 if not found
expected_num_pipelines_sc = 1 # Adjust based on your expectations
assert actual_num_pipelines_sc == expected_num_pipelines_sc, f"Expected to be {expected_num_pipelines_sc}, but got {actual_num_pipelines_sc}"
# Validate the number of alerts
alerts_data_sc = res_data_summary_sc.get("alerts", {})
# Validate the number of scheduled alerts
actual_num_scheduled_alerts_sc = alerts_data_sc.get("num_scheduled", 0) # Default to 0 if not found
expected_num_scheduled_alerts_sc = 1 # Adjust based on your expectations
assert actual_num_scheduled_alerts_sc == expected_num_scheduled_alerts_sc, (
f"Expected {expected_num_scheduled_alerts_sc} scheduled alerts, "
f"but got {actual_num_scheduled_alerts_sc}"
)
# Validate the total number of functions
actual_num_functions_sc = res_data_summary_sc.get("total_functions", 0) # Directly get the value
expected_num_functions_sc = 1 # Adjust based on your expectations
assert actual_num_functions_sc == expected_num_functions_sc, (
f"Expected total_functions to be {expected_num_functions_sc}, "
f"but got {actual_num_functions_sc}"
)
# Validate the number of dashboards
actual_num_dashboards_sc = res_data_summary_sc.get("total_dashboards", 0) # Directly get the value
expected_num_dashboards_sc = 1 # Adjust based on your expectations
assert actual_num_dashboards_sc == expected_num_dashboards_sc, (
f"Expected to be {expected_num_dashboards_sc}, but got {actual_num_dashboards_sc}"
)
# Retrieve the pipeline list
resp_list_pipelines = session.get(f"{url_sc}api/{org_id}/pipelines")
assert resp_list_pipelines.status_code == 200, f"Expected status code 200 but got {resp_list_pipelines.status_code}"
# Parse the pipeline list response
pipelines_list = resp_list_pipelines.json().get("list", [])
assert pipelines_list, "No pipelines found in the list response."
# Extract the pipeline_id for the created pipeline
pipeline_id = None
for pipeline in pipelines_list:
if pipeline["name"] == "p12345":
pipeline_id = pipeline["pipeline_id"]
break
assert pipeline_id, "Pipeline ID not found for the created pipeline."
print(f"Pipeline ID: {pipeline_id}")
resp_get_pipeline = session.get(f"{url_sc}api/{org_id}/pipelines")
print(resp_get_pipeline.json())
assert resp_get_pipeline.status_code == 200
pipeline_list = resp_get_pipeline.json()["list"]
target_pipeline = next((p for p in pipeline_list if p["pipeline_id"] == pipeline_id), None)
assert target_pipeline is not None, f"Pipeline {pipeline_id} not found in the list"
# Delete the pipeline
resp_delete_pipeline = session.delete(f"{url_sc}api/{org_id}/pipelines/{pipeline_id}")
assert resp_delete_pipeline.status_code == 200, f"Expected status code 200 but got {resp_delete_pipeline.status_code}"
print(f"Deleted pipeline {pipeline_id}")
# Make the request to get the list of alerts
resp_get_allalertsnew = session.get(f"{url_sc}api/v2/{org_id}/alerts")
# Ensure the response is successful
assert resp_get_allalertsnew.status_code == 200, f"Failed to fetch alerts: {resp_get_allalertsnew.status_code}"
# Parse the response JSON
response_json = resp_get_allalertsnew.json()
# Check if "list" is in the response and proceed
assert "list" in response_json, "Response does not contain 'list'"
# Get the list of alerts from the response
alerts = response_json["list"]
# Now you can iterate over the alerts
for alert in alerts:
alert_id = alert.get("alert_id")
assert alert_id, f"Alert ID is missing for alert: {alert}"
print(f"Extracted alert_id: {alert_id}")
# Validate the alert existence first
resp_check_alert = session.get(f"{url_sc}api/v2/{org_id}/alerts/{alert_id}")
assert resp_check_alert.status_code == 200, f"Alert {alert_id} does not exist or cannot be retrieved."
print(f"Alert {alert_id} exists and is retrievable.")
# Proceed to delete the alert
resp_delete_alertnew = session.delete(f"{url_sc}api/v2/{org_id}/alerts/{alert_id}")
print(f"Deleted Alert Response: {resp_delete_alertnew.text}")
assert resp_delete_alertnew.status_code == 200, f"Failed to delete alert {alert_id}"
print(f"Successfully deleted alert {alert_id}")
# Proceed to delete the function
resp_delete_function = session.delete(
f"{url_sc}api/{org_id}/functions/pytestfunction"
)
assert (
resp_delete_function.status_code == 200
), f"Deleting this function, but got {resp_delete_function.status_code} {resp_delete_function.content}"
print(f"Function deleted successfully")
# Proceed to delete the dashboard
resp_get_alldashboards = session.get(f"{url_sc}api/{org_id}/dashboards")
assert resp_get_alldashboards.status_code == 200, f"Expected status code 200 but got {resp_get_alldashboards.status_code}"
# Parse the response JSON
dashboards_response = resp_get_alldashboards.json()
print(f"Dashboard list response: {dashboards_response}")
# Extract the dashboard ID
dashboard_id = dashboards_response['dashboards'][0]['v1']['dashboardId'] # Corrected line
print(f"Extracted Dashboard ID: {dashboard_id}")
# Now you can delete the dashboard using the extracted ID
resp_delete_dashboard = session.delete(
f"{url_sc}api/{org_id}/dashboards/{dashboard_id}"
)
# Assert that the deletion was successful
assert resp_delete_dashboard.status_code == 200, f"Failed to delete dashboard, status code: {resp_delete_dashboard.status_code}"
print(f"Successfully deleted dashboard with ID: {dashboard_id}")
# Proceed to delete the stream
resp_delete_stream= session.delete(f"{url_sc}api/{org_id}/streams/{stream_name}?type=logs")
print(f"Deleted Stream Response: {resp_delete_stream.text}")
assert resp_delete_stream.status_code == 200, f"Failed to delete {stream_name}"
print(f"Successfully deleted stream {stream_name}")
def test_summary_validate(create_session, base_url):
"""Run an E2E test for summary mode."""
session = create_session
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD) # Add this line
print("Base URL:", base_url)
time.sleep(5) # Increase this time if necessary
# Retrieve the pipeline list
resp_list_pipelines = session.get(f"{base_url}api/{org_id}/pipelines")
assert resp_list_pipelines.status_code == 200, f"Expected status code 200 but got {resp_list_pipelines.status_code}"
# Parse the pipeline list response
pipelines_list = resp_list_pipelines.json().get("list", [])
# Assert that the pipelines_list is either empty or contains elements
if not pipelines_list:
print("No pipelines found in the list response.")
else:
print(f"Found {len(pipelines_list)} pipelines.")
# Make the request to get the list of alerts
resp_get_allalertsnew = session.get(f"{base_url}api/v2/{org_id}/alerts")
# Ensure the response is successful
assert resp_get_allalertsnew.status_code == 200, f"Failed to fetch alerts: {resp_get_allalertsnew.status_code}"
# Parse the response JSON
response_json = resp_get_allalertsnew.json()
# Check if "list" is in the response and proceed
assert "list" in response_json, "Response does not contain 'list'"
# Get the list of alerts from the response
alerts = response_json["list"]
# Check if the alerts list is empty
if not alerts:
print("No alerts found.")
else:
# Now you can iterate over the alerts
for alert in alerts:
alert_id = alert.get("alert_id")
assert alert_id, f"Alert ID is missing for alert: {alert}"
print(f"Extracted alert_id: {alert_id}")
# Fetch the functions
resp_ver_function = session.get(f"{base_url}api/{org_id}/functions")
print(f"Function verification response: {resp_ver_function.status_code}, content: {resp_ver_function.content}")
# Parse the response JSON
response_json = resp_ver_function.json()
# Assert that the status code indicates success
assert resp_ver_function.status_code == 200, f"Expected status code 200 but got {resp_ver_function.status_code}"
# Check if the "list" is present and assert that it's empty
assert "list" in response_json, "Response does not contain 'list'"
assert response_json["list"] == [], "Expected 'list' to be empty, but it contains elements."
# Fetch the dashboards
resp_ver_dashboard = session.get(f"{base_url}api/{org_id}/dashboards")
print(f"Dashboard verification response: {resp_ver_dashboard.status_code}, content: {resp_ver_dashboard.content}")
# Fetch the list of dashboards after deletion
assert resp_ver_dashboard.status_code == 200, f"Expected status code 200 but got {resp_ver_dashboard.status_code}"
# Parse the response JSON
dashboards_response = resp_ver_dashboard.json()
print(f"Dashboard list response: {dashboards_response}")
# Assert that the dashboards list is empty
assert 'dashboards' in dashboards_response, "Response does not contain 'dashboards'"
assert dashboards_response['dashboards'] == [], "Expected the dashboards list to be empty after deletion."
print("Successfully verified that the dashboard list is empty after deletion.")
res_summary = session.get(f"{base_url}api/{org_id}/summary")
print("Summary response:", res_summary.content) # Add this for debugging
assert res_summary.status_code == 200, f"Summary failed, status code: {res_summary.status_code}"
res_data_summary = res_summary.json()
print("Summary data:", res_data_summary) # Add this for debugging
# Validate the number of streams
actual_num_streams= res_data_summary["streams"]["num_streams"]
expected_num_streams = 0 # Adjust based on our expectations
assert actual_num_streams == expected_num_streams, f"Expected to be {expected_num_streams}, but got {actual_num_streams}"
print("Successfully verified that the number of streams is correct after deletion.")
# Validate the number of pipelines
pipelines_data = res_data_summary.get("pipelines", {})
actual_num_pipelines = pipelines_data.get("num_realtime", 0) # Default to 0 if not found
expected_num_pipelines = 0 # Adjust based on your expectations
assert actual_num_pipelines == expected_num_pipelines, f"Expected to be {expected_num_pipelines}, but got {actual_num_pipelines}"
# Validate the number of alerts
alerts_data = res_data_summary.get("alerts", {})
# Validate the number of scheduled alerts
actual_num_scheduled_alerts = alerts_data.get("num_scheduled", 0) # Default to 0 if not found
expected_num_scheduled_alerts = 0 # Adjust based on your expectations
assert actual_num_scheduled_alerts == expected_num_scheduled_alerts, (
f"Expected {expected_num_scheduled_alerts} scheduled alerts, "
f"but got {actual_num_scheduled_alerts}"
)
# Validate the total number of functions
actual_num_functions = res_data_summary.get("total_functions", 0) # Directly get the value
expected_num_functions = 0 # Adjust based on your expectations
assert actual_num_functions == expected_num_functions, (
f"Expected total_functions to be {expected_num_functions}, "
f"but got {actual_num_functions}"
)
# Validate the number of dashboards
actual_num_dashboards = res_data_summary.get("total_dashboards", 0) # Directly get the value
expected_num_dashboards = 0 # Adjust based on your expectations
assert actual_num_dashboards == expected_num_dashboards, (
f"Expected to be {expected_num_dashboards}, but got {actual_num_dashboards}"
)