Public Access
122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
"""Unit tests for devx.molecule.lease."""
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, mock_open, patch
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.lease import (
|
|
DEFAULT_TTL_SECONDS,
|
|
LEASE_UNTIL_ENV,
|
|
OWNER_ENV,
|
|
emit_github_env,
|
|
lease_env,
|
|
lease_owner,
|
|
main,
|
|
)
|
|
|
|
|
|
class TestLeaseOwner:
|
|
def test_gitea_env_preferred(self) -> None:
|
|
with patch.dict(
|
|
os.environ,
|
|
{"GITEA_RUN_ID": "123", "GITEA_JOB": "tests", "GITHUB_RUN_ID": "999"},
|
|
):
|
|
assert lease_owner() == "run-123-tests"
|
|
|
|
def test_github_env_fallback(self) -> None:
|
|
env = {"GITHUB_RUN_ID": "777", "GITHUB_JOB": "molecule"}
|
|
with patch.dict(os.environ, env, clear=True):
|
|
assert lease_owner() == "run-777-molecule"
|
|
|
|
def test_job_defaults_when_missing(self) -> None:
|
|
with patch.dict(os.environ, {"GITHUB_RUN_ID": "5"}, clear=True):
|
|
assert lease_owner() == "run-5-job"
|
|
|
|
def test_local_fallback(self) -> None:
|
|
with (
|
|
patch.dict(os.environ, {}, clear=True),
|
|
patch("devx.molecule.lease.socket.gethostname", return_value="devbox"),
|
|
):
|
|
assert lease_owner() == "local-devbox"
|
|
|
|
|
|
class TestLeaseEnv:
|
|
def test_arithmetic(self) -> None:
|
|
env = lease_env(ttl=3600, now=1000)
|
|
assert env[LEASE_UNTIL_ENV] == str(1000 + 3600)
|
|
assert OWNER_ENV in env
|
|
|
|
def test_default_now_uses_time(self) -> None:
|
|
with patch("devx.molecule.lease.time.time", return_value=2000.9):
|
|
env = lease_env(ttl=60)
|
|
assert env[LEASE_UNTIL_ENV] == str(2000 + 60)
|
|
|
|
def test_default_ttl(self) -> None:
|
|
env = lease_env(now=0)
|
|
assert env[LEASE_UNTIL_ENV] == str(DEFAULT_TTL_SECONDS)
|
|
|
|
|
|
class TestEmitGithubEnv:
|
|
def test_writes_lines(self) -> None:
|
|
m = mock_open()
|
|
with (
|
|
patch.dict(os.environ, {"GITHUB_ENV": "/tmp/env"}, clear=False),
|
|
patch("builtins.open", m),
|
|
):
|
|
written = emit_github_env({"A": "1", "B": "2"})
|
|
assert written == 2
|
|
m.assert_called_once_with("/tmp/env", "a", encoding="utf-8")
|
|
m().write.assert_any_call("A=1\n")
|
|
m().write.assert_any_call("B=2\n")
|
|
|
|
def test_noop_without_env(self) -> None:
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
assert emit_github_env({"A": "1"}) == 0
|
|
|
|
|
|
class TestCli:
|
|
def test_prints_and_exports(self) -> None:
|
|
runner = CliRunner()
|
|
m = mock_open()
|
|
with (
|
|
patch.dict(os.environ, {"GITHUB_ENV": "/tmp/env", "GITHUB_RUN_ID": "42"}, clear=True),
|
|
patch("builtins.open", m),
|
|
patch("devx.molecule.lease.time.time", return_value=100.0),
|
|
):
|
|
result = runner.invoke(main, ["--ttl", "600"])
|
|
assert result.exit_code == 0
|
|
assert f"{LEASE_UNTIL_ENV}=700" in result.output
|
|
assert f"{OWNER_ENV}=run-42-job" in result.output
|
|
assert "Exported 2 lease vars" in result.output
|
|
|
|
def test_prints_without_github_env(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch.dict(os.environ, {"GITHUB_RUN_ID": "9", "GITHUB_JOB": "tests"}, clear=True),
|
|
patch("devx.molecule.lease.time.time", return_value=50.0),
|
|
):
|
|
result = runner.invoke(main)
|
|
assert result.exit_code == 0
|
|
assert f"{LEASE_UNTIL_ENV}={50 + DEFAULT_TTL_SECONDS}" in result.output
|
|
assert f"{OWNER_ENV}=run-9-tests" in result.output
|
|
assert "GITHUB_ENV" not in result.output
|
|
|
|
|
|
class TestStartDockerIntegration:
|
|
"""start_docker.main() must emit lease env on success (REQ-2)."""
|
|
|
|
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
|
def test_emits_lease_on_success(self, _mock: MagicMock) -> None:
|
|
from devx.molecule.start_docker import main as start_docker_main
|
|
|
|
runner = CliRunner()
|
|
m = mock_open()
|
|
with (
|
|
patch.dict(os.environ, {"GITHUB_ENV": "/tmp/env", "GITHUB_RUN_ID": "7"}, clear=True),
|
|
patch("builtins.open", m),
|
|
):
|
|
result = runner.invoke(start_docker_main, ["--timeout", "1"])
|
|
assert result.exit_code == 0
|
|
assert "Exported 2 lease vars" in result.output
|