mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 00:08:55 +00:00
test: Create Objects and Refactoring POM PyTest (#6357)
This commit is contained in:
Shyam Panjiyar
@@ -15,6 +15,7 @@ env:
|
||||
ZO_BASE_URL: http://localhost:5080/
|
||||
ZO_BASE_URL_SC: http://localhost:5080/
|
||||
WS_ZO_BASE_URL: ws://localhost:5080/
|
||||
ZO_SMTP_ENABLED: true
|
||||
|
||||
|
||||
jobs:
|
||||
|
||||
@@ -33,6 +33,7 @@ env:
|
||||
ZO_MIN_AUTO_REFRESH_INTERVAL: 5
|
||||
ZO_WEBSOCKET_ENABLED: true
|
||||
ZO_COLS_PER_RECORD_LIMIT: "80000"
|
||||
ZO_SMTP_ENABLED: true
|
||||
|
||||
jobs:
|
||||
build_binary:
|
||||
@@ -163,6 +164,7 @@ jobs:
|
||||
echo "ZO_BASE_URL_SC_UI=${ZO_BASE_URL_SC_UI}" >> .env
|
||||
echo "INGESTION_URL=${INGESTION_URL}" >> .env
|
||||
echo "ORGNAME=${ORGNAME}" >> .env
|
||||
echo "ZO_SMTP_ENABLED=${ZO_SMTP_ENABLED}" >> .env
|
||||
mv .env tests/ui-testing
|
||||
cd tests/ui-testing && npm ci && npx playwright test ./playwright-tests/${{ matrix.testfilename }}
|
||||
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add the current directory to sys.path
|
||||
current_dir = Path(__file__).parent
|
||||
sys.path.append(str(current_dir))
|
||||
|
||||
from pages.template_page import TemplatePage
|
||||
from pages.folder_page import FolderPage
|
||||
from pages.dashboard_page import DashboardPage
|
||||
from pages.destination_page import DestinationPage
|
||||
from pages.alert_page import AlertPage
|
||||
from pages.user_page import UserPage
|
||||
from pages.enrichment_page import EnrichmentPage
|
||||
from pages.function_page import FunctionPage
|
||||
from pages.pipeline_page import PipelinePage
|
||||
from pages.report_page import ReportPage
|
||||
from pages.savedview_page import SavedViewPage
|
||||
from pages.serviceaccount_page import ServiceAccountPage
|
||||
|
||||
|
||||
# Use environment variable
|
||||
def create_objects(session, base_url, user_email, user_password, org_id, stream_name, num_objects):
|
||||
"""Create objects in the OpenObserve running instance."""
|
||||
|
||||
|
||||
template_page = TemplatePage(session, base_url, org_id)
|
||||
destination_page = DestinationPage(session, base_url, org_id)
|
||||
alert_page = AlertPage(session, base_url, org_id)
|
||||
folder_page = FolderPage(session, base_url, org_id)
|
||||
dashboard_page = DashboardPage(session, base_url, org_id)
|
||||
user_page = UserPage(session, base_url, org_id)
|
||||
enrichment_page = EnrichmentPage(session, base_url, org_id)
|
||||
function_page = FunctionPage(session, base_url, org_id)
|
||||
pipeline_page = PipelinePage(session, base_url, org_id)
|
||||
report_page = ReportPage(session, base_url, org_id)
|
||||
savedview_page = SavedViewPage(session, base_url, org_id)
|
||||
serviceaccount_page = ServiceAccountPage(session, base_url, org_id)
|
||||
|
||||
for i in range(num_objects):
|
||||
|
||||
# Create templates
|
||||
template_name_webhook = f"template_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.create_template_webhook(session, base_url, user_email, user_password, org_id, template_name_webhook)
|
||||
|
||||
template_name_email = f"template_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.create_template_email(session, base_url, user_email, user_password, org_id, template_name_email)
|
||||
|
||||
# Create destinations
|
||||
|
||||
destination_name_webhook = f"destination_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.create_destination_webhook(session, base_url, org_id, user_email, user_password, template_name_webhook, destination_name_webhook)
|
||||
|
||||
email_address_admin = f"admin{user_page.Unique_value_user}{i}@gmail.com"
|
||||
user_page.create_user_admin(session, base_url, user_email, user_password, org_id, email_address_admin)
|
||||
|
||||
destination_name_email = f"destination_email_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.create_destination_email(session, base_url, org_id, user_email, user_password, email_address_admin, template_name_email, destination_name_email)
|
||||
|
||||
destination_name_pipeline = f"destination_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.create_destination_pipeline(session, base_url, org_id, user_email, user_password, destination_name_pipeline)
|
||||
|
||||
folder_name = f"folder_{folder_page.Unique_value_folder}_{i}"
|
||||
folder_id = folder_page.create_folder(session, base_url, user_email, user_password, org_id, folder_name)
|
||||
|
||||
dashboard_name = f"dashboard_{dashboard_page.Unique_value_dashboard}_{i}"
|
||||
dashboard_id = dashboard_page.create_dashboard(session, base_url, user_email, user_password, org_id, stream_name, folder_id, dashboard_name)
|
||||
|
||||
enrichment_name = f"enrichment_{enrichment_page.Unique_value_enrichment}_{i}"
|
||||
enrichment_page.create_enrichment(session, base_url, user_email, user_password, org_id, enrichment_name)
|
||||
|
||||
function_name = f"function_{function_page.Unique_value_function}_{i}"
|
||||
function_page.create_function(session, base_url, user_email, user_password, org_id, function_name)
|
||||
|
||||
realTime_pipeline_name = f"realTime_pipeline_{pipeline_page.Unique_value_pipeline}_{i}"
|
||||
pipeline_page.create_realTime_pipeline(session, base_url, user_email, user_password, org_id, stream_name, realTime_pipeline_name)
|
||||
|
||||
scheduled_pipeline_name = f"scheduled_pipeline_{pipeline_page.Unique_value_pipeline}_{i}"
|
||||
pipeline_page.create_scheduled_pipeline(session, base_url, user_email, user_password, org_id, stream_name, scheduled_pipeline_name)
|
||||
|
||||
scheduled_report_name = f"scheduled_report_{report_page.Unique_value_report}_{i}"
|
||||
report_page.create_scheduled_report(session, base_url, user_email, user_password, org_id, dashboard_id, folder_id, scheduled_report_name)
|
||||
|
||||
cached_report_name = f"cached_report_{report_page.Unique_value_report}_{i}"
|
||||
report_page.create_cached_report(session, base_url, user_email, user_password, org_id, dashboard_id, folder_id, cached_report_name)
|
||||
|
||||
savedview_name = f"savedview_{savedview_page.Unique_value_savedview}_{i}"
|
||||
savedview_page.create_savedView(session, base_url, user_email, user_password, org_id, stream_name, savedview_name)
|
||||
|
||||
email_address_serviceaccount = f"serviceaccount{user_page.Unique_value_user}{i}@gmail.com"
|
||||
serviceaccount_page.create_service_account(session, base_url, user_email, user_password, org_id, email_address_serviceaccount)
|
||||
|
||||
# Create alerts
|
||||
alert_webhook = f"standard_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert(session, base_url, user_email, user_password, org_id, stream_name, template_name_webhook, destination_name_webhook, alert_webhook)
|
||||
|
||||
alert_email = f"alert_email_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert(session, base_url, user_email, user_password, org_id, stream_name, template_name_email, destination_name_email, alert_email)
|
||||
|
||||
alert_cron = f"alert_cron_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert_cron(session, base_url, user_email, user_password, org_id, stream_name, template_name_email, destination_name_email, alert_cron)
|
||||
|
||||
alert_real_time = f"alert_real_time_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_real_time_alert(session, base_url, user_email, user_password, org_id, stream_name, template_name_webhook, destination_name_webhook, alert_real_time)
|
||||
|
||||
alert_sql = f"alert_sql_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert_sql(session, base_url, user_email, user_password, org_id, stream_name, template_name_webhook, destination_name_webhook, alert_sql)
|
||||
|
||||
|
||||
# Created all objects successfully by admin user
|
||||
|
||||
# Create templates
|
||||
|
||||
template_admin_webhook = f"template_admin_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.create_template_webhook(session, base_url, email_address_admin, "12345678", org_id, template_admin_webhook)
|
||||
|
||||
template_admin_email = f"template_admin_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.create_template_email(session, base_url, email_address_admin, "12345678", org_id, template_admin_email)
|
||||
|
||||
# Create destinations
|
||||
|
||||
destination_admin_webhook = f"destination_admin_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.create_destination_webhook(session, base_url, org_id, email_address_admin, "12345678", template_admin_webhook, destination_admin_webhook)
|
||||
|
||||
email_admin = f"aduser{user_page.Unique_value_user}{i}@gmail.com"
|
||||
user_page.create_user_admin(session, base_url, email_address_admin, "12345678", org_id, email_admin)
|
||||
|
||||
destination_admin_email = f"destination_admin_email_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.create_destination_email(session, base_url, org_id, email_address_admin, "12345678", email_admin, template_admin_email, destination_admin_email)
|
||||
|
||||
destination_admin_pipeline = f"destination_admin_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.create_destination_pipeline(session, base_url, org_id, email_address_admin, "12345678", destination_admin_pipeline)
|
||||
|
||||
folder_admin = f"folder_admin_{folder_page.Unique_value_folder}_{i}"
|
||||
folder_id = folder_page.create_folder(session, base_url, email_address_admin, "12345678", org_id, folder_admin)
|
||||
|
||||
dashboard_admin = f"dashboard_admin_{dashboard_page.Unique_value_dashboard}_{i}"
|
||||
dashboard_id = dashboard_page.create_dashboard(session, base_url, email_address_admin, "12345678", org_id, stream_name, folder_id, dashboard_admin)
|
||||
|
||||
enrichment_admin = f"enrichment_admin_{enrichment_page.Unique_value_enrichment}_{i}"
|
||||
enrichment_page.create_enrichment(session, base_url, email_address_admin, "12345678", org_id, enrichment_admin)
|
||||
|
||||
function_admin = f"function_admin_{function_page.Unique_value_function}_{i}"
|
||||
function_page.create_function(session, base_url, email_address_admin, "12345678", org_id, function_admin)
|
||||
|
||||
realTime_pipeline_admin = f"realTime_pipeline_admin_{pipeline_page.Unique_value_pipeline}_{i}"
|
||||
pipeline_page.create_realTime_pipeline(session, base_url, email_address_admin, "12345678", org_id, stream_name, realTime_pipeline_admin)
|
||||
|
||||
scheduled_pipeline_admin = f"scheduled_pipeline_admin_{pipeline_page.Unique_value_pipeline}_{i}"
|
||||
pipeline_page.create_scheduled_pipeline(session, base_url, email_address_admin, "12345678", org_id, stream_name, scheduled_pipeline_admin)
|
||||
|
||||
scheduled_report_admin = f"scheduled_report_admin_{report_page.Unique_value_report}_{i}"
|
||||
report_page.create_scheduled_report(session, base_url, email_address_admin, "12345678", org_id, dashboard_id, folder_id, scheduled_report_admin)
|
||||
|
||||
cached_report_admin = f"cached_report_admin_{report_page.Unique_value_report}_{i}"
|
||||
report_page.create_cached_report(session, base_url, email_address_admin, "12345678", org_id, dashboard_id, folder_id, cached_report_admin)
|
||||
|
||||
savedview_admin = f"savedview_admin_{savedview_page.Unique_value_savedview}_{i}"
|
||||
savedview_page.create_savedView(session, base_url, email_address_admin, "12345678", org_id, stream_name, savedview_admin)
|
||||
|
||||
email_address_serviceaccount_admin = f"serviceaccount_admin{user_page.Unique_value_user}{i}@gmail.com"
|
||||
serviceaccount_page.create_service_account(session, base_url, email_address_admin, "12345678", org_id, email_address_serviceaccount_admin)
|
||||
|
||||
# Create alerts
|
||||
alert_webhook_admin = f"alert_webhook_admin_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert(session, base_url, email_address_admin, "12345678", org_id, stream_name, template_name_webhook, destination_name_webhook, alert_webhook_admin)
|
||||
|
||||
alert_email_admin = f"alert_email_admin_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert(session, base_url, email_address_admin, "12345678", org_id, stream_name, template_name_email, destination_name_email, alert_email_admin)
|
||||
|
||||
alert_cron_admin = f"alert_cron_admin_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert_cron(session, base_url, email_address_admin, "12345678", org_id, stream_name, template_name_email, destination_name_email, alert_cron_admin)
|
||||
|
||||
alert_real_time_admin = f"alert_real_time_admin_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_real_time_alert(session, base_url, email_address_admin, "12345678", org_id, stream_name, template_name_webhook, destination_name_webhook, alert_real_time_admin)
|
||||
|
||||
alert_sql_admin = f"alert_sql_admin_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.create_standard_alert_sql(session, base_url, email_address_admin, "12345678", org_id, stream_name, template_name_webhook, destination_name_webhook, alert_sql_admin)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
from pages.template_page import TemplatePage
|
||||
from pages.destination_page import DestinationPage
|
||||
from pages.alert_page import AlertPage
|
||||
|
||||
def delete_objects(session, base_url, user_email, user_password, org_id, stream_name, num_objects):
|
||||
"""Delete objects in the OpenObserve running instance."""
|
||||
|
||||
template_page = TemplatePage(session, base_url, org_id)
|
||||
destination_page = DestinationPage(session, base_url, org_id)
|
||||
alert_page = AlertPage(session, base_url, org_id)
|
||||
|
||||
for i in range(num_objects):
|
||||
# Delete alerts first since they depend on destinations
|
||||
standard_alert = f"standard_{AlertPage.Unique_value_alert}_{i}"
|
||||
alert_page.delete_alert(session, base_url, user_email, user_password, org_id, stream_name, standard_alert)
|
||||
|
||||
standard_sql_alert = f"standard_sql_{AlertPage.Unique_value_alert}_{i}"
|
||||
alert_page.delete_alert(session, base_url, user_email, user_password, org_id, stream_name, standard_sql_alert)
|
||||
|
||||
standard_cron_alert = f"standard_cron_{AlertPage.Unique_value_alert}_{i}"
|
||||
alert_page.delete_alert(session, base_url, user_email, user_password, org_id, stream_name, standard_cron_alert)
|
||||
|
||||
real_time_alert = f"real_time_{AlertPage.Unique_value_alert}_{i}"
|
||||
alert_page.delete_alert(session, base_url, user_email, user_password, org_id, stream_name, real_time_alert)
|
||||
|
||||
# Then delete destinations
|
||||
destination_name_webhook = f"destination_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.delete_destination(session, base_url, user_email, user_password, org_id, destination_name_webhook)
|
||||
|
||||
destination_name_email = f"destination_email_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.delete_destination(session, base_url, user_email, user_password, org_id, destination_name_email)
|
||||
|
||||
destination_name_pipeline = f"destination_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.delete_destination(session, base_url, user_email, user_password, org_id, destination_name_pipeline)
|
||||
|
||||
# Finally delete templates
|
||||
template_name_webhook = f"template_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.delete_template(session, base_url, user_email, user_password, org_id, template_name_webhook)
|
||||
|
||||
template_name_email = f"template_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.delete_template(session, base_url, user_email, user_password, org_id, template_name_email)
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
from pages.template_page import TemplatePage
|
||||
from pages.destination_page import DestinationPage
|
||||
from pages.user_page import UserPage
|
||||
from pages.alert_page import AlertPage
|
||||
import random
|
||||
|
||||
def edit_objects(session, base_url, user_email, user_password, org_id, stream_name, num_objects):
|
||||
"""Edit objects in the OpenObserve running instance."""
|
||||
template_page = TemplatePage(session, base_url, org_id)
|
||||
destination_page = DestinationPage(session, base_url, org_id)
|
||||
user_page = UserPage(session, base_url, org_id)
|
||||
alert_page = AlertPage(session, base_url, org_id)
|
||||
for i in range(num_objects):
|
||||
|
||||
|
||||
|
||||
# Update templates
|
||||
template_name_webhook = f"template_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.update_template_webhook(session, base_url, user_email, user_password, org_id, template_name_webhook)
|
||||
|
||||
template_name_email = f"template_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.update_template_email(session, base_url, user_email, user_password, org_id, template_name_email)
|
||||
|
||||
# Update destinations
|
||||
destination_name_webhook = f"destination_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.update_destination_webhook(session, base_url, user_email, user_password, org_id, template_name_webhook, destination_name_webhook)
|
||||
|
||||
destination_name_email = f"destination_email_{destination_page.Unique_value_destination}_{i}"
|
||||
email_address_admin = f"admin_{destination_page.Unique_value_destination}_{i}@{random.randint(100000, 999999)}.com"
|
||||
user_page.create_user_admin(session, base_url, user_email, user_password, org_id, email_address_admin)
|
||||
destination_page.update_destination_email(session, base_url, user_email, user_password, org_id, email_address_admin, template_name_email, destination_name_email)
|
||||
|
||||
destination_name_pipeline = f"destination_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.update_destination_pipeline(session, base_url, user_email, user_password, org_id, destination_name_pipeline)
|
||||
|
||||
# Update alerts
|
||||
alert_webhook = f"alert_webhook_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.update_standard_alert(session, base_url, user_email, user_password, org_id, stream_name, template_name_webhook, destination_name_webhook, alert_webhook)
|
||||
|
||||
alert_email = f"alert_email_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.update_standard_alert(session, base_url, user_email, user_password, org_id, stream_name, template_name_email, destination_name_email, alert_email)
|
||||
|
||||
alert_cron = f"alert_cron_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.update_standard_alert_cron(session, base_url, user_email, user_password, org_id, stream_name, template_name_email, destination_name_email, alert_cron)
|
||||
|
||||
alert_real_time = f"alert_real_time_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.update_real_time_alert(session, base_url, user_email, user_password, org_id, stream_name, template_name_webhook, destination_name_webhook, alert_real_time)
|
||||
|
||||
alert_sql = f"alert_sql_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.update_standard_alert_sql(session, base_url, user_email, user_password, org_id, stream_name, template_name_webhook, destination_name_webhook, alert_sql)
|
||||
|
||||
@@ -0,0 +1,764 @@
|
||||
import random
|
||||
import uuid
|
||||
from requests.auth import HTTPBasicAuth
|
||||
import time
|
||||
|
||||
class AlertPage:
|
||||
# Make Unique_value_alert a class variable
|
||||
Unique_value_alert = f"d4m21_{random.randint(100000, 999999)}"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_standard_alert(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"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": 3,
|
||||
"cron": "",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "minutes",
|
||||
"timezone": "UTC",
|
||||
"tolerance_in_secs": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test"
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs", json=payload)
|
||||
if response.status_code == 409:
|
||||
# Handle alert already exists scenario
|
||||
print(f"Alert already exists: {payload['name']}.")
|
||||
return False # Indicate that creation failed due to existing alert
|
||||
assert response.status_code == 200, f"Failed to create alert: {response.content}"
|
||||
return True # Indicate successful creation
|
||||
|
||||
def create_standard_alert_sql(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"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": f"select log from \"{stream_name}\" where log = 200",
|
||||
"promql": "",
|
||||
"type": "sql",
|
||||
"promql_condition": None,
|
||||
"vrl_function": None,
|
||||
"multi_time_range": []
|
||||
},
|
||||
"trigger_condition": {
|
||||
"period": 10,
|
||||
"operator": ">=",
|
||||
"frequency": 3,
|
||||
"cron": "",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "minutes",
|
||||
"timezone": "UTC",
|
||||
"tolerance_in_secs": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test"
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs", json=payload)
|
||||
if response.status_code == 409:
|
||||
# Handle alert already exists scenario
|
||||
print(f"Alert already exists: {payload['name']}.")
|
||||
return False
|
||||
assert response.status_code == 200, f"Failed to create alert: {response.content}"
|
||||
return True
|
||||
|
||||
def create_real_time_alert(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"stream_type": "logs",
|
||||
"stream_name": stream_name,
|
||||
"is_real_time": True,
|
||||
"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": 3,
|
||||
"cron": "",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "minutes",
|
||||
"timezone": "UTC",
|
||||
"tolerance_in_secs": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test"
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs", json=payload)
|
||||
if response.status_code == 409:
|
||||
# Handle alert already exists scenario
|
||||
print(f"Alert already exists: {payload['name']}.")
|
||||
return False # Indicate that creation failed due to existing alert
|
||||
assert response.status_code == 200, f"Failed to create alert: {response.content}"
|
||||
return True # Indicate successful creation
|
||||
|
||||
def create_standard_alert_cron(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"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": 3,
|
||||
"cron": "1 40 * * * *",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "cron",
|
||||
"timezone": "UTC",
|
||||
"tz_offset": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test"
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs", json=payload)
|
||||
if response.status_code == 409:
|
||||
# Handle alert already exists scenario
|
||||
print(f"Alert already exists: {payload['name']}.")
|
||||
return False # Indicate that creation failed due to existing alert
|
||||
assert response.status_code == 200, f"Failed to create alert: {response.content}"
|
||||
return True # Indicate successful creation
|
||||
|
||||
def retrieve_alerts_standard(self, session, base_url, user_email, user_password, org_id, stream_name):
|
||||
"""Retrieve standard alerts."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
# Parse the response
|
||||
response_data = response.json()
|
||||
|
||||
# Handle the case where response is wrapped in a 'data' field
|
||||
alerts = response_data.get('data', []) if isinstance(response_data, dict) else response_data
|
||||
|
||||
|
||||
|
||||
# Ensure alerts is a list
|
||||
if not isinstance(alerts, list):
|
||||
print(f"Unexpected alerts format: {alerts}")
|
||||
return {"count": 0, "alerts": []}
|
||||
|
||||
# Filter alerts
|
||||
standard_alerts = []
|
||||
for alert in alerts:
|
||||
try:
|
||||
if (isinstance(alert, dict) and
|
||||
isinstance(alert.get('name'), str) and
|
||||
alert['name'].startswith(f"standard_{AlertPage.Unique_value_alert}")):
|
||||
standard_alerts.append(alert)
|
||||
except (KeyError, TypeError) as e:
|
||||
print(f"Error processing alert {alert}: {str(e)}")
|
||||
continue
|
||||
|
||||
print(f"Filtered standard alerts: {standard_alerts}")
|
||||
|
||||
assert len(standard_alerts) > 0, f"No alerts found with title starting with 'standard_{AlertPage.Unique_value_alert}'"
|
||||
|
||||
return {
|
||||
"count": len(standard_alerts),
|
||||
"alerts": standard_alerts
|
||||
}
|
||||
|
||||
def retrieve_alert_standard(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Retrieve a standard alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alert: {response.content}"
|
||||
alert_standard_webhook = response.json() # Assuming the response is a single alert, not a list.
|
||||
|
||||
# Assert that the response is a valid template object
|
||||
assert isinstance(alert_standard_webhook, dict), "Response is not a valid alert object"
|
||||
|
||||
return {
|
||||
"name_alert_standard_webhook": alert_standard_webhook["name"]
|
||||
}
|
||||
|
||||
def retrieve_alerts_standard_sql(self, session, base_url, user_email, user_password, org_id, stream_name):
|
||||
"""Retrieve standard SQL alerts."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
# Parse the response
|
||||
response_data = response.json()
|
||||
|
||||
# Handle the case where response is wrapped in a 'data' field
|
||||
alerts = response_data.get('data', []) if isinstance(response_data, dict) else response_data
|
||||
|
||||
|
||||
# Ensure alerts is a list
|
||||
if not isinstance(alerts, list):
|
||||
# print(f"Unexpected alerts format: {alerts}")
|
||||
return {"count": 0, "alerts": []}
|
||||
|
||||
# Filter alerts
|
||||
standard_sql_alerts = []
|
||||
for alert in alerts:
|
||||
try:
|
||||
if (isinstance(alert, dict) and
|
||||
isinstance(alert.get('name'), str) and
|
||||
"alert_sql" in alert['name'] and
|
||||
alert['name'].startswith(f"alert_sql_{AlertPage.Unique_value_alert}")):
|
||||
standard_sql_alerts.append(alert)
|
||||
except (KeyError, TypeError) as e:
|
||||
print(f"Error processing alert {alert}: {str(e)}")
|
||||
continue
|
||||
|
||||
# Assert that the response is a valid alert object
|
||||
assert isinstance(standard_sql_alerts, list), "Response is not a valid alert object"
|
||||
|
||||
return {
|
||||
"count": len(standard_sql_alerts),
|
||||
"alerts": standard_sql_alerts
|
||||
}
|
||||
|
||||
def retrieve_alert_standard_sql(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Retrieve a standard SQL alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alert: {response.content}"
|
||||
alert_standard_sql = response.json() # Assuming the response is a single alert, not a list.
|
||||
|
||||
# Assert that the response is a valid alert object
|
||||
assert isinstance(alert_standard_sql, dict), "Response is not a valid alert object"
|
||||
|
||||
return {
|
||||
"name_alert_standard_sql": alert_standard_sql["name"]
|
||||
}
|
||||
|
||||
def retrieve_alerts_sql(self, session, base_url, user_email, user_password, org_id, stream_name):
|
||||
"""Retrieve all SQL alerts."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
# Parse the JSON response
|
||||
alerts = response.json()
|
||||
assert len(alerts) > 0, "No alerts found"
|
||||
|
||||
# Filter templates that contain 'template_webhook' and match the expected name
|
||||
sql_alerts = [alert for alert in alerts
|
||||
if "sql" in alert["name"]
|
||||
and alert["name"].startswith(f"alert_sql_{AlertPage.Unique_value_alert}")]
|
||||
|
||||
assert len(sql_alerts) > 0, f"No alerts found containing 'sql' with title 'alert_sql_{AlertPage.Unique_value_alert}'"
|
||||
|
||||
return {
|
||||
"count": len(sql_alerts),
|
||||
"alerts": sql_alerts
|
||||
}
|
||||
|
||||
def retrieve_alert_sql(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Retrieve a SQL alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alert: {response.content}"
|
||||
alert_sql = response.json() # Assuming the response is a single alert, not a list.
|
||||
|
||||
# Assert that the response is a valid alert object
|
||||
assert isinstance(alert_sql, dict), "Response is not a valid alert object"
|
||||
|
||||
return {
|
||||
"name_alert_sql": alert_sql["name"]
|
||||
}
|
||||
|
||||
def retrieve_alerts_real_time(self, session, base_url, user_email, user_password, org_id, stream_name):
|
||||
"""Retrieve real-time alerts."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
alerts = response.json()
|
||||
if not isinstance(alerts, list):
|
||||
alerts = alerts.get('data', []) if isinstance(alerts, dict) else []
|
||||
|
||||
real_time_alerts = [
|
||||
alert for alert in alerts
|
||||
if isinstance(alert, dict) and
|
||||
isinstance(alert.get('name'), str) and
|
||||
"real_time" in alert["name"] and
|
||||
alert["name"].startswith(f"alert_real_time _{AlertPage.Unique_value_alert}")
|
||||
]
|
||||
|
||||
return {
|
||||
"count": len(real_time_alerts),
|
||||
"alerts": real_time_alerts
|
||||
}
|
||||
|
||||
def retrieve_alert_real_time(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Retrieve a real-time alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alert: {response.content}"
|
||||
alert_real_time = response.json() # Assuming the response is a single alert, not a list.
|
||||
|
||||
# Assert that the response is a valid alert object
|
||||
assert isinstance(alert_real_time, dict), "Response is not a valid alert object"
|
||||
|
||||
return {
|
||||
"name_alert_real_time": alert_real_time["name"]
|
||||
}
|
||||
|
||||
def retrieve_alerts_cron(self, session, base_url, user_email, user_password, org_id, stream_name):
|
||||
"""Retrieve cron alerts."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
alerts = response.json()
|
||||
if not isinstance(alerts, list):
|
||||
alerts = alerts.get('data', []) if isinstance(alerts, dict) else []
|
||||
|
||||
cron_alerts = [
|
||||
alert for alert in alerts
|
||||
if isinstance(alert, dict) and
|
||||
isinstance(alert.get('name'), str) and
|
||||
"cron" in alert["name"] and
|
||||
alert["name"].startswith(f"alert_cron_{AlertPage.Unique_value_alert}")
|
||||
]
|
||||
|
||||
return {
|
||||
"count": len(cron_alerts),
|
||||
"alerts": cron_alerts
|
||||
}
|
||||
|
||||
def retrieve_alert_cron(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Retrieve a cron alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alert: {response.content}"
|
||||
alert_cron = response.json() # Assuming the response is a single alert, not a list.
|
||||
|
||||
# Assert that the response is a valid alert object
|
||||
assert isinstance(alert_cron, dict), "Response is not a valid alert object"
|
||||
|
||||
return {
|
||||
"name_alert_cron": alert_cron["name"]
|
||||
}
|
||||
|
||||
|
||||
def update_standard_alert(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
"""Update a standard alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
# Update the alert with the new payload
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"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": 3,
|
||||
"cron": "",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "minutes",
|
||||
"timezone": "UTC",
|
||||
"tolerance_in_secs": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test updated"
|
||||
}
|
||||
|
||||
|
||||
response = session.put(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs", json=payload)
|
||||
assert response.status_code == 200, f"Failed to update alert: {response.content}"
|
||||
return response
|
||||
|
||||
def update_standard_alert_sql(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
"""Update a standard SQL alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
# Update the alert with the new payload
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"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": f"select log from \"{stream_name}\" where log = 200",
|
||||
"promql": "",
|
||||
"type": "sql",
|
||||
"promql_condition": None,
|
||||
"vrl_function": None,
|
||||
"multi_time_range": []
|
||||
},
|
||||
"trigger_condition": {
|
||||
"period": 10,
|
||||
"operator": ">=",
|
||||
"frequency": 3,
|
||||
"cron": "",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "minutes",
|
||||
"timezone": "UTC",
|
||||
"tolerance_in_secs": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test updated"
|
||||
}
|
||||
|
||||
response = session.put(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs", json=payload)
|
||||
assert response.status_code == 200, f"Failed to update alert: {response.content}"
|
||||
return response
|
||||
|
||||
def update_real_time_alert(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
"""Update a real-time alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
# Update the alert with the new payload
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"stream_type": "logs",
|
||||
"stream_name": stream_name,
|
||||
"is_real_time": True,
|
||||
"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": 3,
|
||||
"cron": "",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "minutes",
|
||||
"timezone": "UTC",
|
||||
"tolerance_in_secs": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test updated"
|
||||
}
|
||||
|
||||
response = session.put(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs", json=payload)
|
||||
assert response.status_code == 200, f"Failed to update alert: {response.content}"
|
||||
return response
|
||||
|
||||
def update_standard_alert_cron(self, session, base_url, user_email, user_password, org_id, stream_name, template_name, destination_name, alert_name):
|
||||
"""Update a standard cron alert."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
# Update the alert with the new payload
|
||||
payload = {
|
||||
"name": alert_name,
|
||||
"row_template": template_name,
|
||||
"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": 3,
|
||||
"cron": "1 40 * * * *",
|
||||
"threshold": 3,
|
||||
"silence": 10,
|
||||
"frequency_type": "cron",
|
||||
"timezone": "UTC",
|
||||
"tz_offset": 0
|
||||
},
|
||||
"org_id": org_id,
|
||||
"destinations": [destination_name],
|
||||
"enabled": True,
|
||||
"description": "test updated"
|
||||
}
|
||||
|
||||
response = session.put(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}?type=logs", json=payload)
|
||||
assert response.status_code == 200, f"Failed to update alert: {response.content}"
|
||||
return response
|
||||
|
||||
|
||||
|
||||
def delete_alert(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Delete alerts of a specified type."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
resp_delete_alert = session.delete(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}")
|
||||
assert resp_delete_alert.status_code == 200, f"Failed to delete alert {alert_name}: {resp_delete_alert.content}"
|
||||
|
||||
# Wait for a few seconds to allow the data to be deleted
|
||||
time.sleep(10) # Increase this time if necessary
|
||||
|
||||
# Verify deletion
|
||||
resp_ver_alert = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts/{alert_name}")
|
||||
assert resp_ver_alert.status_code == 404, f"Expected 404 for {alert_name}, but got {resp_ver_alert.status_code}"
|
||||
|
||||
return resp_delete_alert # Return the list of deleted alert names for verification if needed
|
||||
|
||||
def validate_deleted_standard_alert(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Running an E2E test for validating deleted alerts in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all alerts
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts?type=logs")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
# Get the response data
|
||||
response_data = response.json()
|
||||
|
||||
|
||||
# Handle different response formats
|
||||
if isinstance(response_data, dict):
|
||||
alerts = response_data.get('data', [])
|
||||
elif isinstance(response_data, str):
|
||||
# print(f"Unexpected string response: {response_data}")
|
||||
alerts = []
|
||||
elif isinstance(response_data, list):
|
||||
alerts = response_data
|
||||
else:
|
||||
print(f"Unexpected response type: {type(response_data)}")
|
||||
alerts = []
|
||||
|
||||
|
||||
# Filter alerts with proper type checking
|
||||
standard_alerts = []
|
||||
for alert in alerts:
|
||||
if not isinstance(alert, dict):
|
||||
# print(f"Skipping non-dict alert: {alert}")
|
||||
continue
|
||||
|
||||
name = alert.get('name')
|
||||
if not isinstance(name, str):
|
||||
# print(f"Skipping alert with non-string name: {alert}")
|
||||
continue
|
||||
|
||||
if ("standard" in name and
|
||||
name.startswith(f"standard_{AlertPage.Unique_value_alert}")):
|
||||
standard_alerts.append(alert)
|
||||
|
||||
|
||||
# Check if any alerts of the specified type exist
|
||||
if len(standard_alerts) == 0:
|
||||
print(f"No alerts found of type '{alert_name}' (expected for deletion verification)")
|
||||
else:
|
||||
print(f"Alerts found of type '{alert_name}': {[alert.get('name') for alert in standard_alerts]}")
|
||||
|
||||
return standard_alerts
|
||||
|
||||
def validate_deleted_standard_alert_sql(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Running an E2E test for validating deleted templates in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all templates
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts")
|
||||
assert response.status_code == 200, f"Failed to retrieve templates: {response.content}"
|
||||
|
||||
alerts = response.json()
|
||||
|
||||
# Filter templates that contain 'template_email' and match the expected name
|
||||
standard_sql_alerts = [alert for alert in alerts
|
||||
if "standard_sql" in alert["name"]
|
||||
and alert["name"].startswith(f"standard_sql_{AlertPage.Unique_value_alert}")]
|
||||
|
||||
# Check if any templates of the specified type exist
|
||||
if len(standard_sql_alerts) == 0:
|
||||
print(f"No alerts found of type '{alert_name}'")
|
||||
# Optionally, you can choose to skip the assertion or handle it differently
|
||||
else:
|
||||
assert len(standard_sql_alerts) > 0, f"No alerts found of type '{alert_name}'"
|
||||
# print(f"Alerts found of type '{alert_name}': {[alert['name'] for alert in standard_sql_alerts]}")
|
||||
|
||||
return standard_sql_alerts
|
||||
|
||||
def validate_deleted_real_time_alert(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Running an E2E test for validating deleted alerts in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all alerts
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
alerts = response.json()
|
||||
|
||||
# Filter alerts that contain 'real_time' and match the expected name
|
||||
real_time_alerts = [alert for alert in alerts
|
||||
if "real_time" in alert["name"]
|
||||
and alert["name"].startswith(f"real_time_{AlertPage.Unique_value_alert}")]
|
||||
|
||||
if len(real_time_alerts) == 0:
|
||||
print(f"No alerts found of type '{alert_name}'")
|
||||
else:
|
||||
print(f"Alerts found of type '{alert_name}': {[alert['name'] for alert in real_time_alerts]}")
|
||||
return real_time_alerts
|
||||
|
||||
def validate_deleted_standard_alert_cron(self, session, base_url, user_email, user_password, org_id, stream_name, alert_name):
|
||||
"""Running an E2E test for validating deleted alerts in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all alerts
|
||||
response = session.get(f"{base_url}api/{org_id}/{stream_name}/alerts")
|
||||
assert response.status_code == 200, f"Failed to retrieve alerts: {response.content}"
|
||||
|
||||
alerts = response.json()
|
||||
|
||||
# Filter alerts that contain 'cron' and match the expected name
|
||||
standard_cron_alerts = [alert for alert in alerts
|
||||
if "standard_cron" in alert["name"]
|
||||
and alert["name"].startswith(f"standard_cron_{AlertPage.Unique_value_alert}")]
|
||||
|
||||
if len(standard_cron_alerts) == 0:
|
||||
print(f"No alerts found of type '{alert_name}'")
|
||||
else:
|
||||
print(f"Alerts found of type '{alert_name}': {[alert['name'] for alert in standard_cron_alerts]}")
|
||||
return standard_cron_alerts
|
||||
|
||||
@@ -0,0 +1,683 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
class DashboardPage:
|
||||
# Make Unique_value_destination a class variable
|
||||
Unique_value_dashboard = f"d3m21_{random.randint(100000, 999999)}"
|
||||
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
|
||||
|
||||
def create_dashboard(self, session, base_url, user_email, user_password, org_id, stream_name, folder_id, dashboard_name):
|
||||
"""Create a dashboard."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Custom-Header": "value"
|
||||
}
|
||||
|
||||
payload = {
|
||||
"version": 5,
|
||||
"title": dashboard_name,
|
||||
"description": "",
|
||||
"role": "",
|
||||
"owner": user_email, # Set to the user's email
|
||||
"tabs": [
|
||||
{
|
||||
"tabId": "default",
|
||||
"name": "Default",
|
||||
"panels": [
|
||||
{
|
||||
"id": "Panel_ID9034710",
|
||||
"type": "bar",
|
||||
"title": "P1",
|
||||
"description": "",
|
||||
"config": {
|
||||
"show_legends": True,
|
||||
"legends_position": None,
|
||||
"decimals": 2,
|
||||
"line_thickness": 1.5,
|
||||
"step_value": "0",
|
||||
"top_results_others": False,
|
||||
"axis_border_show": False,
|
||||
"label_option": {
|
||||
"rotate": 0
|
||||
},
|
||||
"show_symbol": True,
|
||||
"line_interpolation": "smooth",
|
||||
"legend_width": {
|
||||
"unit": "px"
|
||||
},
|
||||
"base_map": {
|
||||
"type": "osm"
|
||||
},
|
||||
"map_type": {
|
||||
"type": "world"
|
||||
},
|
||||
"map_view": {
|
||||
"zoom": 1,
|
||||
"lat": 0,
|
||||
"lng": 0
|
||||
},
|
||||
"map_symbol_style": {
|
||||
"size": "by Value",
|
||||
"size_by_value": {
|
||||
"min": 1,
|
||||
"max": 100
|
||||
},
|
||||
"size_fixed": 2
|
||||
},
|
||||
"drilldown": [],
|
||||
"mark_line": [],
|
||||
"override_config": [],
|
||||
"connect_nulls": False,
|
||||
"no_value_replacement": "",
|
||||
"wrap_table_cells": False,
|
||||
"table_transpose": False,
|
||||
"table_dynamic_columns": False,
|
||||
"color": {
|
||||
"mode": "palette-classic-by-series",
|
||||
"fixedColor": [
|
||||
"#53ca53"
|
||||
],
|
||||
"seriesBy": "last"
|
||||
},
|
||||
"trellis": {
|
||||
"layout": None,
|
||||
"num_of_columns": 1
|
||||
}
|
||||
},
|
||||
"queryType": "sql",
|
||||
"queries": [
|
||||
{
|
||||
"query": f"SELECT histogram(_timestamp) as \"x_axis_1\", count(kubernetes_namespace_name) as \"y_axis_1\", kubernetes_labels_name as \"breakdown_1\" FROM \"{stream_name}\" GROUP BY x_axis_1, breakdown_1 ORDER BY x_axis_1 ASC",
|
||||
"vrlFunctionQuery": "",
|
||||
"customQuery": False,
|
||||
"fields": {
|
||||
"stream": stream_name,
|
||||
"stream_type": "logs",
|
||||
"x": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "x_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": None,
|
||||
"aggregationFunction": "histogram",
|
||||
"sortBy": "ASC",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"y": [
|
||||
{
|
||||
"label": "Kubernetes Namespace Name",
|
||||
"alias": "y_axis_1",
|
||||
"column": "kubernetes_namespace_name",
|
||||
"color": "#5960b2",
|
||||
"aggregationFunction": "count",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"z": [],
|
||||
"breakdown": [
|
||||
{
|
||||
"label": "Kubernetes Labels Name",
|
||||
"alias": "breakdown_1",
|
||||
"column": "kubernetes_labels_name",
|
||||
"color": None,
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"filter": {
|
||||
"filterType": "group",
|
||||
"logicalOperator": "AND",
|
||||
"conditions": []
|
||||
}
|
||||
},
|
||||
"config": {
|
||||
"promql_legend": "",
|
||||
"layer_type": "scatter",
|
||||
"weight_fixed": 1,
|
||||
"limit": 0,
|
||||
"min": 0,
|
||||
"max": 100,
|
||||
"time_shift": []
|
||||
}
|
||||
}
|
||||
],
|
||||
"layout": {
|
||||
"x": 0,
|
||||
"y": 0,
|
||||
"w": 24,
|
||||
"h": 9,
|
||||
"i": 1,
|
||||
"moved": False
|
||||
},
|
||||
"htmlContent": "",
|
||||
"markdownContent": "",
|
||||
"customChartContent": " // To know more about ECharts , \n// visit: https://echarts.apache.org/examples/en/index.html \n// Example: https://echarts.apache.org/examples/en/editor.html?c=line-simple \n// Define your ECharts 'option' here. \n// 'data' variable is available for use and contains the response data from the search result and it is an array.\noption = { \n \n};\n "
|
||||
},
|
||||
{
|
||||
"id": "Panel_ID8353510",
|
||||
"type": "h-bar",
|
||||
"title": "P4",
|
||||
"description": "",
|
||||
"config": {
|
||||
"show_legends": True,
|
||||
"legends_position": None,
|
||||
"decimals": 2,
|
||||
"line_thickness": 1.5,
|
||||
"step_value": "0",
|
||||
"top_results_others": False,
|
||||
"axis_border_show": False,
|
||||
"label_option": {
|
||||
"rotate": 0
|
||||
},
|
||||
"show_symbol": True,
|
||||
"line_interpolation": "smooth",
|
||||
"legend_width": {
|
||||
"unit": "px"
|
||||
},
|
||||
"base_map": {
|
||||
"type": "osm"
|
||||
},
|
||||
"map_type": {
|
||||
"type": "world"
|
||||
},
|
||||
"map_view": {
|
||||
"zoom": 1,
|
||||
"lat": 0,
|
||||
"lng": 0
|
||||
},
|
||||
"map_symbol_style": {
|
||||
"size": "by Value",
|
||||
"size_by_value": {
|
||||
"min": 1,
|
||||
"max": 100
|
||||
},
|
||||
"size_fixed": 2
|
||||
},
|
||||
"drilldown": [],
|
||||
"mark_line": [],
|
||||
"override_config": [],
|
||||
"connect_nulls": False,
|
||||
"no_value_replacement": "",
|
||||
"wrap_table_cells": False,
|
||||
"table_transpose": False,
|
||||
"table_dynamic_columns": False,
|
||||
"color": {
|
||||
"mode": "palette-classic-by-series",
|
||||
"fixedColor": [
|
||||
"#53ca53"
|
||||
],
|
||||
"seriesBy": "last"
|
||||
},
|
||||
"trellis": {
|
||||
"layout": None,
|
||||
"num_of_columns": 1
|
||||
}
|
||||
},
|
||||
"queryType": "sql",
|
||||
"queries": [
|
||||
{
|
||||
"query": f"SELECT histogram(_timestamp) as \"x_axis_1\", count(kubernetes_namespace_name) as \"y_axis_1\", kubernetes_labels_name as \"breakdown_1\" FROM \"{stream_name}\" GROUP BY x_axis_1, breakdown_1 ORDER BY x_axis_1 ASC",
|
||||
"vrlFunctionQuery": "",
|
||||
"customQuery": False,
|
||||
"fields": {
|
||||
"stream": stream_name,
|
||||
"stream_type": "logs",
|
||||
"x": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "x_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": None,
|
||||
"aggregationFunction": "histogram",
|
||||
"sortBy": "ASC",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"y": [
|
||||
{
|
||||
"label": "Kubernetes Namespace Name",
|
||||
"alias": "y_axis_1",
|
||||
"column": "kubernetes_namespace_name",
|
||||
"color": "#5960b2",
|
||||
"aggregationFunction": "count",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"z": [],
|
||||
"breakdown": [
|
||||
{
|
||||
"label": "Kubernetes Labels Name",
|
||||
"alias": "breakdown_1",
|
||||
"column": "kubernetes_labels_name",
|
||||
"color": None,
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"filter": {
|
||||
"filterType": "group",
|
||||
"logicalOperator": "AND",
|
||||
"conditions": []
|
||||
}
|
||||
},
|
||||
"config": {
|
||||
"promql_legend": "",
|
||||
"layer_type": "scatter",
|
||||
"weight_fixed": 1,
|
||||
"limit": 0,
|
||||
"min": 0,
|
||||
"max": 100,
|
||||
"time_shift": []
|
||||
}
|
||||
}
|
||||
],
|
||||
"layout": {
|
||||
"x": 24,
|
||||
"y": 0,
|
||||
"w": 24,
|
||||
"h": 9,
|
||||
"i": 2,
|
||||
"moved": False
|
||||
},
|
||||
"htmlContent": "",
|
||||
"markdownContent": "",
|
||||
"customChartContent": " // To know more about ECharts , \n// visit: https://echarts.apache.org/examples/en/index.html \n// Example: https://echarts.apache.org/examples/en/editor.html?c=line-simple \n// Define your ECharts 'option' here. \n// 'data' variable is available for use and contains the response data from the search result and it is an array.\noption = { \n \n};\n "
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"tabId": "39193",
|
||||
"name": "count",
|
||||
"panels": [
|
||||
{
|
||||
"id": "Panel_ID1546910",
|
||||
"type": "bar",
|
||||
"title": "count",
|
||||
"description": "",
|
||||
"config": {
|
||||
"show_legends": True,
|
||||
"legends_position": None,
|
||||
"decimals": 2,
|
||||
"line_thickness": 1.5,
|
||||
"step_value": "0",
|
||||
"top_results_others": False,
|
||||
"axis_border_show": False,
|
||||
"label_option": {
|
||||
"rotate": 0
|
||||
},
|
||||
"show_symbol": True,
|
||||
"line_interpolation": "smooth",
|
||||
"legend_width": {
|
||||
"unit": "px"
|
||||
},
|
||||
"base_map": {
|
||||
"type": "osm"
|
||||
},
|
||||
"map_type": {
|
||||
"type": "world"
|
||||
},
|
||||
"map_view": {
|
||||
"zoom": 1,
|
||||
"lat": 0,
|
||||
"lng": 0
|
||||
},
|
||||
"map_symbol_style": {
|
||||
"size": "by Value",
|
||||
"size_by_value": {
|
||||
"min": 1,
|
||||
"max": 100
|
||||
},
|
||||
"size_fixed": 2
|
||||
},
|
||||
"drilldown": [],
|
||||
"mark_line": [],
|
||||
"override_config": [],
|
||||
"connect_nulls": False,
|
||||
"no_value_replacement": "",
|
||||
"wrap_table_cells": False,
|
||||
"table_transpose": False,
|
||||
"table_dynamic_columns": False,
|
||||
"color": {
|
||||
"mode": "palette-classic-by-series",
|
||||
"fixedColor": [
|
||||
"#53ca53"
|
||||
],
|
||||
"seriesBy": "last"
|
||||
},
|
||||
"trellis": {
|
||||
"layout": None,
|
||||
"num_of_columns": 1
|
||||
}
|
||||
},
|
||||
"queryType": "sql",
|
||||
"queries": [
|
||||
{
|
||||
"query": f"SELECT histogram(_timestamp) as \"x_axis_1\", count(_timestamp) as \"y_axis_1\" FROM \"{stream_name}\" GROUP BY x_axis_1 ORDER BY x_axis_1 ASC",
|
||||
"vrlFunctionQuery": "",
|
||||
"customQuery": False,
|
||||
"fields": {
|
||||
"stream": stream_name,
|
||||
"stream_type": "logs",
|
||||
"x": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "x_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": None,
|
||||
"aggregationFunction": "histogram",
|
||||
"sortBy": "ASC",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"y": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "y_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": "#5960b2",
|
||||
"aggregationFunction": "count",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"z": [],
|
||||
"breakdown": [],
|
||||
"filter": {
|
||||
"filterType": "group",
|
||||
"logicalOperator": "AND",
|
||||
"conditions": []
|
||||
}
|
||||
},
|
||||
"config": {
|
||||
"promql_legend": "",
|
||||
"layer_type": "scatter",
|
||||
"weight_fixed": 1,
|
||||
"limit": 0,
|
||||
"min": 0,
|
||||
"max": 100,
|
||||
"time_shift": []
|
||||
}
|
||||
}
|
||||
],
|
||||
"layout": {
|
||||
"x": 0,
|
||||
"y": 0,
|
||||
"w": 24,
|
||||
"h": 9,
|
||||
"i": 1,
|
||||
"moved": False
|
||||
},
|
||||
"htmlContent": "",
|
||||
"markdownContent": "",
|
||||
"customChartContent": " // To know more about ECharts , \n// visit: https://echarts.apache.org/examples/en/index.html \n// Example: https://echarts.apache.org/examples/en/editor.html?c=line-simple \n// Define your ECharts 'option' here. \n// 'data' variable is available for use and contains the response data from the search result and it is an array.\noption = { \n \n};\n "
|
||||
},
|
||||
{
|
||||
"id": "Panel_ID1055010",
|
||||
"type": "h-bar",
|
||||
"title": "C1",
|
||||
"description": "",
|
||||
"config": {
|
||||
"show_legends": True,
|
||||
"legends_position": None,
|
||||
"decimals": 2,
|
||||
"line_thickness": 1.5,
|
||||
"step_value": "0",
|
||||
"top_results_others": False,
|
||||
"axis_border_show": False,
|
||||
"label_option": {
|
||||
"rotate": 0
|
||||
},
|
||||
"show_symbol": True,
|
||||
"line_interpolation": "smooth",
|
||||
"legend_width": {
|
||||
"unit": "px"
|
||||
},
|
||||
"base_map": {
|
||||
"type": "osm"
|
||||
},
|
||||
"map_type": {
|
||||
"type": "world"
|
||||
},
|
||||
"map_view": {
|
||||
"zoom": 1,
|
||||
"lat": 0,
|
||||
"lng": 0
|
||||
},
|
||||
"map_symbol_style": {
|
||||
"size": "by Value",
|
||||
"size_by_value": {
|
||||
"min": 1,
|
||||
"max": 100
|
||||
},
|
||||
"size_fixed": 2
|
||||
},
|
||||
"drilldown": [],
|
||||
"mark_line": [],
|
||||
"override_config": [],
|
||||
"connect_nulls": False,
|
||||
"no_value_replacement": "",
|
||||
"wrap_table_cells": False,
|
||||
"table_transpose": False,
|
||||
"table_dynamic_columns": False,
|
||||
"color": {
|
||||
"mode": "palette-classic-by-series",
|
||||
"fixedColor": [
|
||||
"#53ca53"
|
||||
],
|
||||
"seriesBy": "last"
|
||||
},
|
||||
"trellis": {
|
||||
"layout": None,
|
||||
"num_of_columns": 1
|
||||
}
|
||||
},
|
||||
"queryType": "sql",
|
||||
"queries": [
|
||||
{
|
||||
"query": f"SELECT histogram(_timestamp) as \"x_axis_1\", count(_timestamp) as \"y_axis_1\" FROM \"{stream_name}\" GROUP BY x_axis_1 ORDER BY x_axis_1 ASC",
|
||||
"vrlFunctionQuery": "",
|
||||
"customQuery": False,
|
||||
"fields": {
|
||||
"stream": stream_name,
|
||||
"stream_type": "logs",
|
||||
"x": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "x_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": None,
|
||||
"aggregationFunction": "histogram",
|
||||
"sortBy": "ASC",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"y": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "y_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": "#5960b2",
|
||||
"aggregationFunction": "count",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"z": [],
|
||||
"breakdown": [],
|
||||
"filter": {
|
||||
"filterType": "group",
|
||||
"logicalOperator": "AND",
|
||||
"conditions": []
|
||||
}
|
||||
},
|
||||
"config": {
|
||||
"promql_legend": "",
|
||||
"layer_type": "scatter",
|
||||
"weight_fixed": 1,
|
||||
"limit": 0,
|
||||
"min": 0,
|
||||
"max": 100,
|
||||
"time_shift": []
|
||||
}
|
||||
}
|
||||
],
|
||||
"layout": {
|
||||
"x": 24,
|
||||
"y": 0,
|
||||
"w": 24,
|
||||
"h": 9,
|
||||
"i": 2,
|
||||
"moved": False
|
||||
},
|
||||
"htmlContent": "",
|
||||
"markdownContent": "",
|
||||
"customChartContent": " // To know more about ECharts , \n// visit: https://echarts.apache.org/examples/en/index.html \n// Example: https://echarts.apache.org/examples/en/editor.html?c=line-simple \n// Define your ECharts 'option' here. \n// 'data' variable is available for use and contains the response data from the search result and it is an array.\noption = { \n \n};\n "
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"tabId": "34948",
|
||||
"name": "Tes",
|
||||
"panels": [
|
||||
{
|
||||
"id": "Panel_ID5902410",
|
||||
"type": "bar",
|
||||
"title": "Test2",
|
||||
"description": "",
|
||||
"config": {
|
||||
"show_legends": True,
|
||||
"legends_position": None,
|
||||
"decimals": 2,
|
||||
"line_thickness": 1.5,
|
||||
"step_value": "0",
|
||||
"top_results_others": False,
|
||||
"axis_border_show": False,
|
||||
"label_option": {
|
||||
"rotate": 0
|
||||
},
|
||||
"show_symbol": True,
|
||||
"line_interpolation": "smooth",
|
||||
"legend_width": {
|
||||
"unit": "px"
|
||||
},
|
||||
"base_map": {
|
||||
"type": "osm"
|
||||
},
|
||||
"map_type": {
|
||||
"type": "world"
|
||||
},
|
||||
"map_view": {
|
||||
"zoom": 1,
|
||||
"lat": 0,
|
||||
"lng": 0
|
||||
},
|
||||
"map_symbol_style": {
|
||||
"size": "by Value",
|
||||
"size_by_value": {
|
||||
"min": 1,
|
||||
"max": 100
|
||||
},
|
||||
"size_fixed": 2
|
||||
},
|
||||
"drilldown": [],
|
||||
"mark_line": [],
|
||||
"override_config": [],
|
||||
"connect_nulls": False,
|
||||
"no_value_replacement": "",
|
||||
"wrap_table_cells": False,
|
||||
"table_transpose": False,
|
||||
"table_dynamic_columns": False,
|
||||
"color": {
|
||||
"mode": "palette-classic-by-series",
|
||||
"fixedColor": [
|
||||
"#53ca53"
|
||||
],
|
||||
"seriesBy": "last"
|
||||
},
|
||||
"trellis": {
|
||||
"layout": None,
|
||||
"num_of_columns": 1
|
||||
}
|
||||
},
|
||||
"queryType": "sql",
|
||||
"queries": [
|
||||
{
|
||||
"query": f"SELECT histogram(_timestamp) as \"x_axis_1\", count(distinct(_timestamp)) as \"y_axis_1\" FROM \"{stream_name}\" GROUP BY x_axis_1 ORDER BY x_axis_1 ASC",
|
||||
"vrlFunctionQuery": "",
|
||||
"customQuery": False,
|
||||
"fields": {
|
||||
"stream": stream_name,
|
||||
"stream_type": "logs",
|
||||
"x": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "x_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": None,
|
||||
"aggregationFunction": "histogram",
|
||||
"sortBy": "ASC",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"y": [
|
||||
{
|
||||
"label": "Timestamp",
|
||||
"alias": "y_axis_1",
|
||||
"column": "_timestamp",
|
||||
"color": "#5960b2",
|
||||
"aggregationFunction": "count-distinct",
|
||||
"isDerived": False
|
||||
}
|
||||
],
|
||||
"z": [],
|
||||
"breakdown": [],
|
||||
"filter": {
|
||||
"filterType": "group",
|
||||
"logicalOperator": "AND",
|
||||
"conditions": []
|
||||
}
|
||||
},
|
||||
"config": {
|
||||
"promql_legend": "",
|
||||
"layer_type": "scatter",
|
||||
"weight_fixed": 1,
|
||||
"limit": 0,
|
||||
"min": 0,
|
||||
"max": 100,
|
||||
"time_shift": []
|
||||
}
|
||||
}
|
||||
],
|
||||
"layout": {
|
||||
"x": 0,
|
||||
"y": 0,
|
||||
"w": 24,
|
||||
"h": 9,
|
||||
"i": 1,
|
||||
"moved": False
|
||||
},
|
||||
"htmlContent": "",
|
||||
"markdownContent": "",
|
||||
"customChartContent": " // To know more about ECharts , \n// visit: https://echarts.apache.org/examples/en/index.html \n// Example: https://echarts.apache.org/examples/en/editor.html?c=line-simple \n// Define your ECharts 'option' here. \n// 'data' variable is available for use and contains the response data from the search result and it is an array.\noption = { \n \n};\n "
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"variables": {
|
||||
"list": [],
|
||||
"showDynamicFilters": True
|
||||
},
|
||||
"defaultDatetimeDuration": {
|
||||
"type": "relative",
|
||||
"relativeTimePeriod": "45m"
|
||||
}
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/dashboards?folder={folder_id}", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create panel: {response.content.decode()}"
|
||||
dashboard_id = response.json()["v5"]["dashboardId"]
|
||||
return dashboard_id
|
||||
@@ -0,0 +1,341 @@
|
||||
import random
|
||||
import time
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
|
||||
class DestinationPage:
|
||||
# Make Unique_value_destination a class variable
|
||||
Unique_value_destination = f"d4m21_{random.randint(100000, 999999)}"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_destination_webhook(self, session, base_url, org_id, user_email, user_password, template_name, destination_name):
|
||||
"""Create a webhook destination."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
skip_tls_verify_value = False
|
||||
|
||||
payload = {
|
||||
"url": "https://jsonplaceholder.typicode.com/todos",
|
||||
"method": "get",
|
||||
"template": template_name,
|
||||
"headers": {},
|
||||
"name": destination_name,
|
||||
"skip_tls_verify": skip_tls_verify_value
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/alerts/destinations", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create destination: {response.content}"
|
||||
return response
|
||||
|
||||
def create_destination_email(self, session, base_url, org_id, user_email, user_password, email_address_admin, template_email, destination_name):
|
||||
"""Create a email destination."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
skip_tls_verify_value = False
|
||||
payload = {
|
||||
"url": "",
|
||||
"type": "email",
|
||||
"emails": [email_address_admin],
|
||||
"method": "post",
|
||||
"template": template_email,
|
||||
"headers": {},
|
||||
"name": destination_name,
|
||||
"skip_tls_verify": skip_tls_verify_value
|
||||
}
|
||||
response = session.post(f"{base_url}api/{org_id}/alerts/destinations", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create destination: {response.content}"
|
||||
return response
|
||||
|
||||
def create_destination_pipeline(self, session, base_url, org_id, user_email, user_password, destination_name):
|
||||
"""Create a pipeline destination."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
skip_tls_verify_value = False # Define the skip_tls_verify_value
|
||||
payload = {
|
||||
"url": "https://jsonplaceholder.typicode.com/todos",
|
||||
"method": "get",
|
||||
"template": "",
|
||||
"headers": {}, # Ensure this is formatted correctly
|
||||
"name": destination_name,
|
||||
"skip_tls_verify": skip_tls_verify_value
|
||||
}
|
||||
response = session.post(f"{base_url}api/{org_id}/alerts/destinations", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create destination: {response.content}"
|
||||
return response
|
||||
|
||||
def retrieve_destinations_webhook(self, session, base_url, user_email, user_password, org_id):
|
||||
"""Retrieve a webhook destination."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations")
|
||||
assert response.status_code == 200, f"Failed to validate destination: {response.content}"
|
||||
destinations = response.json()
|
||||
assert len(destinations) > 0, "No destinations found"
|
||||
|
||||
webhook_destinations = [destination for destination in destinations if "destination_webhook" in destination["name"]
|
||||
and destination["name"].startswith(f"destination_webhook_{self.Unique_value_destination}")]
|
||||
|
||||
assert len(webhook_destinations) > 0, f"No destinations found containing 'destination_webhook' with title 'destination_webhook_{self.Unique_value_destination}'"
|
||||
return {
|
||||
"count": len(webhook_destinations),
|
||||
"destinations": webhook_destinations
|
||||
}
|
||||
|
||||
|
||||
def retrieve_destination_webhook(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Retrieve the destination webhook and assert the body."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
||||
assert response.status_code == 200, f"Failed to validate destination: {response.content}"
|
||||
|
||||
destination_webhook = response.json() # Assuming the response is a single template, not a list.
|
||||
|
||||
# Assert that the response is a valid template object
|
||||
assert isinstance(destination_webhook, dict), "Response is not a valid destination object"
|
||||
|
||||
return {
|
||||
"name_destination_webhook": destination_webhook["name"]
|
||||
}
|
||||
|
||||
def retrieve_destinations_email(self, session, base_url, user_email, user_password, org_id):
|
||||
"""Retrieve destinations and extract the first destination name containing 'destination_email' with title 'email_title'."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations")
|
||||
assert response.status_code == 200, f"Failed to validate destinations: {response.content}"
|
||||
|
||||
destinations = response.json()
|
||||
assert len(destinations) > 0, "No destinations found"
|
||||
|
||||
|
||||
email_destinations = [
|
||||
destination for destination in destinations
|
||||
if destination["name"].startswith(f"destination_email_{DestinationPage.Unique_value_destination}")
|
||||
]
|
||||
|
||||
# Assert that there are templates matching the criteria
|
||||
assert len(email_destinations) > 0, f"No destinations found containing 'destination_email' with title 'destination_email_{DestinationPage.Unique_value_destination}'"
|
||||
|
||||
# Return structured data
|
||||
return {
|
||||
"count": len(email_destinations),
|
||||
"destinations": email_destinations
|
||||
}
|
||||
|
||||
def retrieve_destination_email(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Retrieve the destination email and assert the title."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
||||
assert response.status_code == 200, f"Failed to validate destination: {response.content}"
|
||||
|
||||
destination_email = response.json() # Assuming the response is a single template, not a list.
|
||||
assert isinstance(destination_email, dict), "Response is not a valid destination object"
|
||||
|
||||
|
||||
return {
|
||||
"name_destination_email": destination_email["name"]
|
||||
}
|
||||
|
||||
def retrieve_destinations_pipeline(self, session, base_url, user_email, user_password, org_id):
|
||||
"""Retrieve destinations and extract the first destination name containing 'destination_pipeline' with title 'pipeline_title'."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations")
|
||||
assert response.status_code == 200, f"Failed to validate destinations: {response.content}"
|
||||
|
||||
destinations = response.json()
|
||||
assert len(destinations) > 0, "No destinations found"
|
||||
|
||||
pipeline_destinations = [destination for destination in destinations if "destination_pipeline" in destination["name"]
|
||||
and destination["name"].startswith(f"destination_pipeline_{DestinationPage.Unique_value_destination}")]
|
||||
|
||||
assert len(pipeline_destinations) > 0, f"No destinations found containing 'destination_pipeline' with title 'destination_pipeline_{DestinationPage.Unique_value_destination}'"
|
||||
|
||||
return {
|
||||
"count": len(pipeline_destinations),
|
||||
"destinations": pipeline_destinations
|
||||
}
|
||||
|
||||
def retrieve_destination_pipeline(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Retrieve the destination pipeline and assert the title."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
||||
assert response.status_code == 200, f"Failed to validate destination: {response.content}"
|
||||
|
||||
destination_pipeline = response.json() # Assuming the response is a single template, not a list.
|
||||
assert isinstance(destination_pipeline, dict), "Response is not a valid destination object"
|
||||
|
||||
return {
|
||||
"name_destination_pipeline": destination_pipeline["name"]
|
||||
}
|
||||
|
||||
def update_destination_webhook(self, session, base_url, user_email, user_password, org_id, template_name, destination_name):
|
||||
"""Update an Webhook destination."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
skip_tls_verify_value = False
|
||||
payload = {
|
||||
"url": "https://jsonplaceholder.typicode.com/todos",
|
||||
"method": "get",
|
||||
"template": template_name,
|
||||
"headers": {},
|
||||
"name": destination_name,
|
||||
"skip_tls_verify": skip_tls_verify_value
|
||||
}
|
||||
response = session.put(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to update destination: {response.content}"
|
||||
destination_webhook = response.json()
|
||||
assert isinstance(destination_webhook, dict), "Response is not a valid destination object"
|
||||
|
||||
def update_destination_email(self, session, base_url, user_email, user_password, org_id, email_address_admin, template_email, destination_name):
|
||||
"""Update an Email destination."""
|
||||
|
||||
# Set up authentication
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
# Define the payload for the PUT request
|
||||
payload = {
|
||||
"url": "",
|
||||
"type": "email",
|
||||
"emails": [email_address_admin],
|
||||
"method": "post",
|
||||
"template": template_email,
|
||||
"headers": {},
|
||||
"name": destination_name,
|
||||
"skip_tls_verify": False
|
||||
}
|
||||
|
||||
# Perform the API request to update the destination
|
||||
response = session.put(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}", json=payload)
|
||||
|
||||
# Check for a successful response
|
||||
if response.status_code != 200:
|
||||
print(f"Failed to update destination: {response.content}") # Log the response content
|
||||
response.raise_for_status() # Raise an exception for HTTP errors
|
||||
|
||||
# For successful updates, we'll verify the update by getting the destination
|
||||
verify_response = session.get(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
||||
|
||||
if verify_response.status_code != 200:
|
||||
print(f"Failed to verify destination update: {verify_response.content}")
|
||||
verify_response.raise_for_status()
|
||||
|
||||
destination_email = verify_response.json()
|
||||
|
||||
# Validate the response structure
|
||||
if not isinstance(destination_email, dict):
|
||||
raise ValueError("Response is not a valid destination object")
|
||||
|
||||
# Verify the update was successful by checking the name matches
|
||||
if destination_email.get('name') != destination_name:
|
||||
raise ValueError(f"Updated destination name mismatch. Expected: {destination_name}, Got: {destination_email.get('name')}")
|
||||
|
||||
return {
|
||||
"name_destination_email": destination_name
|
||||
}
|
||||
|
||||
def update_destination_pipeline(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Update a pipeline destination."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
skip_tls_verify_value = False
|
||||
payload = {
|
||||
"url": "https://jsonplaceholder.typicode.com/todos",
|
||||
"method": "get",
|
||||
"template": "",
|
||||
"headers": {},
|
||||
"name": destination_name,
|
||||
"skip_tls_verify": skip_tls_verify_value
|
||||
}
|
||||
response = session.put(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to update destination: {response.content}"
|
||||
destination_pipeline = response.json()
|
||||
assert isinstance(destination_pipeline, dict), "Response is not a valid destination object"
|
||||
|
||||
|
||||
def delete_destination(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Delete destinations of a specified type."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
resp_delete_destination = session.delete(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
||||
assert resp_delete_destination.status_code == 200, f"Failed to delete destination {destination_name}: {resp_delete_destination.content}"
|
||||
|
||||
# Wait for a few seconds to allow the data to be deleted
|
||||
time.sleep(10) # Increase this time if necessary
|
||||
|
||||
# Verify deletion
|
||||
resp_ver_destination = session.get(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
||||
assert resp_ver_destination.status_code == 404, f"Expected 404 for {destination_name}, but got {resp_ver_destination.status_code}"
|
||||
|
||||
return resp_delete_destination # Return the list of deleted destination names for verification if needed
|
||||
|
||||
def validate_deleted_destination_webhook(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Running an E2E test for validating deleted destinations in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all destinations
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations")
|
||||
assert response.status_code == 200, f"Failed to retrieve destinations: {response.content}"
|
||||
|
||||
destinations = response.json()
|
||||
|
||||
# Filter destinations that contain 'destination_webhook' and match the expected name
|
||||
webhook_destinations = [destination for destination in destinations
|
||||
if "destination_webhook" in destination["name"]
|
||||
and destination["name"].startswith(f"destination_webhook_{DestinationPage.Unique_value_destination}")]
|
||||
|
||||
# Check if any templates of the specified type exist
|
||||
if len(webhook_destinations) == 0:
|
||||
print(f"No destinations found of type '{destination_name}'")
|
||||
else:
|
||||
print(f"Destinations found of type '{destination_name}': {[destination['name'] for destination in webhook_destinations]}")
|
||||
return webhook_destinations
|
||||
|
||||
def validate_deleted_destination_email(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Running an E2E test for validating deleted destinations in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all destinations
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations")
|
||||
assert response.status_code == 200, f"Failed to retrieve destinations: {response.content}"
|
||||
|
||||
destinations = response.json()
|
||||
|
||||
# Filter destinations that contain 'destination_email' and match the expected name
|
||||
email_destinations = [destination for destination in destinations
|
||||
if "destination_email" in destination["name"]
|
||||
and destination["name"].startswith(f"destination_email_{DestinationPage.Unique_value_destination}")]
|
||||
|
||||
# Check if any templates of the specified type exist
|
||||
if len(email_destinations) == 0:
|
||||
print(f"No destinations found of type '{destination_name}'")
|
||||
# Optionally, you can choose to skip the assertion or handle it differently
|
||||
else:
|
||||
assert len(email_destinations) > 0, f"No destinations found of type '{destination_name}'"
|
||||
print(f"Destinations found of type '{destination_name}': {[destination['name'] for destination in email_destinations]}")
|
||||
|
||||
return email_destinations
|
||||
|
||||
def validate_deleted_destination_pipeline(self, session, base_url, user_email, user_password, org_id, destination_name):
|
||||
"""Running an E2E test for validating deleted destinations in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all destinations
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/destinations")
|
||||
assert response.status_code == 200, f"Failed to retrieve destinations: {response.content}"
|
||||
|
||||
destinations = response.json()
|
||||
|
||||
# Filter destinations that contain 'destination_pipeline' and match the expected name
|
||||
pipeline_destinations = [destination for destination in destinations
|
||||
if "destination_pipeline" in destination["name"]
|
||||
and destination["name"].startswith(f"destination_pipeline_{DestinationPage.Unique_value_destination}")]
|
||||
|
||||
# Check if any templates of the specified type exist
|
||||
if len(pipeline_destinations) == 0:
|
||||
print(f"No destinations found of type '{destination_name}'")
|
||||
else:
|
||||
print(f"Destinations found of type '{destination_name}': {[destination['name'] for destination in pipeline_destinations]}")
|
||||
return pipeline_destinations
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
import random
|
||||
import base64
|
||||
import io
|
||||
from pathlib import Path
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
root_dir = Path(__file__).parent.parent.parent # Navigate up to the root directory
|
||||
|
||||
class EnrichmentPage:
|
||||
# Make Unique_value_enrichment a class variable
|
||||
Unique_value_enrichment = f"d4m20_{random.randint(100000, 999999)}"
|
||||
|
||||
BOUNDARY = "----WebKitFormBoundaryaQgmYHuE6dQrlLss"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_multipart_data(self, fields, boundary):
|
||||
"""Create multipart form data with a specified boundary."""
|
||||
boundary_str = f"--{boundary}"
|
||||
lines = []
|
||||
|
||||
# Add the fields to the multipart data
|
||||
for key, value in fields.items():
|
||||
if isinstance(value, tuple):
|
||||
# This is the file field
|
||||
filename, file_obj, content_type = value
|
||||
lines.append(f"{boundary_str}")
|
||||
lines.append(f'Content-Disposition: form-data; name="{key}"; filename="{filename}"')
|
||||
lines.append(f'Content-Type: {content_type}')
|
||||
lines.append("")
|
||||
# Read the file content as binary
|
||||
lines.append(file_obj.read()) # Read as binary data
|
||||
else:
|
||||
# Regular fields
|
||||
lines.append(f"{boundary_str}")
|
||||
lines.append(f'Content-Disposition: form-data; name="{key}"')
|
||||
lines.append("")
|
||||
lines.append(value)
|
||||
|
||||
# End the multipart data
|
||||
lines.append(f"{boundary_str}--")
|
||||
|
||||
return b"\r\n".join(line.encode('utf-8') if isinstance(line, str) else line for line in lines)
|
||||
|
||||
def create_enrichment_table(self, session, base_url, user_email, user_password, org_id, enrichment_table_name):
|
||||
"""Create an enrichment table."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
headers = {
|
||||
'Accept': 'application/json, text/plain, */*',
|
||||
'Accept-Language': 'en-US,en;q=0.9',
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'Content-Type': f'multipart/form-data; boundary={self.BOUNDARY}',
|
||||
'Pragma': 'no-cache',
|
||||
'Custom-Header': 'value'
|
||||
}
|
||||
|
||||
cookies = {
|
||||
'auth_ext': '{"auth_ext":"","refresh_token":"","request_time":0,"expires_in":0}',
|
||||
'auth_tokens': f'{{"access_token":"Basic {base64.b64encode((user_email + ":" + user_password).encode()).decode()}","refresh_token":""}}'
|
||||
}
|
||||
|
||||
# Create sample data directly instead of reading from file
|
||||
sample_data = """protocol_number,keyword,protocol_description
|
||||
0,HOPOPT,IPv6 Hop-by-Hop Option
|
||||
1,ICMP,Internet Control Message
|
||||
2,IGMP,Internet Group Management"""
|
||||
|
||||
# Create in-memory file-like object with the sample data
|
||||
file_like_object = io.BytesIO(sample_data.encode())
|
||||
|
||||
fields = {
|
||||
'file': ('protocols.csv', file_like_object, 'text/csv')
|
||||
}
|
||||
|
||||
multipart_data = self.create_multipart_data(fields, self.BOUNDARY)
|
||||
url = f"{base_url}api/{org_id}/enrichment_tables/{enrichment_table_name}?append=false"
|
||||
|
||||
response = session.post(url, headers=headers, cookies=cookies, data=multipart_data)
|
||||
assert response.status_code == 200, f"Failed to enrich table: {response.content.decode()}"
|
||||
return response
|
||||
|
||||
def create_enrichment(self, session, base_url, user_email, user_password, org_id, enrichment_name):
|
||||
"""Create an enrichment."""
|
||||
# First create the enrichment table
|
||||
response = self.create_enrichment_table(session, base_url, user_email, user_password, org_id, enrichment_name)
|
||||
|
||||
# Return the enrichment ID if needed
|
||||
if response.status_code == 200:
|
||||
return enrichment_name # or response.json()['id'] if the API returns an ID
|
||||
return None
|
||||
@@ -0,0 +1,27 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
class FolderPage:
|
||||
# Make Unique_value_destination a class variable
|
||||
Unique_value_folder = f"d4m21_{random.randint(100000, 999999)}"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_folder(self, session, base_url, user_email, user_password, org_id, folder_name):
|
||||
"""Create a folder."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
payload = {
|
||||
"description": folder_name,
|
||||
"folderId": "",
|
||||
"name": folder_name
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/folders", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create folder: {response.content}"
|
||||
folder_id = response.json()["folderId"]
|
||||
return folder_id
|
||||
@@ -0,0 +1,30 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
|
||||
class FunctionPage:
|
||||
# Make Unique_value_destination a class variable
|
||||
Unique_value_function = f"d4m22_{random.randint(100000, 999999)}"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_function(self, session, base_url, user_email, user_password, org_id, function_name):
|
||||
"""Create a function."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Custom-Header": "value"
|
||||
}
|
||||
# created_time = datetime.now().isoformat()
|
||||
payload = {
|
||||
"function": ".a=190025552",
|
||||
"name": function_name,
|
||||
"params": "row",
|
||||
"transType": 0,
|
||||
}
|
||||
response = session.post(f"{base_url}api/{org_id}/functions", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create function: {response.content.decode()}"
|
||||
return response
|
||||
@@ -0,0 +1,152 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
class PipelinePage:
|
||||
# Make Unique_value_destination a class variable
|
||||
Unique_value_pipeline = f"d4m23_{random.randint(100000, 999999)}"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_realTime_pipeline(self, session, base_url, user_email, user_password, org_id, stream_name, realTime_pipeline_name):
|
||||
"""Create a realTime pipeline."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
payload = {
|
||||
"name": realTime_pipeline_name,
|
||||
"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,
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/pipelines", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create scheduled pipeline: {response.content.decode()}"
|
||||
return response
|
||||
|
||||
|
||||
def create_scheduled_pipeline(self, session, base_url, user_email, user_password, org_id, stream_name, scheduled_pipeline_name):
|
||||
"""Create a scheduled pipeline."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
payload = {
|
||||
"name": scheduled_pipeline_name,
|
||||
"description": "",
|
||||
"source": {"source_type": "scheduled"},
|
||||
"nodes": [
|
||||
{
|
||||
"id": "8b03771c-2e94-4c2f-b902-8fb8b8e66df2",
|
||||
"type": "input",
|
||||
"dimensions": {"width": 140, "height": 42},
|
||||
"computedPosition": {"x": 212.1047849020086, "y": 93.00563781047852, "z": 0},
|
||||
"handleBounds": {
|
||||
"source": [{"id": "output", "position": "bottom", "x": 55.03916432328342, "y": 36.00006521998462, "width": 30, "height": 10}],
|
||||
"target": []
|
||||
},
|
||||
"position": {"x": 212.1047849020086, "y": 93.00563781047852},
|
||||
"data": {
|
||||
"label": "8b03771c-2e94-4c2f-b902-8fb8b8e66df2",
|
||||
"node_type": "query",
|
||||
"stream_type": "logs",
|
||||
"org_id": org_id,
|
||||
"query_condition": {
|
||||
"type": "sql",
|
||||
"conditions": None,
|
||||
"sql": "select * from default",
|
||||
"promql": None,
|
||||
"promql_condition": None,
|
||||
"aggregation": None,
|
||||
"vrl_function": None,
|
||||
"search_event_type": "DerivedStream"
|
||||
},
|
||||
"trigger_condition": {
|
||||
"period": 15,
|
||||
"operator": "=",
|
||||
"threshold": 0,
|
||||
"frequency": 15,
|
||||
"cron": "",
|
||||
"frequency_type": "minutes",
|
||||
"silence": 0
|
||||
}
|
||||
},
|
||||
"io_type": "input"
|
||||
},
|
||||
{
|
||||
"id": "cfa476a0-24f3-4ca3-a14f-b56b04729922",
|
||||
"type": "output",
|
||||
"dimensions": {"width": 250, "height": 42},
|
||||
"computedPosition": {"x": 269.2781219770897, "y": 161.6932759313432, "z": 0},
|
||||
"handleBounds": {
|
||||
"source": [],
|
||||
"target": [{"id": "input", "position": "top", "x": 110.11741720560622, "y": -4.000002219844093, "width": 30, "height": 10}]
|
||||
},
|
||||
"position": {"x": 269.2781219770897, "y": 161.6932759313432},
|
||||
"data": {
|
||||
"label": "cfa476a0-24f3-4ca3-a14f-b56b04729922",
|
||||
"node_type": "stream",
|
||||
"stream_type": "logs",
|
||||
"stream_name": stream_name,
|
||||
"org_id": org_id
|
||||
},
|
||||
"io_type": "output"
|
||||
}
|
||||
],
|
||||
"edges": [
|
||||
{
|
||||
"id": "e8b03771c-2e94-4c2f-b902-8fb8b8e66df2-cfa476a0-24f3-4ca3-a14f-b56b04729922",
|
||||
"type": "custom",
|
||||
"source": "8b03771c-2e94-4c2f-b902-8fb8b8e66df2",
|
||||
"target": "cfa476a0-24f3-4ca3-a14f-b56b04729922",
|
||||
"data": {},
|
||||
"markerEnd": {"type": "arrowclosed", "width": 20, "height": 20},
|
||||
"style": {"strokeWidth": 2},
|
||||
"animated": True
|
||||
}
|
||||
],
|
||||
"org": org_id,
|
||||
"schedule": {
|
||||
"enabled": True,
|
||||
"frequency": "daily", # Example frequency, adjust as needed
|
||||
"time": "02:00", # Example time, adjust as needed
|
||||
}
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/pipelines", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create scheduled pipeline: {response.content.decode()}"
|
||||
return response
|
||||
@@ -0,0 +1,111 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
|
||||
|
||||
class ReportPage:
|
||||
Unique_value_report = f"d4m21_{random.randint(100000, 999999)}" # Class variable
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_scheduled_report(self, session, base_url, user_email, user_password, org_id, dashboard_id, folder_id, scheduled_report_name):
|
||||
"""Create a scheduled report."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"dashboards": [
|
||||
{
|
||||
"folder": folder_id,
|
||||
"dashboard": dashboard_id,
|
||||
"tabs": ["default"],
|
||||
"variables": [],
|
||||
"timerange": {
|
||||
"type": "relative",
|
||||
"period": "30m",
|
||||
"from": 1741687525799000,
|
||||
"to": 1741689325799000
|
||||
}
|
||||
}
|
||||
],
|
||||
"description": "",
|
||||
"destinations": [
|
||||
{
|
||||
"email": user_email
|
||||
}
|
||||
],
|
||||
"enabled": True,
|
||||
"media_type": "Pdf",
|
||||
"name": scheduled_report_name,
|
||||
"title": scheduled_report_name,
|
||||
"message": "hi",
|
||||
"orgId": org_id,
|
||||
"start": 1741689360000000,
|
||||
"frequency": {
|
||||
"interval": 1,
|
||||
"type": "once",
|
||||
"cron": ""
|
||||
},
|
||||
"user": "",
|
||||
"password": "",
|
||||
"timezone": "Asia/Calcutta",
|
||||
"timezoneOffset": 330,
|
||||
"owner": user_email,
|
||||
"lastEditedBy": user_email,
|
||||
"report_type": "PDF"
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/reports", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create scheduled report: {response.content.decode()}"
|
||||
return response
|
||||
|
||||
def create_cached_report(self, session, base_url, user_email, user_password, org_id, dashboard_id, folder_id, cached_report_name):
|
||||
"""Create a cached report."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
payload = {
|
||||
"dashboards": [
|
||||
{
|
||||
"folder": folder_id,
|
||||
"dashboard": dashboard_id,
|
||||
"tabs": ["default"],
|
||||
"variables": [],
|
||||
"timerange": {
|
||||
"type": "relative",
|
||||
"period": "30m",
|
||||
"from": 1741687525799000,
|
||||
"to": 1741689325799000
|
||||
}
|
||||
}
|
||||
],
|
||||
"description": "",
|
||||
"destinations": [
|
||||
|
||||
],
|
||||
"enabled": True,
|
||||
"media_type": "Pdf",
|
||||
"name": cached_report_name,
|
||||
"title": cached_report_name,
|
||||
"message": "hi",
|
||||
"orgId": org_id,
|
||||
"start": 1741689360000000,
|
||||
"frequency": {
|
||||
"interval": 1,
|
||||
"type": "once",
|
||||
"cron": ""
|
||||
},
|
||||
"user": "",
|
||||
"password": "",
|
||||
"timezone": "Asia/Calcutta",
|
||||
"timezoneOffset": 330,
|
||||
"owner": user_email,
|
||||
"lastEditedBy": user_email,
|
||||
"report_type": "PDF"
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/reports", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create cached report: {response.content.decode()}"
|
||||
return response
|
||||
@@ -0,0 +1,86 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
class SavedViewPage:
|
||||
# Make Unique_value_destination a class variable
|
||||
Unique_value_savedview = f"d4m21_{random.randint(100000, 999999)}"
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_savedView(self, session, base_url, user_email, user_password, org_id, stream_name, savedview_name):
|
||||
"""Create a saved view."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"organizationIdentifier": org_id,
|
||||
"runQuery": False,
|
||||
"loading": False,
|
||||
"loadingHistogram": False,
|
||||
"loadingCounter": False,
|
||||
"loadingStream": False,
|
||||
"loadingSavedView": False,
|
||||
"shouldIgnoreWatcher": False,
|
||||
"communicationMethod": "http",
|
||||
"config": {
|
||||
"splitterModel": 20,
|
||||
"lastSplitterPosition": 0,
|
||||
"splitterLimit": [0, 40],
|
||||
"fnSplitterModel": 60,
|
||||
"fnLastSplitterPosition": 0,
|
||||
"fnSplitterLimit": [40, 100],
|
||||
"refreshTimes": [
|
||||
[{"label": "5 sec", "value": 5}, {"label": "1 min", "value": 60}, {"label": "1 hr", "value": 3600}],
|
||||
[{"label": "10 sec", "value": 10}, {"label": "5 min", "value": 300}, {"label": "2 hr", "value": 7200}],
|
||||
[{"label": "15 sec", "value": 15}, {"label": "15 min", "value": 900}, {"label": "1 day", "value": 86400}],
|
||||
[{"label": "30 sec", "value": 30}, {"label": "30 min", "value": 1800}]
|
||||
]
|
||||
},
|
||||
"meta": {
|
||||
# Include the rest of the meta fields as needed
|
||||
"logsVisualizeToggle": "logs",
|
||||
"refreshInterval": 0,
|
||||
"refreshIntervalLabel": "Off",
|
||||
"refreshHistogram": False,
|
||||
"showFields": True,
|
||||
"showQuery": True,
|
||||
"showHistogram": True,
|
||||
"showDetailTab": False,
|
||||
"toggleFunction": True,
|
||||
"searchApplied": True,
|
||||
"toggleSourceWrap": False,
|
||||
"histogramDirtyFlag": False,
|
||||
"sqlMode": True,
|
||||
"quickMode": False,
|
||||
"queryEditorPlaceholderFlag": True,
|
||||
"functionEditorPlaceholderFlag": True,
|
||||
"resultGrid": {
|
||||
"rowsPerPage": 50,
|
||||
"wrapCells": False,
|
||||
"manualRemoveFields": False,
|
||||
"chartInterval": "30 second",
|
||||
"chartKeyFormat": "HH:mm:ss",
|
||||
"navigation": {"currentRowIndex": 0},
|
||||
"showPagination": True
|
||||
},
|
||||
"jobId": "",
|
||||
"jobRecords": "100",
|
||||
"pageType": "logs",
|
||||
"regions": [],
|
||||
"clusters": [],
|
||||
"useUserDefinedSchemas": "user_defined_schema",
|
||||
"hasUserDefinedSchemas": False,
|
||||
"selectedTraceStream": "",
|
||||
"showSearchScheduler": False
|
||||
},
|
||||
"data": {
|
||||
"query": f"SELECT * FROM \"{stream_name}\"",
|
||||
# Add the rest of the data fields as needed
|
||||
},
|
||||
"view_name": savedview_name
|
||||
}
|
||||
|
||||
response = session.post(f"{base_url}api/{org_id}/savedviews", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create saved view: {response.content}"
|
||||
return response
|
||||
@@ -0,0 +1,67 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
|
||||
|
||||
class ServiceAccountPage:
|
||||
Unique_value_serviceaccount = f"d4m21_{random.randint(100000, 999999)}" # Class variable
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_service_account(self, session, base_url, user_email, user_password, org_id, email_address):
|
||||
|
||||
"""Create a service account."""
|
||||
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
|
||||
payload = {
|
||||
"email": email_address,
|
||||
"organization": org_id,
|
||||
"first_name": "Shyam",
|
||||
"last_name": "Panjiyar"
|
||||
}
|
||||
print("Payload for service account creation:", payload) # Log the payload
|
||||
response = session.post(
|
||||
f"{base_url}api/{org_id}/service_accounts",
|
||||
json=payload,
|
||||
headers=headers
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to create service account: {response.content.decode()}"
|
||||
|
||||
print(f"Service account created successfully: {response.json()}")
|
||||
|
||||
response_get = session.get(
|
||||
f"{base_url}api/{org_id}/service_accounts/{email_address}",
|
||||
headers=headers
|
||||
)
|
||||
assert response_get.status_code == 200, f"Failed to get service account: {response_get.content.decode()}"
|
||||
|
||||
print(f"Service account created successfully: {response_get.json()}")
|
||||
|
||||
token = response_get.json()['token']
|
||||
|
||||
print(f"Token: {token}")
|
||||
|
||||
return token
|
||||
|
||||
|
||||
|
||||
# def test_delete_serviceaccount(create_session, base_url):
|
||||
# """Running an E2E test for deleting a service account."""
|
||||
# session = create_session
|
||||
|
||||
# # Delete the service account
|
||||
# resp_delete_serviceaccount = session.delete(
|
||||
# f"{base_url}api/{org_id}/service_accounts/{email_id}"
|
||||
# )
|
||||
|
||||
# print("Service Account Deleted", resp_delete_serviceaccount.content)
|
||||
|
||||
# assert (
|
||||
# resp_delete_serviceaccount.status_code == 200
|
||||
# ), f"Expected 200, but got {resp_delete_serviceaccount.status_code} {resp_delete_serviceaccount.content}"
|
||||
@@ -0,0 +1,210 @@
|
||||
import random
|
||||
import time
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
class TemplatePage:
|
||||
Unique_value_temp = f"d4m21_{random.randint(100000, 999999)}" # Class variable
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_template_webhook(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Create a Webhook template."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": template_name,
|
||||
"body": '{\n \"text\": \"{alert_name} is active\"\n}',
|
||||
"type": "http",
|
||||
"title": ""
|
||||
}
|
||||
response = session.post(f"{base_url}api/{org_id}/alerts/templates", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create template: {response.content}"
|
||||
return response
|
||||
|
||||
def create_template_email(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Create an Email template."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": template_name,
|
||||
"body": '{"text": "For stream {stream_name} of organization {org} alert {alert_name} of type {alert_type} is active"}',
|
||||
"type": "email",
|
||||
"title": "email_title"
|
||||
}
|
||||
response = session.post(f"{base_url}api/{org_id}/alerts/templates", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create template: {response.content}"
|
||||
return response
|
||||
|
||||
def retrieve_templates_webhook(self, session, base_url, user_email, user_password, org_id):
|
||||
"""Retrieve templates and extract information about all templates matching the specified criteria."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/templates")
|
||||
assert response.status_code == 200, f"Failed to validate templates: {response.content}"
|
||||
|
||||
# Parse the JSON response
|
||||
templates = response.json()
|
||||
assert len(templates) > 0, "No templates found"
|
||||
|
||||
# Filter templates that contain 'template_webhook' and match the expected name
|
||||
webhook_templates = [template for template in templates
|
||||
if "template_webhook" in template["name"]
|
||||
and template["name"].startswith(f"template_webhook_{TemplatePage.Unique_value_temp}")]
|
||||
|
||||
# Assert that there are templates matching the criteria
|
||||
assert len(webhook_templates) > 0, f"No templates found containing 'template_webhook' with title 'template_webhook_{TemplatePage.Unique_value_temp}'"
|
||||
|
||||
# Return structured data
|
||||
return {
|
||||
"count": len(webhook_templates),
|
||||
"templates": webhook_templates
|
||||
}
|
||||
|
||||
def retrieve_template_webhook(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Retrieve the template webhook and assert the body."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/templates/{template_name}")
|
||||
assert response.status_code == 200, f"Failed to validate template: {response.content}"
|
||||
|
||||
template_webhook = response.json() # Assuming the response is a single template, not a list.
|
||||
|
||||
# Assert that the response is a valid template object
|
||||
assert isinstance(template_webhook, dict), "Response is not a valid template object"
|
||||
|
||||
return {
|
||||
"name_template_webhook": template_webhook["name"],
|
||||
"body_template_webhook": template_webhook["body"]
|
||||
}
|
||||
|
||||
def retrieve_templates_email(self, session, base_url, user_email, user_password, org_id):
|
||||
"""Retrieve templates and extract the first template name containing 'template_email' with title 'email_title'."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/templates")
|
||||
assert response.status_code == 200, f"Failed to validate templates: {response.content}"
|
||||
|
||||
templates = response.json()
|
||||
assert len(templates) > 0, "No templates found"
|
||||
|
||||
|
||||
email_templates = [
|
||||
template for template in templates
|
||||
if template["name"].startswith(f"template_email_{TemplatePage.Unique_value_temp}")
|
||||
]
|
||||
|
||||
# Assert that there are templates matching the criteria
|
||||
assert len(email_templates) > 0, f"No templates found containing 'template_email' with title 'template_email_{TemplatePage.Unique_value_temp}'"
|
||||
|
||||
# Return structured data
|
||||
return {
|
||||
"count": len(email_templates),
|
||||
"templates": email_templates
|
||||
}
|
||||
|
||||
def retrieve_template_email(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Retrieve the template email and assert the title."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/templates/{template_name}")
|
||||
assert response.status_code == 200, f"Failed to validate template: {response.content}"
|
||||
|
||||
template_email = response.json() # Assuming the response is a single template, not a list.
|
||||
assert isinstance(template_email, dict), "Response is not a valid template object"
|
||||
|
||||
|
||||
return {
|
||||
"name_template_email": template_email["name"],
|
||||
"title_template_email": template_email["title"]
|
||||
}
|
||||
|
||||
def update_template_webhook(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Update an Webhook template."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": template_name,
|
||||
"body": '{"text": "For stream {stream_name} of organization {org} alert {alert_name} of type {alert_type} is active"}',
|
||||
"type": "http",
|
||||
"title": ""
|
||||
}
|
||||
response = session.put(f"{base_url}api/{org_id}/alerts/templates/{template_name}", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to update template: {response.content}"
|
||||
template_webhook = response.json()
|
||||
assert isinstance(template_webhook, dict), "Response is not a valid template object"
|
||||
|
||||
def update_template_email(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Update an Email template."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
payload = {
|
||||
"name": template_name,
|
||||
"body": '{"text": "For stream {stream_name} of organization {org} alert {alert_name} of type {alert_type} is active"}',
|
||||
"type": "email",
|
||||
"title": "email_title_updated"
|
||||
}
|
||||
response = session.put(f"{base_url}api/{org_id}/alerts/templates/{template_name}", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to update template: {response.content}"
|
||||
template_email = response.json()
|
||||
assert isinstance(template_email, dict), "Response is not a valid template object"
|
||||
|
||||
def delete_template(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Delete templates of a specified type."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
resp_delete_template = session.delete(f"{base_url}api/{org_id}/alerts/templates/{template_name}")
|
||||
assert resp_delete_template.status_code == 200, f"Failed to delete template {template_name}: {resp_delete_template.content}"
|
||||
|
||||
# Wait for a few seconds to allow the data to be deleted
|
||||
time.sleep(10) # Increase this time if necessary
|
||||
|
||||
# Verify deletion
|
||||
resp_ver_template = session.get(f"{base_url}api/{org_id}/alerts/templates/{template_name}")
|
||||
assert resp_ver_template.status_code == 404, f"Expected 404 for {template_name}, but got {resp_ver_template.status_code}"
|
||||
|
||||
return resp_delete_template # Return the list of deleted template names for verification if needed
|
||||
|
||||
def validate_deleted_template_webhook(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Running an E2E test for validating deleted templates in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all templates
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/templates")
|
||||
assert response.status_code == 200, f"Failed to retrieve templates: {response.content}"
|
||||
|
||||
templates = response.json()
|
||||
|
||||
# Filter templates that contain 'template_webhook' and match the expected name
|
||||
webhook_templates = [template for template in templates
|
||||
if "template_webhook" in template["name"]
|
||||
and template["name"].startswith(f"template_webhook_{TemplatePage.Unique_value_temp}")]
|
||||
|
||||
# Check if any templates of the specified type exist
|
||||
if len(webhook_templates) == 0:
|
||||
print(f"No templates found of type '{template_name}'")
|
||||
else:
|
||||
print(f"Templates found of type '{template_name}': {[template['name'] for template in webhook_templates]}")
|
||||
return webhook_templates
|
||||
|
||||
def validate_deleted_template_email(self, session, base_url, user_email, user_password, org_id, template_name):
|
||||
"""Running an E2E test for validating deleted templates in SC."""
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
# Retrieve all templates
|
||||
response = session.get(f"{base_url}api/{org_id}/alerts/templates")
|
||||
assert response.status_code == 200, f"Failed to retrieve templates: {response.content}"
|
||||
|
||||
templates = response.json()
|
||||
|
||||
# Filter templates that contain 'template_email' and match the expected name
|
||||
email_templates = [template for template in templates
|
||||
if "template_email" in template["name"]
|
||||
and template["name"].startswith(f"template_email_{TemplatePage.Unique_value_temp}")]
|
||||
|
||||
# Check if any templates of the specified type exist
|
||||
if len(email_templates) == 0:
|
||||
print(f"No templates found of type '{template_name}'")
|
||||
# Optionally, you can choose to skip the assertion or handle it differently
|
||||
else:
|
||||
assert len(email_templates) > 0, f"No templates found of type '{template_name}'"
|
||||
print(f"Templates found of type '{template_name}': {[template['name'] for template in email_templates]}")
|
||||
|
||||
return email_templates
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
import random
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
|
||||
|
||||
class UserPage:
|
||||
Unique_value_user = f"d4m21_{random.randint(100000, 999999)}" # Class variable
|
||||
|
||||
def __init__(self, session, base_url, org_id):
|
||||
self.session = session
|
||||
self.base_url = base_url
|
||||
self.org_id = org_id
|
||||
|
||||
def create_user_admin(self, session, base_url, user_email, user_password, org_id, email_address):
|
||||
"""Create a user as admin."""
|
||||
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
||||
session.auth = HTTPBasicAuth(user_email, user_password)
|
||||
|
||||
payload = {
|
||||
"organization": org_id,
|
||||
"email": email_address,
|
||||
"password": "12345678",
|
||||
"first_name": "Shyam",
|
||||
"last_name": "P",
|
||||
"role": "admin",
|
||||
}
|
||||
|
||||
print("Payload for user creation:", payload) # Log the payload
|
||||
response = session.post(f"{base_url}api/{org_id}/users", json=payload, headers=headers)
|
||||
assert response.status_code == 200, f"Failed to create user as admin: {response.content.decode()}"
|
||||
return response
|
||||
@@ -0,0 +1,31 @@
|
||||
import os
|
||||
from create_objects import create_objects
|
||||
|
||||
from verify_objects import verify_objects
|
||||
from edit_objects import edit_objects
|
||||
from delete_objects import delete_objects
|
||||
from verifydeleted_objects import verify_deleted_objects
|
||||
from verifyedited_objects import verify_edited_objects
|
||||
|
||||
|
||||
|
||||
# Constants for WebSocket URL and user credentials
|
||||
ZO_BASE_URL = os.environ.get("ZO_BASE_URL") # Use environment variable
|
||||
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
|
||||
ZO_BASE_URL_SC2 = os.environ.get("ZO_BASE_URL_SC2") # Use environment variable
|
||||
|
||||
|
||||
total_count = 2
|
||||
org_id = "default"
|
||||
stream_name = "default"
|
||||
|
||||
def test_workflow(create_session, base_url):
|
||||
|
||||
create_objects(create_session, base_url, ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD, org_id, stream_name, total_count)
|
||||
# verify_objects(create_session, base_url, ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD, org_id, stream_name, total_count)
|
||||
# edit_objects(create_session, base_url, ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD, org_id, stream_name, total_count)
|
||||
# verify_edited_objects(create_session, base_url, ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD, org_id, stream_name, total_count)
|
||||
# delete_objects(create_session, base_url, ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD, org_id, stream_name, total_count)
|
||||
# verify_deleted_objects(create_session, base_url, ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD, org_id, stream_name, total_count)
|
||||
@@ -0,0 +1,61 @@
|
||||
from pages.template_page import TemplatePage
|
||||
from pages.destination_page import DestinationPage
|
||||
from pages.alert_page import AlertPage
|
||||
|
||||
def verify_objects(session, base_url, user_email, user_password, org_id, stream_name, num_objects):
|
||||
"""Verify objects in the OpenObserve running instance."""
|
||||
|
||||
template_page = TemplatePage(session, base_url, org_id)
|
||||
destination_page = DestinationPage(session, base_url, org_id)
|
||||
alert_page = AlertPage(session, base_url, org_id)
|
||||
|
||||
for i in range(num_objects):
|
||||
|
||||
|
||||
# Retrieve templates
|
||||
template_name_webhook = f"template_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.retrieve_templates_webhook(session, base_url, user_email, user_password, org_id)
|
||||
template_page.retrieve_template_webhook(session, base_url, user_email, user_password, org_id, template_name_webhook)
|
||||
|
||||
template_name_email = f"template_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.retrieve_templates_email(session, base_url, user_email, user_password, org_id)
|
||||
template_page.retrieve_template_email(session, base_url, user_email, user_password, org_id, template_name_email)
|
||||
|
||||
# Retrieve destinations
|
||||
destination_name_webhook = f"destination_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.retrieve_destinations_webhook(session, base_url, user_email, user_password, org_id)
|
||||
destination_page.retrieve_destination_webhook(session, base_url, user_email, user_password, org_id, destination_name_webhook)
|
||||
|
||||
destination_name_email = f"destination_email_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.retrieve_destinations_email(session, base_url, user_email, user_password, org_id)
|
||||
destination_page.retrieve_destination_email(session, base_url, user_email, user_password, org_id, destination_name_email)
|
||||
|
||||
destination_name_pipeline = f"destination_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.retrieve_destinations_pipeline(session, base_url, user_email, user_password, org_id)
|
||||
destination_page.retrieve_destination_pipeline(session, base_url, user_email, user_password, org_id, destination_name_pipeline)
|
||||
|
||||
# Retrieve alerts
|
||||
|
||||
alert_webhook = f"alert_webhook_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_standard(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_standard(session, base_url, user_email, user_password, org_id, stream_name, alert_webhook)
|
||||
|
||||
alert_email = f"alert_email_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_standard(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_standard(session, base_url, user_email, user_password, org_id, stream_name, alert_email)
|
||||
|
||||
alert_cron = f"alert_cron_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_cron(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_cron(session, base_url, user_email, user_password, org_id, stream_name, alert_cron)
|
||||
|
||||
alert_real_time = f"alert_real_time_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_real_time(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_real_time(session, base_url, user_email, user_password, org_id, stream_name, alert_real_time)
|
||||
|
||||
alert_sql = f"alert_sql_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_standard_sql(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_standard_sql(session, base_url, user_email, user_password, org_id, stream_name, alert_sql)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
from pages.template_page import TemplatePage
|
||||
from pages.destination_page import DestinationPage
|
||||
from pages.alert_page import AlertPage
|
||||
|
||||
def verify_deleted_objects(session, base_url, user_email, user_password, org_id, stream_name, num_objects):
|
||||
"""Verify deleted objects in the OpenObserve running instance."""
|
||||
|
||||
template_page = TemplatePage(session, base_url, org_id)
|
||||
destination_page = DestinationPage(session, base_url, org_id)
|
||||
alert_page = AlertPage(session, base_url, org_id)
|
||||
|
||||
for i in range(num_objects):
|
||||
|
||||
# Validate deleted alerts
|
||||
alert_webhook = f"alert_webhook_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.validate_deleted_standard_alert(session, base_url, user_email, user_password, org_id, stream_name, alert_webhook)
|
||||
|
||||
alert_email = f"alert_email_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.validate_deleted_standard_alert(session, base_url, user_email, user_password, org_id, stream_name, alert_email)
|
||||
|
||||
alert_cron = f"alert_cron_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.validate_deleted_standard_alert_cron(session, base_url, user_email, user_password, org_id, stream_name, alert_cron)
|
||||
|
||||
alert_real_time = f"alert_real_time_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.validate_deleted_real_time_alert(session, base_url, user_email, user_password, org_id, stream_name, alert_real_time)
|
||||
|
||||
alert_sql = f"alert_sql_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.validate_deleted_standard_alert_sql(session, base_url, user_email, user_password, org_id, stream_name, alert_sql)
|
||||
|
||||
|
||||
# Validate deleted destinations
|
||||
destination_name_webhook = f"destination_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.validate_deleted_destination_webhook(session, base_url, user_email, user_password, org_id, destination_name_webhook)
|
||||
|
||||
destination_name_email = f"destination_email_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.validate_deleted_destination_email(session, base_url, user_email, user_password, org_id, destination_name_email)
|
||||
|
||||
destination_name_pipeline = f"destination_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.validate_deleted_destination_pipeline(session, base_url, user_email, user_password, org_id, destination_name_pipeline)
|
||||
|
||||
# Validate deleted templates
|
||||
template_name_webhook = f"template_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.validate_deleted_template_webhook(session, base_url, user_email, user_password, org_id, template_name_webhook)
|
||||
|
||||
|
||||
template_name_email = f"template_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.validate_deleted_template_email(session, base_url, user_email, user_password, org_id, template_name_email)
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
from pages.template_page import TemplatePage
|
||||
from pages.destination_page import DestinationPage
|
||||
from pages.alert_page import AlertPage
|
||||
def verify_edited_objects(session, base_url, user_email, user_password, org_id, stream_name, num_objects):
|
||||
"""Verify edited objects in the OpenObserve running instance."""
|
||||
|
||||
|
||||
template_page = TemplatePage(session, base_url, org_id)
|
||||
destination_page = DestinationPage(session, base_url, org_id)
|
||||
alert_page = AlertPage(session, base_url, org_id)
|
||||
for i in range(num_objects):
|
||||
|
||||
|
||||
# Retrieve templates
|
||||
template_name_webhook = f"template_webhook_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.retrieve_templates_webhook(session, base_url, user_email, user_password, org_id)
|
||||
template_page.retrieve_template_webhook(session, base_url, user_email, user_password, org_id, template_name_webhook)
|
||||
|
||||
template_name_email = f"template_email_{template_page.Unique_value_temp}_{i}"
|
||||
template_page.retrieve_templates_email(session, base_url, user_email, user_password, org_id)
|
||||
template_page.retrieve_template_email(session, base_url, user_email, user_password, org_id, template_name_email)
|
||||
|
||||
# Retrieve destinations
|
||||
destination_name_webhook = f"destination_webhook_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.retrieve_destinations_webhook(session, base_url, user_email, user_password, org_id)
|
||||
destination_page.retrieve_destination_webhook(session, base_url, user_email, user_password, org_id, destination_name_webhook)
|
||||
|
||||
destination_name_email = f"destination_email_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.retrieve_destinations_email(session, base_url, user_email, user_password, org_id)
|
||||
destination_page.retrieve_destination_email(session, base_url, user_email, user_password, org_id, destination_name_email)
|
||||
|
||||
destination_name_pipeline = f"destination_pipeline_{destination_page.Unique_value_destination}_{i}"
|
||||
destination_page.retrieve_destinations_pipeline(session, base_url, user_email, user_password, org_id)
|
||||
destination_page.retrieve_destination_pipeline(session, base_url, user_email, user_password, org_id, destination_name_pipeline)
|
||||
|
||||
# Retrieve alerts
|
||||
|
||||
alert_webhook = f"alert_webhook_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_standard(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_standard(session, base_url, user_email, user_password, org_id, stream_name, alert_webhook)
|
||||
|
||||
alert_email = f"alert_email_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_standard(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_standard(session, base_url, user_email, user_password, org_id, stream_name, alert_email)
|
||||
|
||||
alert_cron = f"alert_cron_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_cron(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_cron(session, base_url, user_email, user_password, org_id, stream_name, alert_cron)
|
||||
|
||||
alert_real_time = f"alert_real_time_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_real_time(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_real_time(session, base_url, user_email, user_password, org_id, stream_name, alert_real_time)
|
||||
|
||||
alert_sql = f"alert_sql_{alert_page.Unique_value_alert}_{i}"
|
||||
alert_page.retrieve_alerts_standard_sql(session, base_url, user_email, user_password, org_id, stream_name)
|
||||
alert_page.retrieve_alert_standard_sql(session, base_url, user_email, user_password, org_id, stream_name, alert_sql)
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user