47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
import stat
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pytest
|
|
from dotenv import dotenv_values
|
|
|
|
from adcloud_server.container_entrypoint import SECRET_NAMES, bootstrap
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_secret_environment(monkeypatch):
|
|
for name in SECRET_NAMES:
|
|
monkeypatch.delenv(name, raising=False)
|
|
|
|
|
|
def test_generated_secrets_survive_restarts_with_restricted_permissions(
|
|
tmp_path, capsys
|
|
):
|
|
path = tmp_path / "config" / "identity.env"
|
|
first = bootstrap(path)
|
|
assert bootstrap(path) == first
|
|
assert dict(dotenv_values(path)) == first
|
|
assert all(len(value) >= 32 for value in first.values())
|
|
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
assert capsys.readouterr().out == ""
|
|
|
|
|
|
def test_parallel_container_bootstraps_use_the_same_secrets(tmp_path):
|
|
path = tmp_path / "identity.env"
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = list(pool.map(lambda _: bootstrap(path), range(4)))
|
|
assert all(result == results[0] for result in results)
|
|
assert dict(dotenv_values(path)) == results[0]
|
|
|
|
|
|
def test_explicit_secrets_are_persisted_and_invalid_input_cannot_corrupt_file(
|
|
tmp_path, monkeypatch
|
|
):
|
|
path = tmp_path / "identity.env"
|
|
monkeypatch.setenv("IDENTITY_SECRET", "explicit-identity-" + "x" * 32)
|
|
first = bootstrap(path)
|
|
assert first["IDENTITY_SECRET"] == "explicit-identity-" + "x" * 32
|
|
monkeypatch.setenv("SERVICE_TOKEN", "x" * 32 + "\nINJECTED=true")
|
|
with pytest.raises(ValueError):
|
|
bootstrap(path)
|
|
assert dict(dotenv_values(path)) == first
|