18670993611

Committed 20 Oct 2025 04:39PM UTC coverage: 86.897% (+0.001%) from 86.896%

Build # 18670993611

Build Type

push

github

Committed by

web-flow

Commit Message

fix json.assign_to_path with non nested path (#13245)

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

9 existing lines in 3 files now uncovered.

68348 of 78654 relevant lines covered (86.9%)

0.87 hits per line

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

75.44

/localstack-core/localstack/testing/pytest/in_memory_localstack.py

"""Pytest plugin that spins up a single localstack instance in the current interpreter that is shared
across the current test session.

Use in your module as follows::

    pytest_plugins = "localstack.testing.pytest.in_memory_localstack"

    @pytest.hookimpl()
    def pytest_configure(config):
        config.option.start_localstack = True

You can explicitly disable starting localstack by setting ``TEST_SKIP_LOCALSTACK_START=1`` or
``TEST_TARGET=AWS_CLOUD``."""

import logging
import os
import threading

import pytest
from _pytest.config import PytestPluginManager
from _pytest.config.argparsing import Parser
from _pytest.main import Session

from localstack import config as localstack_config
from localstack.config import is_env_true
from localstack.constants import ENV_INTERNAL_TEST_RUN

LOG = logging.getLogger(__name__)
LOG.info("Pytest plugin for in-memory-localstack session loaded.")

if localstack_config.is_collect_metrics_mode():
    pytest_plugins = "localstack.testing.pytest.metric_collection"

_started = threading.Event()


def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager):
    parser.addoption(
        "--start-localstack",
        action="store_true",
        default=False,
    )


@pytest.hookimpl(tryfirst=True)
def pytest_runtestloop(session: Session):
    # avoid starting up localstack if we only collect the tests (-co / --collect-only)
    if session.config.option.collectonly:
        return

    if not session.config.option.start_localstack:
        return

    from localstack.testing.aws.util import is_aws_cloud

    if is_env_true("TEST_SKIP_LOCALSTACK_START"):
        LOG.info("TEST_SKIP_LOCALSTACK_START is set, not starting localstack")
        return

    if is_aws_cloud():
        if not is_env_true("TEST_FORCE_LOCALSTACK_START"):
            LOG.info("Test running against aws, not starting localstack")
            return
        LOG.info("TEST_FORCE_LOCALSTACK_START is set, a Localstack instance will be created.")

    if is_aws_cloud():
        localstack_config.DEFAULT_DELAY = 5
        localstack_config.DEFAULT_MAX_ATTEMPTS = 60

    # configure
    os.environ[ENV_INTERNAL_TEST_RUN] = "1"
    localstack_config.INCLUDE_STACK_TRACES_IN_HTTP_RESPONSE = True

    from localstack.runtime import current

    _started.set()
    runtime = current.initialize_runtime()
    # start runtime asynchronously
    threading.Thread(target=runtime.run).start()

    # wait for runtime to be ready
    if not runtime.ready.wait(timeout=120):
        raise TimeoutError("gave up waiting for runtime to be ready")


@pytest.hookimpl(trylast=True)
def pytest_sessionfinish(session: Session):
    # last pytest lifecycle hook (before pytest exits)
    if not _started.is_set():
        return

    from localstack.runtime import get_current_runtime

    try:
        get_current_runtime()
    except ValueError:
        LOG.warning("Could not access the current runtime in a pytest sessionfinish hook.")
        return

    get_current_runtime().shutdown()
    LOG.info("waiting for runtime to stop")

    # wait for runtime to shut down
    if not get_current_runtime().stopped.wait(timeout=20):
        LOG.warning("gave up waiting for runtime to stop, returning anyway")

1	"""Pytest plugin that spins up a single localstack instance in the current interpreter that is shared
2	across the current test session.
3
4	Use in your module as follows::
5
6	pytest_plugins = "localstack.testing.pytest.in_memory_localstack"
7
8	@pytest.hookimpl()
9	def pytest_configure(config):
10	config.option.start_localstack = True
11
12	You can explicitly disable starting localstack by setting ``TEST_SKIP_LOCALSTACK_START=1`` or
13	``TEST_TARGET=AWS_CLOUD``."""
14
15	import logging	1✔
16	import os	1✔
17	import threading	1✔
18
19	import pytest	1✔
20	from _pytest.config import PytestPluginManager	1✔
21	from _pytest.config.argparsing import Parser	1✔
22	from _pytest.main import Session	1✔
23
24	from localstack import config as localstack_config	1✔
25	from localstack.config import is_env_true	1✔
26	from localstack.constants import ENV_INTERNAL_TEST_RUN	1✔
27
28	LOG = logging.getLogger(__name__)	1✔
29	LOG.info("Pytest plugin for in-memory-localstack session loaded.")	1✔
30
31	if localstack_config.is_collect_metrics_mode():	1✔
32	pytest_plugins = "localstack.testing.pytest.metric_collection"	1✔
33
34	_started = threading.Event()	1✔
35
36
37	def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager):	1✔
38	parser.addoption(	1✔
39	"--start-localstack",
40	action="store_true",
41	default=False,
42	)
43
44
45	@pytest.hookimpl(tryfirst=True)	1✔
46	def pytest_runtestloop(session: Session):	1✔
47	# avoid starting up localstack if we only collect the tests (-co / --collect-only)
48	if session.config.option.collectonly:	1✔
49	return	×
50
51	if not session.config.option.start_localstack:	1✔
52	return	1✔
53
54	from localstack.testing.aws.util import is_aws_cloud	1✔
55
56	if is_env_true("TEST_SKIP_LOCALSTACK_START"):	1✔
57	LOG.info("TEST_SKIP_LOCALSTACK_START is set, not starting localstack")	×
58	return	×
59
60	if is_aws_cloud():	1✔
61	if not is_env_true("TEST_FORCE_LOCALSTACK_START"):	×
62	LOG.info("Test running against aws, not starting localstack")	×
63	return	×
64	LOG.info("TEST_FORCE_LOCALSTACK_START is set, a Localstack instance will be created.")	×
65
66	if is_aws_cloud():	1✔
UNCOV 67	localstack_config.DEFAULT_DELAY = 5	×
UNCOV 68	localstack_config.DEFAULT_MAX_ATTEMPTS = 60	×
69
70	# configure
71	os.environ[ENV_INTERNAL_TEST_RUN] = "1"	1✔
72	localstack_config.INCLUDE_STACK_TRACES_IN_HTTP_RESPONSE = True	1✔
73
74	from localstack.runtime import current	1✔
75
76	_started.set()	1✔
77	runtime = current.initialize_runtime()	1✔
78	# start runtime asynchronously
79	threading.Thread(target=runtime.run).start()	1✔
80
81	# wait for runtime to be ready
82	if not runtime.ready.wait(timeout=120):	1✔
UNCOV 83	raise TimeoutError("gave up waiting for runtime to be ready")	×
84
85
86	@pytest.hookimpl(trylast=True)	1✔
87	def pytest_sessionfinish(session: Session):	1✔
88	# last pytest lifecycle hook (before pytest exits)
89	if not _started.is_set():	1✔
90	return	1✔
91
92	from localstack.runtime import get_current_runtime	1✔
93
94	try:	1✔
95	get_current_runtime()	1✔
UNCOV 96	except ValueError:	×
UNCOV 97	LOG.warning("Could not access the current runtime in a pytest sessionfinish hook.")	×
UNCOV 98	return	×
99
100	get_current_runtime().shutdown()	1✔
101	LOG.info("waiting for runtime to stop")	1✔
102
103	# wait for runtime to shut down
104	if not get_current_runtime().stopped.wait(timeout=20):	1✔
UNCOV 105	LOG.warning("gave up waiting for runtime to stop, returning anyway")	×

localstack / localstack / 18670993611

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