wehub-resource-sync
bf2343b7e4
Integration Tests - MySQL + Elasticsearch / Detect Changes (push) Has been cancelled
Integration Tests - MySQL + Elasticsearch / integration-tests-mysql-elasticsearch (push) Has been cancelled
Integration Tests - PostgreSQL + Elasticsearch + Redis / Detect Changes (push) Has been cancelled
Integration Tests - PostgreSQL + Elasticsearch + Redis / integration-tests-postgres-elasticsearch-redis (push) Has been cancelled
Integration Tests - PostgreSQL + OpenSearch / Detect Changes (push) Has been cancelled
Integration Tests - PostgreSQL + OpenSearch / integration-tests-postgres-opensearch (push) Has been cancelled
Java Checkstyle / java-checkstyle (push) Has been cancelled
Maven Collate Tests / maven-collate-ci (push) Has been cancelled
OpenMetadata Service Unit Tests / openmetadata-service-unit-tests-status (push) Has been cancelled
Publish Package to Maven Central Repository / publish-maven-packages (push) Has been cancelled
OpenMetadata Service Unit Tests / Detect Changes (push) Has been cancelled
OpenMetadata Service Unit Tests / openmetadata-service-unit-tests (push) Has been cancelled
OpenMetadata Service Unit Tests / k8s_operator-unit-tests (push) Has been cancelled
42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
"""Load test for the test case resources"""
|
|
|
|
from locust import TaskSet, task
|
|
|
|
from _openmetadata_testutils.helpers.login_user import login_user
|
|
|
|
TEST_CASE_RESOURCE_PATH = "/api/v1/dataQuality/testCases"
|
|
|
|
|
|
class TestCaseTasks(TaskSet):
|
|
"""Test case resource load test"""
|
|
|
|
def _list_test_cases(self):
|
|
"""Paginate through the test cases"""
|
|
resp = self.client.get(
|
|
f"{TEST_CASE_RESOURCE_PATH}",
|
|
params={"limit": 10},
|
|
auth=self.bearer,
|
|
name=f"{TEST_CASE_RESOURCE_PATH}?limit=10",
|
|
)
|
|
after = resp.json().get("paging", {}).get("after")
|
|
while after:
|
|
resp = self.client.get(
|
|
f"{TEST_CASE_RESOURCE_PATH}",
|
|
params={"limit": 10, "after": after},
|
|
auth=self.bearer,
|
|
name=f"{TEST_CASE_RESOURCE_PATH}?limit=10",
|
|
)
|
|
after = resp.json().get("paging", {}).get("after")
|
|
|
|
@task(2)
|
|
def list_test_cases(self):
|
|
"""List test cases. Weighted 2"""
|
|
self._list_test_cases()
|
|
|
|
@task
|
|
def stop(self):
|
|
self.interrupt()
|
|
|
|
def on_start(self):
|
|
self.bearer = login_user(self.client)
|