16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.18

/localstack-core/localstack/utils/analytics/cli.py

import datetime
import functools
from multiprocessing import Process

import click

from localstack import config

from .client import AnalyticsClient
from .events import Event, EventMetadata
from .metadata import get_session_id
from .publisher import AnalyticsClientPublisher

ANALYTICS_API_RESPONSE_TIMEOUT_SECS = 0.5


def _publish_cmd_as_analytics_event(command_name: str, params: list[str]):
    event = Event(
        name="cli_cmd",
        payload={"cmd": command_name, "params": params},
        metadata=EventMetadata(
            session_id=get_session_id(),
            client_time=str(datetime.datetime.now()),  # TODO: consider using utcnow()
        ),
    )
    publisher = AnalyticsClientPublisher(AnalyticsClient())
    publisher.publish([event])


def _get_parent_commands(ctx: click.Context) -> list[str]:
    parent_commands = []
    parent = ctx.parent
    while parent is not None:
        parent_commands.insert(0, parent.command.name)
        parent = parent.parent
    return parent_commands


def publish_invocation(fn):
    """
    Decorator for capturing CLI commands from Click and publishing them to the backend as analytics events.
    This decorator should only be used on outermost subcommands, e.g. "localstack status docker" not "localstack status"
    otherwise it may publish multiple events for a single invocation.
    If DISABLE_EVENTS is set then nothing is collected.
    For performance reasons, the API call to the backend runs on a separate process and is killed if it takes longer
    than ANALYTICS_API_RESPONSE_TIMEOUT_SECS.
    The emitted event contains the invoked command, plus any parameter names if their associated values are truthy (but
    not the values themselves).
    """

    @functools.wraps(fn)
    def publisher_wrapper(*args, **kwargs):
        if config.DISABLE_EVENTS:
            return fn(*args, **kwargs)

        ctx = click.get_current_context()
        full_command = " ".join(_get_parent_commands(ctx) + [ctx.command.name])
        publish_cmd_process = Process(
            target=_publish_cmd_as_analytics_event,
            args=(full_command, [k for k, v in ctx.params.items() if v]),
        )
        publish_cmd_process.start()
        publish_cmd_process.join(ANALYTICS_API_RESPONSE_TIMEOUT_SECS)
        publish_cmd_process.terminate()
        return fn(*args, **kwargs)

    return publisher_wrapper

1	import datetime	1✔
2	import functools	1✔
3	from multiprocessing import Process	1✔
4
5	import click	1✔
6
7	from localstack import config	1✔
8
9	from .client import AnalyticsClient	1✔
10	from .events import Event, EventMetadata	1✔
11	from .metadata import get_session_id	1✔
12	from .publisher import AnalyticsClientPublisher	1✔
13
14	ANALYTICS_API_RESPONSE_TIMEOUT_SECS = 0.5	1✔
15
16
17	def _publish_cmd_as_analytics_event(command_name: str, params: list[str]):	1✔
UNCOV 18	event = Event(	×
19	name="cli_cmd",
20	payload={"cmd": command_name, "params": params},
21	metadata=EventMetadata(
22	session_id=get_session_id(),
23	client_time=str(datetime.datetime.now()), # TODO: consider using utcnow()
24	),
25	)
UNCOV 26	publisher = AnalyticsClientPublisher(AnalyticsClient())	×
27	publisher.publish([event])	×
28
29
30	def _get_parent_commands(ctx: click.Context) -> list[str]:	1✔
31	parent_commands = []	1✔
32	parent = ctx.parent	1✔
33	while parent is not None:	1✔
34	parent_commands.insert(0, parent.command.name)	1✔
35	parent = parent.parent	1✔
36	return parent_commands	1✔
37
38
39	def publish_invocation(fn):	1✔
40	"""
41	Decorator for capturing CLI commands from Click and publishing them to the backend as analytics events.
42	This decorator should only be used on outermost subcommands, e.g. "localstack status docker" not "localstack status"
43	otherwise it may publish multiple events for a single invocation.
44	If DISABLE_EVENTS is set then nothing is collected.
45	For performance reasons, the API call to the backend runs on a separate process and is killed if it takes longer
46	than ANALYTICS_API_RESPONSE_TIMEOUT_SECS.
47	The emitted event contains the invoked command, plus any parameter names if their associated values are truthy (but
48	not the values themselves).
49	"""
50
51	@functools.wraps(fn)	1✔
52	def publisher_wrapper(args, *kwargs):	1✔
53	if config.DISABLE_EVENTS:	1✔
54	return fn(args, *kwargs)	1✔
55
56	ctx = click.get_current_context()	1✔
57	full_command = " ".join(_get_parent_commands(ctx) + [ctx.command.name])	1✔
58	publish_cmd_process = Process(	1✔
59	target=_publish_cmd_as_analytics_event,
60	args=(full_command, [k for k, v in ctx.params.items() if v]),
61	)
62	publish_cmd_process.start()	1✔
63	publish_cmd_process.join(ANALYTICS_API_RESPONSE_TIMEOUT_SECS)	1✔
64	publish_cmd_process.terminate()	1✔
65	return fn(args, *kwargs)	1✔
66
67	return publisher_wrapper	1✔

localstack / localstack / 16820655284

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous