Add 'gate' command

launchable/__main__.py

@@ -11,6 +11,7 @@
 
 from .commands.compare import compare
 from .commands.detect_flakes import detect_flakes
+from .commands.gate import gate
 from .commands.inspect import inspect
 from .commands.record import record
 from .commands.split_subset import split_subset
@@ -93,6 +94,7 @@ def main(ctx, log_level, plugin_dir, dry_run, skip_cert_verification):
 main.add_command(stats)
 main.add_command(compare)
 main.add_command(detect_flakes, "detect-flakes")
+main.add_command(gate)
 
 if __name__ == '__main__':
     main()

launchable/commands/gate.py

@@ -0,0 +1,101 @@
+import json
+import os
+import sys
+from http import HTTPStatus
+
+import click
+from requests import Response
+from tabulate import tabulate
+
+from launchable.commands.helper import find_or_create_session
+from launchable.utils.click import ignorable_error
+from launchable.utils.env_keys import REPORT_ERROR_KEY
+from launchable.utils.tracking import Tracking, TrackingClient
+
+from ..utils.commands import Command
+from ..utils.launchable_client import LaunchableClient
+
+
+@click.command()
+@click.option(
+    '--session',
+    'session',
+    help='In the format builds/<build-name>/test_sessions/<test-session-id>',
+    type=str,
+    required=True
+)
+@click.option(
+    '--json',
+    'is_json_format',
+    help='display JSON format',
+    is_flag=True
+)
+@click.pass_context
+def gate(ctx: click.core.Context, session: str, is_json_format: bool):
+    tracking_client = TrackingClient(Command.GATE, app=ctx.obj)
+    client = LaunchableClient(app=ctx.obj)
+    session_id = None
+    try:
+        session_id = find_or_create_session(
+            context=ctx,
+            session=session,
+            build_name=None,
+            tracking_client=tracking_client
+        )
+    except click.UsageError as e:
+        click.echo(click.style(str(e), fg="red"), err=True)
+        sys.exit(1)
+    except Exception as e:
+        tracking_client.send_error_event(
+            event_name=Tracking.ErrorEvent.INTERNAL_CLI_ERROR,
+            stack_trace=str(e),
+        )
+        if os.getenv(REPORT_ERROR_KEY):
+            raise e
+        else:
+            click.echo(ignorable_error(e), err=True)
+    if session_id is None:
+        return
+    try:
+        res: Response = client.request("get", "gate", params={"session-id": os.path.basename(session_id)})
+
+        if res.status_code == HTTPStatus.NOT_FOUND:
+            click.echo(click.style(
+                "Gate data currently not available for this workspace.", 'yellow'), err=True)
+            sys.exit()
+
+        res.raise_for_status()
+
+        res_json = res.json()
+
+        if is_json_format:
+            display_as_json(res)
+        else:
+            display_as_table(res)
+
+        # Exit with failure status if gate failed
+        if res_json.get('status') == 'FAILED':
+            sys.exit(1)
+
+    except Exception as e:
+        client.print_exception_and_recover(e, "Warning: failed to fetch gate status")
+
+
+def display_as_json(res: Response):
+    res_json = res.json()
+    click.echo(json.dumps(res_json, indent=2))
+
+
+def display_as_table(res: Response):
+    headers = ["Status", "Quarantined (Ignored)", "Actionable Failures"]
+    res_json = res.json()
+
+    status_icon = "PASSED" if res_json.get('status') == 'PASSED' else "FAILED"
+
+    rows = [[
+        status_icon,
+        res_json.get('quarantinedFailures', 0),
+        res_json.get('actionableFailures', 0)
+    ]]
+
+    click.echo(tabulate(rows, headers, tablefmt="github"))

launchable/utils/commands.py

@@ -9,6 +9,7 @@ class Command(Enum):
     SUBSET = 'SUBSET'
     COMMIT = 'COMMIT'
     DETECT_FLAKE = 'DETECT_FLAKE'
+    GATE = 'GATE'
 
     def display_name(self):
         return self.value.lower().replace('_', ' ')

tests/commands/test_gate.py

@@ -0,0 +1,126 @@
+import json
+import os
+from unittest import mock
+
+import responses
+
+from launchable.utils.http_client import get_base_url
+from tests.cli_test_case import CliTestCase
+
+
+class GateTest(CliTestCase):
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_gate_passed(self):
+        """Test gate command exits with 0 when status is PASSED"""
+        responses.add(
+            responses.GET,
+            "{}/intake/organizations/{}/workspaces/{}/gate".format(
+                get_base_url(),
+                self.organization,
+                self.workspace),
+            json={
+                'status': 'PASSED',
+                'quarantinedFailures': 5,
+                'actionableFailures': 0
+            },
+            status=200)
+
+        result = self.cli('gate', '--session', self.session)
+        self.assert_success(result)
+        self.assertIn('PASSED', result.output)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_gate_failed(self):
+        """Test gate command exits with 1 when status is FAILED"""
+        responses.add(
+            responses.GET,
+            "{}/intake/organizations/{}/workspaces/{}/gate".format(
+                get_base_url(),
+                self.organization,
+                self.workspace),
+            json={
+                'status': 'FAILED',
+                'quarantinedFailures': 2,
+                'actionableFailures': 3
+            },
+            status=200)
+
+        result = self.cli('gate', '--session', self.session)
+        self.assert_exit_code(result, 1)
+        self.assertIn('FAILED', result.output)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_gate_passed_json_format(self):
+        """Test gate command with --json flag when status is PASSED"""
+        gate_data = {
+            'status': 'PASSED',
+            'quarantinedFailures': 5,
+            'actionableFailures': 0
+        }
+
+        responses.add(
+            responses.GET,
+            "{}/intake/organizations/{}/workspaces/{}/gate".format(
+                get_base_url(),
+                self.organization,
+                self.workspace),
+            json=gate_data,
+            status=200)
+
+        result = self.cli('gate', '--session', self.session, '--json')
+        self.assert_success(result)
+
+        # Verify JSON output
+        output_json = json.loads(result.output)
+        self.assertEqual(output_json['status'], 'PASSED')
+        self.assertEqual(output_json['quarantinedFailures'], 5)
+        self.assertEqual(output_json['actionableFailures'], 0)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_gate_failed_json_format(self):
+        """Test gate command with --json flag when status is FAILED"""
+        gate_data = {
+            'status': 'FAILED',
+            'quarantinedFailures': 2,
+            'actionableFailures': 3
+        }
+
+        responses.add(
+            responses.GET,
+            "{}/intake/organizations/{}/workspaces/{}/gate".format(
+                get_base_url(),
+                self.organization,
+                self.workspace),
+            json=gate_data,
+            status=200)
+
+        result = self.cli('gate', '--session', self.session, '--json')
+        self.assert_exit_code(result, 1)
+
+        # Verify JSON output
+        output_json = json.loads(result.output)
+        self.assertEqual(output_json['status'], 'FAILED')
+        self.assertEqual(output_json['quarantinedFailures'], 2)
+        self.assertEqual(output_json['actionableFailures'], 3)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_gate_not_found(self):
+        """Test gate command when gate data is not available"""
+        responses.add(
+            responses.GET,
+            "{}/intake/organizations/{}/workspaces/{}/gate".format(
+                get_base_url(),
+                self.organization,
+                self.workspace),
+            json={},
+            status=404)
+
+        result = self.cli('gate', '--session', self.session)
+        # Should exit with 0 when gate data is not available (non-error case)
+        self.assert_success(result)
+        self.assertIn('Gate data currently not available', result.output)