mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
Neha P
8a158884c8
- Verify full camel case token search: DbException should be found when searching for complete camel case identifiers - Verify atomic token search: UserAccountService should be searchable and return matching results for full service class names - Verify XML acronym tokenization: XMLHttpRequest should be properly tokenized and discoverable through match_all queries - Verify camel case with numbers: OAuth2TokenHandlershould handle mixed alphanumeric camel case tokens correctly - Verify multi-word service search:UserManagementService should support complex multi-word camel case service names
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
import requests
|
|
import pytest
|
|
import os
|
|
import random
|
|
import string
|
|
import logging
|
|
from pathlib import Path
|
|
import base64
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
BASE_URL = os.environ["ZO_BASE_URL"]
|
|
root_dir = Path(__file__).parent.parent.parent
|
|
|
|
@pytest.fixture(scope="module")
|
|
def random_string():
|
|
"""Generate a random string of specified length."""
|
|
def _generate(length=5):
|
|
# ascii_letters consist of alphabets for 'a' to 'z' and 'A' to 'Z'
|
|
# digits consist of '0' to '9'
|
|
characters = string.ascii_letters + string.digits
|
|
return "".join(random.choices(characters, k=length))
|
|
return _generate
|
|
|
|
def _create_session_inner():
|
|
s = requests.Session()
|
|
username = os.environ["ZO_ROOT_USER_EMAIL"]
|
|
password = os.environ["ZO_ROOT_USER_PASSWORD"]
|
|
s.auth = (username, password)
|
|
resp = s.post(BASE_URL)
|
|
return s
|
|
|
|
|
|
def _create_session_inner_v2():
|
|
s = requests.Session()
|
|
username = os.environ["ZO_ROOT_USER_EMAIL"]
|
|
password = os.environ["ZO_ROOT_USER_PASSWORD"]
|
|
# Set Authorization header for all requests
|
|
basic_auth = base64.b64encode(f"{username}:{password}".encode()).decode()
|
|
s.headers.update({"Authorization": f"Basic {basic_auth}"})
|
|
return s
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def create_session():
|
|
"""Create a session and return it."""
|
|
return _create_session_inner_v2()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def base_url():
|
|
"""Return the base URL."""
|
|
return BASE_URL
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def ingest_data():
|
|
"""Ingest data into the openobserve running instance."""
|
|
|
|
session = _create_session_inner()
|
|
|
|
# Ingest main test data
|
|
with open(root_dir / "test-data/logs_data.json") as f:
|
|
data = f.read()
|
|
|
|
stream_name = "stream_pytest_data"
|
|
org = "default"
|
|
url = f"{BASE_URL}api/{org}/{stream_name}/_json"
|
|
resp1 = session.post(url, data=data, headers={"Content-Type": "application/json"})
|
|
logging.info("Main data ingested successfully, status code: %s", resp1.status_code)
|
|
|
|
# Ingest camel case test data
|
|
with open(root_dir / "test-data/match_all.json") as f:
|
|
camel_data = f.read()
|
|
|
|
resp2 = session.post(url, data=camel_data, headers={"Content-Type": "application/json"})
|
|
logging.info("Camel case data ingested successfully, status code: %s", resp2.status_code)
|
|
|
|
return resp1.status_code == 200 and resp2.status_code == 200 |