Track CLI invocation source and full command

smart_tests/__main__.py

@@ -16,6 +16,7 @@
 from smart_tests.commands.subset import subset
 from smart_tests.commands.update import update
 from smart_tests.commands.verify import verify
+from smart_tests.utils.tracking import send_command_tracking
 
 cli = Group(name="cli", callback=Application)
 cli.add_command(record)
@@ -59,7 +60,18 @@ def _load_test_runners():
 
 
 def main():
-    cli.main()
+    argv = sys.argv[:]
+    exit_code = 0
+    try:
+        cli.main()
+    except SystemExit as e:
+        exit_code = e.code if isinstance(e.code, int) else 1
+    finally:
+        try:
+            send_command_tracking(argv=argv, exit_code=exit_code)
+        except Exception:
+            pass
+    sys.exit(exit_code)
 
 
 if __name__ == '__main__':

smart_tests/utils/commands.py

@@ -6,12 +6,19 @@ class Command(Enum):
     RECORD_TESTS = 'RECORD_TESTS'
     RECORD_BUILD = 'RECORD_BUILD'
     RECORD_SESSION = 'RECORD_SESSION'
+    RECORD_ATTACHMENT = 'RECORD_ATTACHMENT'
+    RECORD_DEPLOYMENT = 'RECORD_DEPLOYMENT'
     SUBSET = 'SUBSET'
     COMMIT = 'COMMIT'
     DETECT_FLAKE = 'DETECT_FLAKE'
     GATE = 'GATE'
     UPDATE_ALIAS = 'UPDATE_ALIAS'
-    RECORD_DEPLOYMENT = 'RECORD_DEPLOYMENT'
+    INSPECT_MODEL = 'INSPECT_MODEL'
+    INSPECT_SUBSET = 'INSPECT_SUBSET'
+    STATS_TEST_SESSIONS = 'STATS_TEST_SESSIONS'
+    COMPARE_SUBSETS = 'COMPARE_SUBSETS'
+    GET_DOCS = 'GET_DOCS'
+    UNKNOWN = 'UNKNOWN'
 
     # when you add a new constant here, the server also needs to get a new constant in cli_tracking.proto
 

smart_tests/utils/env_keys.py

@@ -9,6 +9,7 @@
 COMMIT_TIMEOUT = "SMART_TESTS_COMMIT_TIMEOUT"
 SKIP_CERT_VERIFICATION = "SMART_TESTS_SKIP_CERT_VERIFICATION"
 SESSION_DIR_KEY = "SMART_TESTS_SESSION_DIR"
+CALLER_KEY = "SMART_TESTS_CALLER"
 
 # Legacy token key for backward compatibility
 LEGACY_TOKEN_KEY = "LAUNCHABLE_TOKEN"
@@ -17,3 +18,15 @@
 def get_token():
     """Get token with backward compatibility for LAUNCHABLE_TOKEN."""
     return os.getenv(TOKEN_KEY) or os.getenv(LEGACY_TOKEN_KEY)
+
+
+def detect_ci_provider() -> str:
+    if os.environ.get("GITHUB_ACTIONS"):
+        return "github-actions"
+    if os.environ.get("JENKINS_URL"):
+        return "jenkins"
+    if os.environ.get("CIRCLECI"):
+        return "circleci"
+    if os.environ.get("CODEBUILD_BUILD_ID"):
+        return "codebuild"
+    return ""

smart_tests/utils/tracking.py

@@ -1,21 +1,74 @@
+import os
 from enum import Enum
+from itertools import takewhile
 from typing import Any, Dict, Union
 
 from requests import Session
 
 from smart_tests.app import Application
 from smart_tests.utils.authentication import get_org_workspace
+from smart_tests.utils.env_keys import CALLER_KEY, detect_ci_provider
 from smart_tests.utils.http_client import _HttpClient, _join_paths
 from smart_tests.version import __version__
 
 from .commands import Command
 
+# Map CLI subcommand tokens to Command enum values.
+# Longer matches are tried first so "record build" matches before "record".
+_COMMAND_MAP = {
+    ("verify",): Command.VERIFY,
+    ("record", "build"): Command.RECORD_BUILD,
+    ("record", "session"): Command.RECORD_SESSION,
+    ("record", "tests"): Command.RECORD_TESTS,
+    ("record", "commit"): Command.COMMIT,
+    ("record", "attachment"): Command.RECORD_ATTACHMENT,
+    ("record", "deployment"): Command.RECORD_DEPLOYMENT,
+    ("subset",): Command.SUBSET,
+    ("detect-flakes",): Command.DETECT_FLAKE,
+    ("gate",): Command.GATE,
+    ("update", "alias"): Command.UPDATE_ALIAS,
+    ("inspect", "model"): Command.INSPECT_MODEL,
+    ("inspect", "subset"): Command.INSPECT_SUBSET,
+    ("stats", "test_sessions"): Command.STATS_TEST_SESSIONS,
+    ("compare", "subsets"): Command.COMPARE_SUBSETS,
+    ("get", "docs"): Command.GET_DOCS,
+}
+
+
+def _detect_command(argv: list[str]) -> Command:
+    """Best-effort detection of the Command from argv. Returns UNKNOWN for typos."""
+    command_tokens = list(takewhile(lambda a: not a.startswith("-"), argv[1:]))
+
+    for tokens, command in sorted(_COMMAND_MAP.items(), key=lambda x: -len(x[0])):
+        if tuple(command_tokens[:len(tokens)]) == tokens:
+            return command
+    return Command.UNKNOWN
+
+
+def send_command_tracking(argv: list[str], exit_code: int):
+    """Send a single COMMAND_INVOCATION event with the full command string. Fire-and-forget."""
+    client = TrackingClient(_detect_command(argv))
+    metadata = {
+        "exitCode": str(exit_code),
+    }
+
+    raw_command = " ".join(argv)[:2000]
+
+    payload = client.construct_payload(
+        event_name=Tracking.Event.COMMAND_INVOCATION,
+        metadata=metadata,
+        raw_command=raw_command,
+    )
+
+    client.post_payload(payload=payload)
+
 
 class Tracking:
     # General events
     class Event(Enum):
         SHALLOW_CLONE = 'SHALLOW_CLONE'  # this event is an example
         PERFORMANCE = 'PERFORMANCE'
+        COMMAND_INVOCATION = 'COMMAND_INVOCATION'
 
     # Error events
     class ErrorEvent(Enum):
@@ -45,14 +98,8 @@ def send_event(
         event_name: Tracking.Event,
         metadata: Dict[str, Any] | None = None
     ):
-        org, workspace = get_org_workspace()
-        if metadata is None:
-            metadata = {}
-        metadata["organization"] = org or ""
-        metadata["workspace"] = workspace or ""
-        self._post_payload(
-            event_name=event_name,
-            metadata=metadata,
+        self.post_payload(
+            payload=self.construct_payload(event_name=event_name, metadata=metadata),
         )
 
     def send_error_event(
@@ -62,34 +109,49 @@ def send_error_event(
         api: str = "",
         metadata: Dict[str, Any] | None = None
     ):
-        org, workspace = get_org_workspace()
         if metadata is None:
             metadata = {}
         metadata["stackTrace"] = stack_trace
-        metadata["organization"] = org or ""
-        metadata["workspace"] = workspace or ""
         metadata["api"] = api
-        self._post_payload(
-            event_name=event_name,
-            metadata=metadata,
-        )
 
-    def _post_payload(
+        payload = self.construct_payload(event_name=event_name, metadata=metadata)
+        self.post_payload(payload=payload)
+
+    def post_payload(
         self,
-        event_name: Union[Tracking.Event, Tracking.ErrorEvent],
-        metadata: Dict[str, Any]
+        payload: dict,
     ):
-        payload = {
-            "command": self.command.value,
-            "eventName": event_name.value,
-            "cliVersion": __version__,
-            "metadata": metadata,
-        }
         path = _join_paths(
             '/intake',
             'cli_tracking'
         )
         try:
-            self.http_client.request('post', payload=payload, path=path)
+            self.http_client.request('post', payload=payload, path=path, timeout=(2, 2))
         except Exception:
             pass
+
+    def construct_payload(
+            self,
+            event_name: Union[Tracking.Event, Tracking.ErrorEvent],
+            metadata: Dict[str, Any] | None = None,
+            raw_command: str | None = None
+    ) -> dict:
+        org, workspace = get_org_workspace()
+
+        if metadata is None:
+            metadata = {}
+
+        metadata["organization"] = org or ""
+        metadata["workspace"] = workspace or ""
+        metadata["caller"] = os.environ.get(CALLER_KEY) or "cli"
+        metadata["ciProvider"] = detect_ci_provider()
+
+        payload = {
+            "command": self.command.value,
+            "eventName": event_name.value,
+            "cliVersion": __version__,
+            "metadata": metadata,
+            "rawCommand": raw_command,
+        }
+
+        return payload

tests/utils/test_http_client.py

@@ -10,7 +10,12 @@
 class HttpClientTest(TestCase):
     @mock.patch.dict(
         os.environ,
-        {"SMART_TESTS_ORGANIZATION": "launchableinc", "SMART_TESTS_WORKSPACE": "test"},
+        {
+            "SMART_TESTS_ORGANIZATION": "launchableinc",
+            "SMART_TESTS_WORKSPACE": "test",
+            "SMART_TESTS_TOKEN": "",
+            "LAUNCHABLE_TOKEN": "",
+        },
         clear=True,
     )
     def test_header(self):