Public Access
emil
23183df7c7
Post-merge / detect-type (push) Successful in 11s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / configure-repo (push) Successful in 17s
Post-merge / release (push) Successful in 1m22s
Post-merge / vikunja (push) Successful in 32s
Post-merge / badges (push) Successful in 50s
Post-merge / sync-wiki (push) Successful in 57s
193 lines
7.1 KiB
Python
193 lines
7.1 KiB
Python
"""Unit tests for devx.opentofu."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from subprocess import CompletedProcess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from devx.opentofu import get_tofu_output, get_tofu_vm_field, get_tofu_vm_ip
|
|
|
|
|
|
class TestGetTofuOutput:
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_returns_parsed_json(self, mock_run: MagicMock) -> None:
|
|
payload = {"staging": {"ipv4": "1.2.3.4"}}
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "customer_vms"],
|
|
returncode=0,
|
|
stdout=json.dumps(payload),
|
|
stderr="",
|
|
)
|
|
result = get_tofu_output("customer_vms", cwd="/tmp/tofu/staging")
|
|
assert result == payload
|
|
mock_run.assert_called_once()
|
|
call_kwargs = mock_run.call_args
|
|
assert call_kwargs.args[0] == ["tofu", "output", "-json", "customer_vms"]
|
|
assert call_kwargs.kwargs["cwd"] == "/tmp/tofu/staging"
|
|
assert call_kwargs.kwargs["env"] is None
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_with_env(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "obs"],
|
|
returncode=0,
|
|
stdout='{"staging": {"ipv4": "5.6.7.8"}}',
|
|
stderr="",
|
|
)
|
|
env = {"HCLOUD_TOKEN": "secret"}
|
|
result = get_tofu_output("obs", cwd=Path("/tmp"), env=env)
|
|
assert result == {"staging": {"ipv4": "5.6.7.8"}}
|
|
assert mock_run.call_args.kwargs["env"] == env
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_no_cwd(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "x"],
|
|
returncode=0,
|
|
stdout='{"a": 1}',
|
|
stderr="",
|
|
)
|
|
result = get_tofu_output("x")
|
|
assert result == {"a": 1}
|
|
assert mock_run.call_args.kwargs["cwd"] is None
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_pathlib_cwd(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "x"],
|
|
returncode=0,
|
|
stdout="{}",
|
|
stderr="",
|
|
)
|
|
get_tofu_output("x", cwd=Path("/some/path"))
|
|
assert mock_run.call_args.kwargs["cwd"] == "/some/path"
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_failure_raises_runtime_error(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "x"],
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="Error: module not found",
|
|
)
|
|
with pytest.raises(RuntimeError, match="tofu output failed"):
|
|
get_tofu_output("x", cwd="/tmp")
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_invalid_json_raises(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "x"],
|
|
returncode=0,
|
|
stdout="not json",
|
|
stderr="",
|
|
)
|
|
with pytest.raises(json.JSONDecodeError):
|
|
get_tofu_output("x")
|
|
|
|
|
|
class TestGetTofuVmIp:
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_returns_ipv4(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "customer_vms"],
|
|
returncode=0,
|
|
stdout=json.dumps({"oblachno": {"ipv4": "10.0.0.1"}}),
|
|
stderr="",
|
|
)
|
|
ip = get_tofu_vm_ip("customer_vms", "oblachno", cwd="/tmp")
|
|
assert ip == "10.0.0.1"
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_missing_vm_returns_empty(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "customer_vms"],
|
|
returncode=0,
|
|
stdout=json.dumps({"other": {"ipv4": "10.0.0.2"}}),
|
|
stderr="",
|
|
)
|
|
ip = get_tofu_vm_ip("customer_vms", "missing", cwd="/tmp")
|
|
assert ip == ""
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_missing_ip_field_returns_empty(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "customer_vms"],
|
|
returncode=0,
|
|
stdout=json.dumps({"vm1": {"name": "test"}}),
|
|
stderr="",
|
|
)
|
|
ip = get_tofu_vm_ip("customer_vms", "vm1", cwd="/tmp")
|
|
assert ip == ""
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_custom_ip_field(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "vms"],
|
|
returncode=0,
|
|
stdout=json.dumps({"vm1": {"address": "192.168.1.1"}}),
|
|
stderr="",
|
|
)
|
|
ip = get_tofu_vm_ip("vms", "vm1", cwd="/tmp", ip_field="address")
|
|
assert ip == "192.168.1.1"
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_non_dict_output_returns_empty(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "vms"],
|
|
returncode=0,
|
|
stdout='["not", "a", "dict"]',
|
|
stderr="",
|
|
)
|
|
ip = get_tofu_vm_ip("vms", "vm1", cwd="/tmp")
|
|
assert ip == ""
|
|
|
|
|
|
class TestGetTofuVmField:
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_returns_field_value(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "obs"],
|
|
returncode=0,
|
|
stdout=json.dumps({"staging": {"volume_linux_device": "/dev/sda1"}}),
|
|
stderr="",
|
|
)
|
|
val = get_tofu_vm_field("obs", "staging", "volume_linux_device", cwd="/tmp")
|
|
assert val == "/dev/sda1"
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_missing_field_returns_empty(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "obs"],
|
|
returncode=0,
|
|
stdout=json.dumps({"staging": {"ipv4": "1.2.3.4"}}),
|
|
stderr="",
|
|
)
|
|
val = get_tofu_vm_field("obs", "staging", "volume_linux_device", cwd="/tmp")
|
|
assert val == ""
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_missing_vm_returns_empty(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "obs"],
|
|
returncode=0,
|
|
stdout=json.dumps({"prod": {"x": "y"}}),
|
|
stderr="",
|
|
)
|
|
val = get_tofu_vm_field("obs", "staging", "x", cwd="/tmp")
|
|
assert val == ""
|
|
|
|
@patch("devx.opentofu.subprocess.run")
|
|
def test_non_dict_output_returns_empty(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = CompletedProcess(
|
|
args=["tofu", "output", "-json", "obs"],
|
|
returncode=0,
|
|
stdout='"a string"',
|
|
stderr="",
|
|
)
|
|
val = get_tofu_vm_field("obs", "staging", "x", cwd="/tmp")
|
|
assert val == ""
|