test: add test function pytest cases (#5667)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **Tests**
	- Added comprehensive test cases for API function creation
- Included tests for invalid payloads, text-only inputs, and special
character scenarios
	- Verified successful function creation with valid payloads

- **Chores**
- Updated `.gitignore` to exclude temporary directories, configuration
files, and build artifacts

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Neha P
2025-01-10 19:17:47 +05:30
committed by GitHub
parent 7a9a2c52eb
commit 2bf8bb3720
2 changed files with 161 additions and 0 deletions
+14
View File
@@ -21,6 +21,20 @@ flamegraph.html
/tmp/
/data/
/.idea/
/.vscode
/.env
/.env*
/integration-tests
/download/
#web distribution
/web/dist
/web/.env
/tmp/
/data/
/.idea/
+147
View File
@@ -202,3 +202,150 @@ def test_e2e_addDeleteStreamFunction(create_session, base_url):
assert (
resp_delete_function.status_code == 200
), f"Deleting this function, but got {resp_delete_function.status_code} {resp_delete_function.content}"
def test_e2e_testinvalidfunction(create_session, base_url):
"""Running an E2E test for invalid test function"""
session = create_session
org_id = "default"
payload = {
"function": ".test=2jehwkhhe\n.",
"events": [
{
"_timestamp": 1735128523652186,
"job": "test",
"level": "info",
"log": "test message for openobserve"
},
{
"_timestamp": 1735128522644223,
"job": "test",
"level": "info",
"log": "test message for openobserve"
}
]
}
resp_test_function = session.post(
f"{base_url}api/{org_id}/functions/test", json=payload
)
print(resp_test_function.content)
assert (
resp_test_function.status_code == 400
), f"Invalid function creation, but got {resp_test_function.status_code} {resp_test_function.content}"
def test_e2e_onlytextfunction(create_session, base_url):
"""Running an E2E test for invalid test function"""
session = create_session
org_id = "default"
payload = {
"function": "test",
"events": [
{
"_timestamp": 1735128523652186,
"job": "test",
"level": "info",
"log": "test message for openobserve"
},
{
"_timestamp": 1735128522644223,
"job": "test",
"level": "info",
"log": "test message for openobserve"
}
]
}
resp_test_function = session.post(
f"{base_url}api/{org_id}/functions/test", json=payload
)
print(resp_test_function.content)
assert (
resp_test_function.status_code == 400
), f"Invalid function creation, but got {resp_test_function.status_code} {resp_test_function.content}"
def test_e2e_testonlyspecialcharfunction(create_session, base_url):
"""Running an E2E test for invalid test function"""
session = create_session
org_id = "default"
payload = {
"function": "=====",
"events": [
{
"_timestamp": 1735128523652186,
"job": "test",
"level": "info",
"log": "test message for openobserve"
},
{
"_timestamp": 1735128522644223,
"job": "test",
"level": "info",
"log": "test message for openobserve"
}
]
}
resp_test_function = session.post(
f"{base_url}api/{org_id}/functions/test", json=payload
)
print(resp_test_function.content)
assert (
resp_test_function.status_code == 400
), f"Invalid function creation, but got {resp_test_function.status_code} {resp_test_function.content}"
def test_e2e_testfunction(create_session, base_url):
"""Running an E2E test for the test function"""
session = create_session
# Define organization ID and payload
org_id = "default"
payload = {
"function": ".test=2\n.",
"events": [
{
"_timestamp": 1735128523652186,
"job": "test",
"level": "info",
"log": "test message for openobserve"
},
{
"_timestamp": 1735128522644223,
"job": "test",
"level": "info",
"log": "test message for openobserve"
}
]
}
# Send a POST request to the API
resp_test_function = session.post(
f"{base_url}api/{org_id}/functions/test", json=payload
)
# Print the response content for debugging
print(resp_test_function.content)
# Assert that the response status code is 200
assert resp_test_function.status_code == 200, (
f"Expected status code 200, but got {resp_test_function.status_code}. "
f"Response content: {resp_test_function.content}"
)
# Parse the response JSON
response_json = resp_test_function.json()
# Assert that `test=2` is present in each event in the response
for result in response_json.get("results", []):
assert result["event"].get("test") == 2, (
f"Expected 'test=2' in the response event, but got {result}."
)