127 lines
3.5 KiB
Python
127 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import stat
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
|
|
import pytest
|
|
|
|
|
|
def _prepare_module() -> ModuleType:
|
|
path = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "deploy"
|
|
/ "telemetry-plane"
|
|
/ "prepare.py"
|
|
)
|
|
specification = importlib.util.spec_from_file_location(
|
|
"mission_core_telemetry_prepare",
|
|
path,
|
|
)
|
|
assert specification is not None
|
|
assert specification.loader is not None
|
|
module = importlib.util.module_from_spec(specification)
|
|
specification.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
prepare = _prepare_module()
|
|
|
|
|
|
def test_initialize_environment_generates_private_unique_secrets(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
env_path = tmp_path / ".env"
|
|
monkeypatch.setattr(prepare, "ENV_PATH", env_path)
|
|
|
|
prepare._initialize_environment("192.0.2.15")
|
|
|
|
values = dict(
|
|
line.split("=", 1)
|
|
for line in env_path.read_text(encoding="utf-8").splitlines()
|
|
)
|
|
assert values["MISSIONCORE_MQTT_BIND_ADDRESS"] == "192.0.2.15"
|
|
secrets = {
|
|
values["MISSIONCORE_DB_PASSWORD"],
|
|
values["MISSIONCORE_DB_INGEST_PASSWORD"],
|
|
values["MISSIONCORE_MQTT_INGEST_PASSWORD"],
|
|
values["MISSIONCORE_MQTT_WORKER_006_PASSWORD"],
|
|
}
|
|
assert len(secrets) == 4
|
|
assert all(len(secret) >= 40 for secret in secrets)
|
|
assert stat.S_IMODE(env_path.stat().st_mode) == 0o600
|
|
|
|
|
|
def test_initialize_environment_refuses_to_replace_credentials(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
env_path = tmp_path / ".env"
|
|
env_path.write_text("existing=true\n", encoding="utf-8")
|
|
monkeypatch.setattr(prepare, "ENV_PATH", env_path)
|
|
|
|
with pytest.raises(RuntimeError, match="refusing to overwrite"):
|
|
prepare._initialize_environment("127.0.0.1")
|
|
|
|
assert env_path.read_text(encoding="utf-8") == "existing=true\n"
|
|
|
|
|
|
def test_environment_migration_adds_only_new_private_values(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
env_path = tmp_path / ".env"
|
|
env_path.write_text(
|
|
"MISSIONCORE_DB_PASSWORD=keep-me\n"
|
|
"MISSIONCORE_MQTT_WORKER_006_USER=worker-006\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(prepare, "ENV_PATH", env_path)
|
|
|
|
prepare._migrate_environment()
|
|
first = env_path.read_text(encoding="utf-8")
|
|
prepare._migrate_environment()
|
|
|
|
assert "MISSIONCORE_DB_PASSWORD=keep-me" in first
|
|
assert "MISSIONCORE_DB_INGEST_PASSWORD=" in first
|
|
assert "MISSIONCORE_MQTT_WORKER_006_CONTOUR=worker-006" in first
|
|
assert env_path.read_text(encoding="utf-8") == first
|
|
assert stat.S_IMODE(env_path.stat().st_mode) == 0o600
|
|
|
|
|
|
def test_existing_password_file_is_updated_without_recreation(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
password_path = tmp_path / "passwords"
|
|
password_path.write_text("existing", encoding="utf-8")
|
|
calls: list[tuple[str, bool]] = []
|
|
|
|
def capture(
|
|
path: Path,
|
|
username: str,
|
|
password: str,
|
|
*,
|
|
create: bool,
|
|
) -> None:
|
|
assert path == password_path
|
|
assert password
|
|
calls.append((username, create))
|
|
|
|
monkeypatch.setattr(prepare, "_password_entry", capture)
|
|
|
|
prepare._prepare_password_entries(
|
|
password_path,
|
|
"missioncore-ingest",
|
|
"ingest-secret",
|
|
"worker-006",
|
|
"worker-secret",
|
|
)
|
|
|
|
assert calls == [
|
|
("missioncore-ingest", False),
|
|
("worker-006", False),
|
|
]
|