@@ -501,6 +501,65 @@ def test_query_invalid_search_key(self) -> None:
501501 assert "query" in response .data
502502 assert "Invalid key for this search: tpe" in str (response .data ["query" ])
503503
504+ def test_query_by_workflow (self ) -> None :
505+ workflow = self .create_workflow (organization_id = self .organization .id )
506+ workflow_2 = self .create_workflow (organization_id = self .organization .id )
507+ detector_a = self .create_detector (
508+ project = self .project , name = "Detector A" , type = MetricIssue .slug
509+ )
510+ detector_b = self .create_detector (
511+ project = self .project , name = "Detector B" , type = MetricIssue .slug
512+ )
513+ self .create_detector (project = self .project , name = "Detector C" , type = MetricIssue .slug )
514+ self .create_detector_workflow (detector = detector_a , workflow = workflow )
515+ self .create_detector_workflow (detector = detector_b , workflow = workflow )
516+ self .create_detector_workflow (detector = detector_b , workflow = workflow_2 )
517+
518+ # Filter by single workflow
519+ response = self .get_success_response (
520+ self .organization .slug ,
521+ qs_params = {"project" : self .project .id , "query" : f"workflow:{ workflow .id } " },
522+ )
523+ assert {d ["name" ] for d in response .data } == {detector_a .name , detector_b .name }
524+
525+ # Filter by a different workflow
526+ response = self .get_success_response (
527+ self .organization .slug ,
528+ qs_params = {"project" : self .project .id , "query" : f"workflow:{ workflow_2 .id } " },
529+ )
530+ assert {d ["name" ] for d in response .data } == {detector_b .name }
531+
532+ # Filter by multiple workflows (IN)
533+ response = self .get_success_response (
534+ self .organization .slug ,
535+ qs_params = {
536+ "project" : self .project .id ,
537+ "query" : f"workflow:[{ workflow .id } , { workflow_2 .id } ]" ,
538+ },
539+ )
540+ assert {d ["name" ] for d in response .data } == {detector_a .name , detector_b .name }
541+
542+ # Negation
543+ response = self .get_success_response (
544+ self .organization .slug ,
545+ qs_params = {"project" : self .project .id , "query" : f"!workflow:{ workflow .id } " },
546+ )
547+ returned_names = {d ["name" ] for d in response .data }
548+ assert detector_a .name not in returned_names
549+ assert detector_b .name not in returned_names
550+
551+ # Negation with list (!IN)
552+ response = self .get_success_response (
553+ self .organization .slug ,
554+ qs_params = {
555+ "project" : self .project .id ,
556+ "query" : f"!workflow:[{ workflow .id } , { workflow_2 .id } ]" ,
557+ },
558+ )
559+ returned_names = {d ["name" ] for d in response .data }
560+ assert detector_a .name not in returned_names
561+ assert detector_b .name not in returned_names
562+
504563 def test_query_by_assignee_user_email (self ) -> None :
505564 user = self .create_user (email = "assignee@example.com" )
506565 self .create_member (organization = self .organization , user = user )
0 commit comments