import base64

import io

import itertools

import os

import time

import zipfile

from datetime import UTC, date, datetime, timedelta

from zoneinfo import ZoneInfo

import pytest

import yaml

from localstack import config

from localstack.utils import common

from localstack.utils.archives import unzip

from localstack.utils.common import (

ExternalServicePortsManager,

PaginatedList,

PortNotAvailableException,

fully_qualified_class_name,

get_free_tcp_port,

is_empty_dir,

load_file,

mkdir,

new_tmp_dir,

rm_rf,

save_file,

short_uid,

)

from localstack.utils.objects import Mock

from localstack.utils.strings import base64_decode, to_bytes

from localstack.utils.testutil import create_zip_file

class TestCommon:

def test_first_char_to_lower(self):

env = common.first_char_to_lower("Foobar")

assert env == "foobar"

def test_truncate(self):

env = common.truncate("foobar", 3)

assert env == "foo..."

def test_isoformat_milliseconds(self):

env = common.isoformat_milliseconds(datetime(2010, 3, 20, 7, 24, 00, 0))

assert env == "2010-03-20T07:24:00.000"

@pytest.mark.parametrize(

"str_format",

[

"2025-09-18T14:07:30",

"2025-09-18T14:07:30Z",

"2025-09-18T14:07:30.70300Z",

"18/Sep/2025:14:07:30 +0000",

],

)

def test_parse_timestamp_timezone_aware(self, str_format):

datetime_obj = common.parse_timestamp(str_format)

# we cannot assert that tzinfo is `datetime.UTC` because it is only supported starting Python 3.11

# so we assert manually that the returned `tzinfo` is UTC

# we are using ZoneInfo("UTC") in the `parse_timestamp` utility as it is in the import path of the CLI

assert datetime_obj.tzinfo is not None

assert datetime_obj.utcoffset() == timedelta(0)

def test_base64_to_hex(self):

env = common.base64_to_hex("Zm9vIGJhcg ==")

assert env == b"666f6f20626172"

def test_base64_decode(self):

def roundtrip(data):

encoded = base64.urlsafe_b64encode(to_bytes(data))

result = base64_decode(encoded)

assert to_bytes(data) == result

# simple examples

roundtrip("test")

roundtrip(b"test \x64 \x01 \x55")

# strings that require urlsafe encoding (containing "-" or "/" in base64 encoded form)

examples = ((b"=@~", b"PUB+"), (b"???", b"Pz8/"))

for decoded, encoded in examples:

assert base64.b64encode(decoded) == encoded

expected = encoded.replace(b"+", b"-").replace(b"/", b"_")

assert base64.urlsafe_b64encode(decoded) == expected

roundtrip(decoded)

def test_now(self):

env = common.now()

test = time.time()

assert test == pytest.approx(env, 1)

def test_now_utc(self):

env = common.now_utc()

test = datetime.now(UTC).timestamp()

assert test == pytest.approx(env, 1)

@pytest.mark.parametrize(

"value,is_number",

[

(5, True),

(-12.1, True),

(2e15, True),

("test", False),

(False, False),

(True, False),

(None, False),

],

)

def test_is_number(self, value, is_number):

assert common.is_number(value) == is_number

def test_is_ip_address(self):

assert common.is_ip_address("10.0.0.1")

assert not common.is_ip_address("abcde")

def test_is_base64(self):

assert not common.is_base64("foobar")

def test_mktime(self):

now = common.mktime(datetime.now())

assert int(now) == int(time.time())

def test_mktime_with_tz(self):

# see https://en.wikipedia.org/wiki/File:1000000000seconds.jpg

dt = datetime(2001, 9, 9, 1, 46, 40, 0, tzinfo=UTC)

assert int(common.mktime(dt)) == 1000000000

dt = datetime(2001, 9, 9, 1, 46, 40, 0, tzinfo=ZoneInfo("EST"))

assert int(common.mktime(dt)) == 1000000000 + (5 * 60 * 60) # EST is UTC-5

def test_mktime_millis_with_tz(self):

# see https://en.wikipedia.org/wiki/File:1000000000

dt = datetime(2001, 9, 9, 1, 46, 40, 0, tzinfo=UTC)

assert int(common.mktime(dt, millis=True) / 1000) == 1000000000

dt = datetime(2001, 9, 9, 1, 46, 40, 0, tzinfo=ZoneInfo("EST"))

assert int(common.mktime(dt, millis=True)) / 1000 == 1000000000 + (

5 * 60 * 60

) # EST is UTC-5

def test_mktime_millis(self):

now = common.mktime(datetime.now(), millis=True)

assert int(time.time()) == int(now / 1000)

def test_timestamp_millis(self):

result = common.timestamp_millis(datetime.now())

assert "T" in result

result = common.timestamp_millis(date.today())

assert "00:00:00" in result

assert "T" in result

def test_extract_jsonpath(self):

obj = {"a": {"b": [{"c": 123}, "foo"]}, "e": 234}

result = common.extract_jsonpath(obj, "$.a.b")

assert result == [{"c": 123}, "foo"]

result = common.extract_jsonpath(obj, "$.a.b.c")

assert not result

result = common.extract_jsonpath(obj, "$.foobar")

assert not result

result = common.extract_jsonpath(obj, "$.e")

assert result == 234

result = common.extract_jsonpath(obj, "$.a.b[0]")

assert result == {"c": 123}

result = common.extract_jsonpath(obj, "$.a.b[0].c")

assert result == 123

result = common.extract_jsonpath(obj, "$.a.b[1]")

assert result == "foo"

def test_str_to_bool(self):

assert common.str_to_bool("true") is True

assert common.str_to_bool("True") is True

assert common.str_to_bool("1") is False

assert common.str_to_bool("0") is False

assert common.str_to_bool("TRUE") is False

assert common.str_to_bool("false") is False

assert common.str_to_bool("False") is False

assert common.str_to_bool(0) == 0

def test_parse_yaml_nodes(self):

obj = {"test": yaml.ScalarNode("tag:yaml.org,2002:int", "123")}

result = common.clone_safe(obj)

assert result == {"test": 123}

obj = {

"foo": [

yaml.ScalarNode("tag:yaml.org,2002:str", "value"),

yaml.ScalarNode("tag:yaml.org,2002:int", "123"),

yaml.ScalarNode("tag:yaml.org,2002:float", "1.23"),

yaml.ScalarNode("tag:yaml.org,2002:bool", "true"),

]

}

result = common.clone_safe(obj)

assert result == {"foo": ["value", 123, 1.23, True]}

def test_to_unique_item_list(self):

assert common.to_unique_items_list([1, 1, 2, 2, 3]) == [1, 2, 3]

assert common.to_unique_items_list(["a"]) == ["a"]

assert common.to_unique_items_list(["a", "b", "a"]) == ["a", "b"]

assert common.to_unique_items_list("aba") == ["a", "b"]

assert common.to_unique_items_list([]) == []

def comparator_lower(first, second):

return first.lower() == second.lower()

assert common.to_unique_items_list(["a", "A", "a"]) == ["a", "A"]

assert common.to_unique_items_list(["a", "A", "a"], comparator_lower) == ["a"]

assert common.to_unique_items_list(["a", "A", "a"], comparator_lower) == ["a"]

def comparator_str_int(first, second):

return int(first) - int(second)

assert common.to_unique_items_list(["1", "2", "1", "2"], comparator_str_int) == ["1", "2"]

def test_retry(self):

exceptions = []

count = itertools.count()

def fn():

i = next(count)

e = RuntimeError(f"exception {i:d}")

exceptions.append(e)

if i == 2:

return "two"

raise e

ret = common.retry(fn, retries=3, sleep=0.001)

assert ret == "two"

assert len(exceptions) == 3

def test_retry_raises_last_exception(self):

exceptions = []

count = itertools.count()

def fn():

i = next(count)

e = RuntimeError(f"exception {i:d}")

exceptions.append(e)

raise e

with pytest.raises(RuntimeError) as ctx:

common.retry(fn, retries=3, sleep=0.001)

assert exceptions[-1] is ctx.value

assert len(exceptions) == 4

def test_run(self):

cmd = "echo 'foobar'"

result = common.run(cmd)

assert result.strip() == "foobar"

def test_is_command_available(self):

assert common.is_command_available("python3")

assert not common.is_command_available("hopefullydoesntexist")

def test_camel_to_snake_case(self):

fn = common.camel_to_snake_case

assert fn("Foo") == "foo"

assert fn("FoobarEd") == "foobar_ed"

assert fn("FooBarEd") == "foo_bar_ed"

assert fn("Foo_Bar") == "foo_bar"

assert fn("Foo__Bar") == "foo__bar"

assert fn("FooBAR") == "foo_bar"

assert fn("HTTPRequest") == "http_request"

assert fn("HTTP_Request") == "http_request"

assert fn("VerifyHTTPRequest") == "verify_http_request"

assert fn("IsHTTP") == "is_http"

assert fn("IsHTTP2Request") == "is_http2_request"

def test_snake_to_camel_case(self):

fn = common.snake_to_camel_case

assert fn("foo") == "Foo"

assert fn("foobar_ed") == "FoobarEd"

assert fn("foo_bar_ed") == "FooBarEd"

assert fn("foo_bar") == "FooBar"

assert fn("foo__bar") == "FooBar"

assert fn("foo_b_a_r") == "FooBAR"

def test_obj_to_xml(self):

fn = common.obj_to_xml

# primitive

assert fn(42) == "42"

assert fn(False) == "False"

assert fn("a") == "a"

# dict only

assert fn({"foo": "bar"}) == "<foo>bar</foo>"

assert fn({"a": 42}) == "<a>42</a>"

assert fn({"a": 42, "foo": "bar"}) == "<a>42</a><foo>bar</foo>"

# list of dicts

assert fn([{"a": 42}, {"a": 43}]) == "<a>42</a><a>43</a>"

# dict with lists

assert fn({"f": [{"a": 42}, {"a": 43}]}) == "<f><a>42</a><a>43</a></f>"

# empty types

assert fn(None) == "None"

assert fn("") == ""

def test_parse_json_or_yaml_with_json(self):

markup = """{"foo": "bar", "why": 42, "mylist": [1,2,3]}"""

doc = common.parse_json_or_yaml(markup)

assert doc == {"foo": "bar", "why": 42, "mylist": [1, 2, 3]}

def test_parse_json_or_yaml_with_yaml(self):

markup = """

foo: bar

why: 42

mylist:

- 1

- 2

- 3

"""

doc = common.parse_json_or_yaml(markup)

assert doc == {"foo": "bar", "why": 42, "mylist": [1, 2, 3]}

def test_parse_json_or_yaml_with_invalid_syntax_returns_content(self):

markup = "<a href='foobar'>baz</a>"

doc = common.parse_json_or_yaml(markup)

assert doc == markup # FIXME: not sure if this is good behavior

def test_parse_json_or_yaml_with_empty_string_returns_none(self):

doc = common.parse_json_or_yaml("")

assert doc is None

def test_format_bytes(self):

fn = common.format_bytes

assert fn(1) == "1B"

assert fn(100) == "100B"

assert fn(999) == "999B"

assert fn(1e3) == "1KB"

assert fn(1e6) == "1MB"

assert fn(1e7) == "10MB"

assert fn(1e8) == "100MB"

assert fn(1e9) == "1GB"

assert fn(1e12) == "1TB"

# comma values

assert fn(1e12 + 1e11) == "1.1TB"

assert fn(1e15) == "1000TB"

# string input

assert fn("123") == "123B"

# invalid number

assert fn("abc") == "n/a"

# negative number

assert fn(-1) == "n/a" # TODO: seems we could support this case

def test_format_number(self):

fn = common.format_number

assert fn(12, decimals=0) == "12"

assert fn(12, decimals=1) == "12"

assert fn(12.421, decimals=0) == "12"

assert fn(12.521, decimals=0) == "13"

assert fn(12.521, decimals=2) == "12.52"

assert fn(12.521, decimals=3) == "12.521"

assert fn(12.521, decimals=4) == "12.521"

assert fn(-12.521, decimals=4) == "-12.521"

assert fn(-1.2234354123e3, decimals=4) == "-1223.4354"

def test_proxy_map(self):

old_http_proxy = config.OUTBOUND_HTTP_PROXY

old_https_proxy = config.OUTBOUND_HTTPS_PROXY

config.OUTBOUND_HTTP_PROXY = "http://localhost"

config.OUTBOUND_HTTPS_PROXY = "https://localhost"

assert common.get_proxies() == {

"http": config.OUTBOUND_HTTP_PROXY,

"https": config.OUTBOUND_HTTPS_PROXY,

}

config.OUTBOUND_HTTP_PROXY = ""

assert common.get_proxies() == {"https": config.OUTBOUND_HTTPS_PROXY}

config.OUTBOUND_HTTPS_PROXY = ""

assert common.get_proxies() == {}

config.OUTBOUND_HTTP_PROXY = old_http_proxy

config.OUTBOUND_HTTPS_PROXY = old_https_proxy

def test_fully_qualified_class_name(self):

assert fully_qualified_class_name(Mock) == "localstack.utils.objects.Mock"

class TestCommonFileOperations:

def test_disk_usage(self, tmp_path):

f1 = tmp_path / "f1.blob"

f1.write_bytes(b"0" * 100)

f2 = tmp_path / "f2.blob"

f2.write_bytes(b"0" * 100)

# subdir

f3_dir = tmp_path / "foo"

f3_dir.mkdir()

f3 = f3_dir / "f3.blob"

f3.write_bytes(b"0" * 100)

# trees

assert common.disk_usage(tmp_path) == pytest.approx(300, abs=5)

assert common.disk_usage(f3_dir) == pytest.approx(100, abs=5)

# single file

assert common.disk_usage(f3) == pytest.approx(100, abs=5)

# invalid path

assert common.disk_usage(tmp_path / "not_in_path") == 0

# None

with pytest.raises(TypeError):

assert common.disk_usage(None) == 0

def test_replace_in_file(self, tmp_path):

content = """

1: {search}

2: {search}

3: {sear}

"""

expected = """

1: foo

2: foo

3: {sear}

"""

fp = tmp_path / "file.txt"

fp.write_text(content)

common.replace_in_file("{search}", "foo", fp)

assert fp.read_text() == expected

# try again, nothing should change

common.replace_in_file("{search}", "foo", fp)

assert fp.read_text() == expected

def test_replace_in_file_with_non_existing_path(self, tmp_path):

fp = tmp_path / "non_existing_file.txt"

assert not fp.exists()

common.replace_in_file("foo", "bar", fp)

assert not fp.exists()

def test_cp_r(self, tmp_path):

source = tmp_path / "source"

target = tmp_path / "target"

f1 = source / "f1.txt"

f2 = source / "d1" / "f2.txt"

f3 = source / "d1" / "d2" / "f3.txt"

source.mkdir()

target.mkdir()

f3.parent.mkdir(parents=True)

f1.write_text("f1")

f2.write_text("f2")

f3.write_text("f3")

common.cp_r(source, target)

assert (target / "f1.txt").is_file()

assert (target / "d1" / "f2.txt").is_file()

assert (target / "d1" / "f2.txt").is_file()

assert (target / "d1" / "d2" / "f3.txt").is_file()

assert (target / "d1" / "d2" / "f3.txt").read_text() == "f3"

def test_is_dir_empty(self):

tmp_dir = new_tmp_dir()

assert is_empty_dir(tmp_dir)

def _check(fname, is_dir):

test_entry = os.path.join(tmp_dir, fname)

mkdir(test_entry) if is_dir else save_file(test_entry, "test content")

assert not is_empty_dir(tmp_dir)

assert is_empty_dir(tmp_dir, ignore_hidden=True) == (fname == ".hidden")

rm_rf(test_entry)

assert is_empty_dir(tmp_dir)

for name in ["regular", ".hidden"]:

for is_dir in [True, False]:

_check(name, is_dir)

def test_create_archive(self):

# create archive from empty directory

tmp_dir = new_tmp_dir()

content = create_zip_file(tmp_dir, get_content=True)

zip_obj = zipfile.ZipFile(io.BytesIO(content))

assert zip_obj.infolist() == []

rm_rf(tmp_dir)

# create archive from non-empty directory

tmp_dir = new_tmp_dir()

save_file(os.path.join(tmp_dir, "testfile"), "content 123")

content = create_zip_file(tmp_dir, get_content=True)

zip_obj = zipfile.ZipFile(io.BytesIO(content))

assert len(zip_obj.infolist()) == 1

assert zip_obj.infolist()[0].filename == "testfile"

rm_rf(tmp_dir)

def test_unzip_bad_crc(self):

"""Test unzipping of files with incorrect CRC codes - usually works with native `unzip` command,

but seems to fail with zipfile module under certain Python versions (extracts 0-bytes files)

"""

# base64-encoded zip file with a single entry with incorrect CRC (created by Node.js 18 / Serverless)

zip_base64 = """

UEsDBBQAAAAIAAAAIQAAAAAAJwAAAAAAAAAjAAAAbm9kZWpzL25vZGVfbW9kdWxlcy9sb2Rhc2gvaW5k

ZXguanPLzU8pzUnVS60oyC8qKVawVShKLSzNLErVUNfTz8lPSSzOUNe0BgBQSwECLQMUAAAACAAAACEA

AAAAACcAAAAAAAAAIwAAAAAAAAAAACAApIEAAAAAbm9kZWpzL25vZGVfbW9kdWxlcy9sb2Rhc2gvaW5k

ZXguanNQSwUGAAAAAAEAAQBRAAAAaAAAAAAA

"""

tmp_dir = new_tmp_dir()

zip_file = os.path.join(tmp_dir, "test.zip")

save_file(zip_file, base64.b64decode(zip_base64))

unzip(zip_file, tmp_dir)

content = load_file(os.path.join(tmp_dir, "nodejs", "node_modules", "lodash", "index.js"))

assert content.strip() == "module.exports = require('./lodash');"

rm_rf(tmp_dir)

def test_save_load_file(tmp_path):

file_name = tmp_path / (f"normal_permissions_{short_uid()}")

content = f"some_content_{short_uid()}"

more_content = f"some_more_content_{short_uid()}"

save_file(file_name, content)

assert content == load_file(file_name)

save_file(file_name, more_content, append=True)

assert content + more_content == load_file(file_name)

def test_load_file_strict(tmp_path):

file_name = tmp_path / short_uid()

assert not os.path.isfile(file_name)

assert load_file(file_name) is None

with pytest.raises(FileNotFoundError):

load_file(file_name, strict=True)

def test_save_load_file_with_permissions(tmp_path):

file_name = tmp_path / (f"special_permissions_{short_uid()}")

content = f"some_content_{short_uid()}"

more_content = f"some_more_content_{short_uid()}"

permissions = 0o600

save_file(file_name, content, permissions=permissions)

assert permissions == os.stat(file_name).st_mode & 0o777

assert content == load_file(file_name)

save_file(file_name, more_content, append=True)

assert permissions == os.stat(file_name).st_mode & 0o777

assert content + more_content == load_file(file_name)

def test_save_load_file_with_changing_permissions(tmp_path):

file_name = tmp_path / (f"changing_permissions_{short_uid()}")

content = f"some_content_{short_uid()}"

more_content = f"some_more_content_{short_uid()}"

permissions = 0o600

save_file(file_name, content)

assert permissions != os.stat(file_name).st_mode & 0o777

assert content == load_file(file_name)

# setting the permissions on append should not change the permissions

save_file(file_name, more_content, append=True, permissions=permissions)

assert permissions != os.stat(file_name).st_mode & 0o777

assert content + more_content == load_file(file_name)

# overwriting the file also will not change the permissions

save_file(file_name, content, permissions=permissions)

assert permissions != os.stat(file_name).st_mode & 0o777

assert content == load_file(file_name)

@pytest.fixture()

def external_service_ports_manager():

previous_start = config.EXTERNAL_SERVICE_PORTS_START

previous_end = config.EXTERNAL_SERVICE_PORTS_END

# Limit the range to only contain a single port

config.EXTERNAL_SERVICE_PORTS_START = get_free_tcp_port()

config.EXTERNAL_SERVICE_PORTS_END = config.EXTERNAL_SERVICE_PORTS_START + 1

yield ExternalServicePortsManager()

config.EXTERNAL_SERVICE_PORTS_END = previous_end

config.EXTERNAL_SERVICE_PORTS_START = previous_start

class TestExternalServicePortsManager:

def test_reserve_port_within_range(

self, external_service_ports_manager: ExternalServicePortsManager

):

port = external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START)

assert port == config.EXTERNAL_SERVICE_PORTS_START

def test_reserve_port_outside_range(

self, external_service_ports_manager: ExternalServicePortsManager

):

with pytest.raises(PortNotAvailableException):

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_END + 1)

def test_reserve_any_port_within_range(

self, external_service_ports_manager: ExternalServicePortsManager

):

port = external_service_ports_manager.reserve_port()

assert port == config.EXTERNAL_SERVICE_PORTS_START

def test_reserve_port_all_reserved(

self, external_service_ports_manager: ExternalServicePortsManager

):

# the external service ports manager fixture only has 2 ports available,

# reserving 3 has to raise an error, but this could also happen earlier

# (if one of the ports is blocked by something else, like a previous test)

with pytest.raises(PortNotAvailableException):

external_service_ports_manager.reserve_port()

external_service_ports_manager.reserve_port()

external_service_ports_manager.reserve_port()

def test_reserve_same_port_twice(

self, external_service_ports_manager: ExternalServicePortsManager

):

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START)

with pytest.raises(PortNotAvailableException):

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START)

def test_reserve_custom_expiry(

self, external_service_ports_manager: ExternalServicePortsManager

):

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START, duration=1)

with pytest.raises(PortNotAvailableException):

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START)

time.sleep(1)

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START)

def test_check_is_port_reserved(

self, external_service_ports_manager: ExternalServicePortsManager

):

assert not external_service_ports_manager.is_port_reserved(

config.EXTERNAL_SERVICE_PORTS_START

)

external_service_ports_manager.reserve_port(config.EXTERNAL_SERVICE_PORTS_START)

assert external_service_ports_manager.is_port_reserved(config.EXTERNAL_SERVICE_PORTS_START)

@pytest.fixture()

def paginated_list():

yield PaginatedList([{"Id": i, "Filter": i.upper()} for i in ["a", "b", "c", "d", "e"]])

class TestPaginatedList:

def test_list_smaller_than_max(self, paginated_list):

page, next_token = paginated_list.get_page(lambda i: i["Id"], page_size=6)

assert len(page) == 5

assert next_token is None

def test_next_token(self, paginated_list):

page, next_token = paginated_list.get_page(lambda i: i["Id"], page_size=2)

assert len(page) == 2

assert next_token == "c"

def test_continuation(self, paginated_list):

page, next_token = paginated_list.get_page(lambda i: i["Id"], page_size=2, next_token="c")

assert len(page) == 2

assert next_token == "e"

def test_end(self, paginated_list):

page, next_token = paginated_list.get_page(lambda i: i["Id"], page_size=2, next_token="e")

assert len(page) == 1

assert next_token is None

def test_filter(self, paginated_list):

page, next_token = paginated_list.get_page(

lambda i: i["Id"], page_size=6, filter_function=lambda i: i["Filter"] in ["B", "E"]

)

assert len(page) == 2

ids = [i["Id"] for i in page]

assert "b" in ids and "e" in ids

assert "a" not in ids

assert next_token is None