Add option to list workflows API endpoint to filter by schedule (#822)

* Add option to filter workflows by empty or non-empty schedule

* Add tests
This commit is contained in:
Tessa Walsh 2023-05-05 15:05:19 -04:00 committed by GitHub
parent 542ad7a24a
commit 48d34bc3c4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 32 additions and 0 deletions

View File

@ -469,6 +469,7 @@ class CrawlConfigOps:
name: str = None,
description: str = None,
tags: Optional[List[str]] = None,
schedule: Optional[bool] = None,
sort_by: str = None,
sort_direction: int = -1,
):
@ -495,6 +496,12 @@ class CrawlConfigOps:
if description:
match_query["description"] = description
if schedule is not None:
if schedule:
match_query["schedule"] = {"$nin": ["", None]}
else:
match_query["schedule"] = {"$in": ["", None]}
# pylint: disable=duplicate-code
aggregate = [
{"$match": match_query},
@ -977,6 +984,7 @@ def init_crawl_config_api(
name: Optional[str] = None,
description: Optional[str] = None,
tag: Union[List[str], None] = Query(default=None),
schedule: Optional[bool] = None,
sortBy: str = None,
sortDirection: int = -1,
):
@ -998,6 +1006,7 @@ def init_crawl_config_api(
name=name,
description=description,
tags=tag,
schedule=schedule,
page_size=pageSize,
page=page,
sort_by=sortBy,

View File

@ -11,6 +11,7 @@ UPDATED_TAGS = ["tag3", "tag4"]
def test_add_crawl_config(crawler_auth_headers, default_org_id, sample_crawl_data):
# Create crawl config
sample_crawl_data["schedule"] = "0 0 * * *"
r = requests.post(
f"{API_PREFIX}/orgs/{default_org_id}/crawlconfigs/",
headers=crawler_auth_headers,

View File

@ -67,6 +67,28 @@ def test_get_configs_by_description(
assert config["description"] == description
def test_get_configs_by_schedule_true(crawler_auth_headers, default_org_id, crawler_crawl_id):
r = requests.get(
f"{API_PREFIX}/orgs/{default_org_id}/crawlconfigs?schedule=True",
headers=crawler_auth_headers
)
data = r.json()
assert data["total"] == 1
workflow = data["items"][0]
assert workflow.get("schedule") not in ("", None)
def test_get_configs_by_schedule_false(crawler_auth_headers, default_org_id, crawler_crawl_id):
r = requests.get(
f"{API_PREFIX}/orgs/{default_org_id}/crawlconfigs?schedule=False",
headers=crawler_auth_headers
)
data = r.json()
assert data["total"] >= 1
for config in data["items"]:
assert config.get("schedule") in ("", None)
def test_ensure_crawl_and_admin_user_crawls(
default_org_id, crawler_auth_headers, crawler_crawl_id, admin_crawl_id
):