test: root and admin search automation using pytest (#6433)

tests/api-testing/tests/create_objects.py

@@ -17,6 +17,7 @@ from pages.pipeline_page import PipelinePage
 from pages.report_page import ReportPage
 from pages.savedview_page import SavedViewPage
 from pages.serviceaccount_page import ServiceAccountPage
+from pages.search_page import SearchPage
 
 
 # Use environment variable
@@ -36,6 +37,7 @@ def create_objects(session, base_url, user_email, user_password, org_id, stream_
     report_page = ReportPage(session, base_url, org_id)
     savedview_page = SavedViewPage(session, base_url, org_id)
     serviceaccount_page = ServiceAccountPage(session, base_url, org_id)
+    search_page = SearchPage(session, base_url, org_id) 
 
     for i in range(num_objects):
 
@@ -106,6 +108,10 @@ def create_objects(session, base_url, user_email, user_password, org_id, stream_
         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)
 
+        # Create search
+        search_page.search_partition_logs_query_2_hours_stream(session, base_url, user_email, user_password, org_id, stream_name)   
+        search_page.search_cache_logs_query_2_hours_stream(session, base_url, user_email, user_password, org_id, stream_name)
+        search_page.search_histogram_logs_query_2_hours_stream(session, base_url, user_email, user_password, org_id, stream_name)
 
     # Created all objects successfully by admin user
 
@@ -177,5 +183,8 @@ def create_objects(session, base_url, user_email, user_password, org_id, stream_
         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)
 
-
+        # Create search
+        search_page.search_partition_logs_query_2_hours_stream(session, base_url, email_address_admin, "12345678", org_id, stream_name)   
+        search_page.search_cache_logs_query_2_hours_stream(session, base_url, email_address_admin, "12345678", org_id, stream_name)
+        search_page.search_histogram_logs_query_2_hours_stream(session, base_url, email_address_admin, "12345678", org_id, stream_name)
         

tests/api-testing/tests/pages/search_page.py

@@ -0,0 +1,101 @@
+import random
+
+from requests.auth import HTTPBasicAuth
+from datetime import datetime, timedelta, timezone
+
+class SearchPage:
+    # Make Unique_value_destination a class variable
+    Unique_value_search = f"d4m20_{random.randint(100000, 999999)}"
+
+    @property
+    def now(self):
+        return datetime.now(timezone.utc)
+    @property
+    def one_min_ago(self):
+        return int((self.now - timedelta(minutes=1)).timestamp() * 1000000)
+    @property
+    def one_hour_ago(self):
+        return int((self.now - timedelta(hours=1)).timestamp() * 1000000)
+    @property
+    def one_day_ago(self):
+        return int((self.now - timedelta(days=1)).timestamp() * 1000000)
+    @property
+    def two_hours_ago(self):
+        return int((self.now - timedelta(hours=2)).timestamp() * 1000000)
+    @property
+    def two_days_ago(self):
+        return int((self.now - timedelta(days=2)).timestamp() * 1000000)
+    
+    def __init__(self, session, base_url, org_id):
+        self.session = session
+        self.base_url = base_url
+        self.org_id = org_id
+
+    def search_partition_logs_query_2_hours_stream(self, session, base_url, user_email, user_password, org_id, stream_name):
+        headers = {"Content-Type": "application/json", "Custom-Header": "value"}
+        session.auth = HTTPBasicAuth(user_email, user_password)
+
+        now = datetime.now(timezone.utc)
+        end_time = int(now.timestamp() * 1000000)
+        payload = {
+            "sql": f"SELECT * FROM \"{stream_name}\"",
+            "start_time": self.two_hours_ago,
+            "end_time": end_time,
+            "sql_mode": "full"
+        }
+        response = session.post(f"{base_url}api/{org_id}/_search_partition?type=logs", headers=headers, json=payload)
+        assert response.status_code == 200, f"Failed to search logs: {response.content}"
+        print(f"Search Partition Logs {user_email} Response: {response.content}")
+        return response
+
+
+    def search_cache_logs_query_2_hours_stream(self, session, base_url, user_email, user_password, org_id, stream_name):
+        """Search logs with caching enabled."""
+        headers = {"Content-Type": "application/json", "Custom-Header": "value"}
+        session.auth = HTTPBasicAuth(user_email, user_password)
+
+        now = datetime.now(timezone.utc)
+        end_time = int(now.timestamp() * 1000000)
+
+        payload = {
+            "query": {
+                "sql": f"SELECT * FROM \"{stream_name}\"",
+                "start_time": self.two_hours_ago,
+                "end_time": end_time,
+                "from": 0,
+                "size": 50,
+                "quick_mode": False,
+                "sql_mode": "full"
+            }
+        }
+
+        response = session.post(
+            f"{base_url}api/{org_id}/_search?type=logs&search_type=ui&use_cache=true",
+            headers=headers,
+            json=payload
+        )
+        assert response.status_code == 200, f"Failed to search logs: {response.content}"
+        print(f"Search Cache Logs {user_email} Response: {response.content}")
+        return response
+
+    def search_histogram_logs_query_2_hours_stream(self, session, base_url, user_email, user_password, org_id, stream_name):
+        headers = {"Content-Type": "application/json", "Custom-Header": "value"}
+        session.auth = HTTPBasicAuth(user_email, user_password)
+
+        now = datetime.now(timezone.utc)
+        end_time = int(now.timestamp() * 1000000)
+        payload = {
+            "query": {
+                "sql": f"SELECT histogram(_timestamp, '1 minute') AS \"zo_sql_key\", COUNT(*) AS \"zo_sql_num\" FROM \"{stream_name}\" GROUP BY zo_sql_key ORDER BY zo_sql_key ASC",
+                "start_time": self.two_hours_ago,
+                "end_time": end_time,
+                "sql_mode": "full",
+                "size": -1
+            }
+        }
+        response = session.post(f"{base_url}api/{org_id}/_search?type=logs&search_type=ui&use_cache=true", headers=headers, json=payload)
+        assert response.status_code == 200, f"Failed to search logs: {response.content}"
+        print(f"Search Histogram Logs {user_email} Response: {response.content}")
+        return response
+
+