384 lines
13 KiB
Python
384 lines
13 KiB
Python
"""Tests for credential provisioning."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from api.auth.manifest import (
|
|
CREDENTIALS,
|
|
MANAGED_VARS,
|
|
POSTGRES_DATABASES,
|
|
POSTGRES_ROLES,
|
|
Format,
|
|
Provisioner,
|
|
Tier,
|
|
)
|
|
from api.auth.provision import (
|
|
derive_all,
|
|
provision,
|
|
write_env,
|
|
)
|
|
|
|
ROOT = bytes.fromhex("deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
|
|
COMMIT = "abc1234"
|
|
|
|
|
|
class TestManifestInvariants:
|
|
def test_no_duplicate_env_vars(self):
|
|
env_vars = [c.env_var for c in CREDENTIALS]
|
|
assert len(env_vars) == len(set(env_vars))
|
|
|
|
def test_all_managed_vars_in_credentials(self):
|
|
assert MANAGED_VARS == frozenset(c.env_var for c in CREDENTIALS)
|
|
|
|
def test_postgres_roles_have_credentials(self):
|
|
for env_var in POSTGRES_ROLES.values():
|
|
assert env_var in MANAGED_VARS
|
|
|
|
def test_postgres_roles_have_databases(self):
|
|
for role in POSTGRES_ROLES:
|
|
assert role in POSTGRES_DATABASES
|
|
|
|
def test_aliases_share_scope(self):
|
|
by_var = {c.env_var: c for c in CREDENTIALS}
|
|
for cred in CREDENTIALS:
|
|
if cred.alias_of:
|
|
parent = by_var[cred.alias_of]
|
|
assert cred.scope == parent.scope
|
|
assert cred.fmt == parent.fmt
|
|
|
|
def test_credential_count(self):
|
|
assert len(CREDENTIALS) == 17
|
|
|
|
|
|
class TestDeriveAll:
|
|
def test_returns_dict(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
assert isinstance(values, dict)
|
|
|
|
def test_skips_skip_provisioner(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
skip_vars = {
|
|
c.env_var for c in CREDENTIALS if c.provisioner is Provisioner.SKIP
|
|
}
|
|
for var in skip_vars:
|
|
assert var not in values
|
|
|
|
def test_hex_format_values(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
for cred in CREDENTIALS:
|
|
if cred.provisioner is Provisioner.SKIP:
|
|
continue
|
|
if cred.fmt is Format.HEX:
|
|
v = values[cred.env_var]
|
|
assert len(v) == 64
|
|
bytes.fromhex(v)
|
|
|
|
def test_password_format_values(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
for cred in CREDENTIALS:
|
|
if cred.provisioner is Provisioner.SKIP:
|
|
continue
|
|
if cred.fmt is Format.PASSWORD:
|
|
v = values[cred.env_var]
|
|
assert "+" not in v
|
|
assert "/" not in v
|
|
|
|
def test_deterministic(self):
|
|
a = derive_all(ROOT, COMMIT)
|
|
b = derive_all(ROOT, COMMIT)
|
|
assert a == b
|
|
|
|
def test_different_commit_different_service_values(self):
|
|
a = derive_all(ROOT, "commit-a")
|
|
b = derive_all(ROOT, "commit-b")
|
|
for cred in CREDENTIALS:
|
|
if cred.tier is Tier.BOOTSTRAP and cred.provisioner is not Provisioner.SKIP:
|
|
assert a[cred.env_var] == b[cred.env_var]
|
|
service_vars = [
|
|
c.env_var
|
|
for c in CREDENTIALS
|
|
if c.tier is Tier.SERVICE
|
|
and c.provisioner is not Provisioner.SKIP
|
|
and c.alias_of is None
|
|
]
|
|
assert any(a[v] != b[v] for v in service_vars)
|
|
|
|
def test_aliases_match_parent(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
for cred in CREDENTIALS:
|
|
if cred.alias_of and cred.provisioner is not Provisioner.SKIP:
|
|
assert values[cred.env_var] == values[cred.alias_of]
|
|
|
|
|
|
class TestWriteEnv:
|
|
def test_creates_new_file(self, tmp_path):
|
|
p = tmp_path / ".env"
|
|
write_env({"A": "1", "B": "2"}, p)
|
|
lines = p.read_text().splitlines()
|
|
assert lines == ["A=1", "B=2"]
|
|
|
|
def test_preserves_unmanaged_keys(self, tmp_path):
|
|
p = tmp_path / ".env"
|
|
p.write_text("DATABRICKS_TOKEN=ext\nOLD_KEY=val\n")
|
|
write_env({"NEW_KEY": "new"}, p)
|
|
content = p.read_text()
|
|
assert "DATABRICKS_TOKEN=ext" in content
|
|
assert "NEW_KEY=new" in content
|
|
assert "OLD_KEY=val" in content
|
|
|
|
def test_overwrites_managed_keys(self, tmp_path):
|
|
p = tmp_path / ".env"
|
|
p.write_text("MY_KEY=old\n")
|
|
write_env({"MY_KEY": "new"}, p)
|
|
lines = p.read_text().splitlines()
|
|
assert "MY_KEY=new" in lines
|
|
assert "MY_KEY=old" not in lines
|
|
|
|
def test_sorted_output(self, tmp_path):
|
|
p = tmp_path / ".env"
|
|
write_env({"Z": "3", "A": "1", "M": "2"}, p)
|
|
lines = p.read_text().splitlines()
|
|
keys = [ln.split("=")[0] for ln in lines]
|
|
assert keys == sorted(keys)
|
|
|
|
def test_skips_comments_and_blanks(self, tmp_path):
|
|
p = tmp_path / ".env"
|
|
p.write_text("# comment\n\nKEY=val\n")
|
|
write_env({"NEW": "v"}, p)
|
|
content = p.read_text()
|
|
assert "KEY=val" in content
|
|
assert "NEW=v" in content
|
|
assert "# comment" not in content
|
|
|
|
|
|
class TestProvisionPostgres:
|
|
def test_uses_psql_variable_for_password(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
with patch("api.auth.provision.subprocess.run") as mock_run:
|
|
from api.auth.provision import provision_postgres
|
|
|
|
provision_postgres(values, container="test-pg")
|
|
|
|
assert mock_run.call_count == len(POSTGRES_ROLES)
|
|
for c in mock_run.call_args_list:
|
|
cmd = c[0][0]
|
|
assert cmd[0] == "docker"
|
|
assert cmd[1] == "exec"
|
|
assert "test-pg" in cmd
|
|
assert "-e" in cmd
|
|
assert "ALTER ROLE" in cmd[-1]
|
|
assert "PASSWORD" in cmd[-1]
|
|
|
|
|
|
class TestProvisionGitea:
|
|
def test_uses_api_not_docker_exec(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
values["GITEA_TOKEN"] = "current-token"
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = []
|
|
mock_client.post.return_value.json.return_value = {"sha1": "new-tok"}
|
|
|
|
with patch("api.auth.provision.GiteaClient", return_value=mock_client):
|
|
from api.auth.provision import provision_gitea
|
|
|
|
token = provision_gitea(values, base_url="http://test:3000/api/v1")
|
|
|
|
assert token == "new-tok"
|
|
mock_client.change_admin_password.assert_called_once()
|
|
|
|
def test_creates_fresh_token(self):
|
|
values = derive_all(ROOT, COMMIT)
|
|
values["GITEA_TOKEN"] = "old-token"
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = [
|
|
{"id": 1, "name": "deploy-old"},
|
|
{"id": 2, "name": "other-token"},
|
|
]
|
|
mock_client.post.return_value.json.return_value = {"sha1": "fresh"}
|
|
|
|
with patch("api.auth.provision.GiteaClient", return_value=mock_client):
|
|
from api.auth.provision import provision_gitea
|
|
|
|
token = provision_gitea(values)
|
|
|
|
assert token == "fresh"
|
|
mock_client.delete.assert_called_once()
|
|
|
|
|
|
class TestProvisionEndToEnd:
|
|
def test_partial_failure_still_writes_env(self, tmp_path):
|
|
env = tmp_path / ".env"
|
|
with (
|
|
patch(
|
|
"api.auth.provision.provision_postgres",
|
|
side_effect=Exception("pg down"),
|
|
),
|
|
patch(
|
|
"api.auth.provision.provision_gitea",
|
|
side_effect=Exception("gitea down"),
|
|
),
|
|
patch("api.auth.provision.time.sleep"),
|
|
):
|
|
result = provision(ROOT, COMMIT, env)
|
|
|
|
assert result.env_written
|
|
assert env.exists()
|
|
assert len(result.errors) == 2
|
|
assert not result.postgres
|
|
assert not result.gitea
|
|
|
|
def test_full_success(self, tmp_path):
|
|
env = tmp_path / ".env"
|
|
with (
|
|
patch("api.auth.provision.provision_postgres"),
|
|
patch("api.auth.provision.provision_gitea", return_value="tok-new"),
|
|
patch("api.auth.provision.time.sleep"),
|
|
):
|
|
result = provision(ROOT, COMMIT, env)
|
|
|
|
assert result.ok
|
|
assert result.postgres
|
|
assert result.gitea
|
|
assert result.env_written
|
|
content = env.read_text()
|
|
assert "GITEA_TOKEN=tok-new" in content
|
|
|
|
def test_skip_backends(self, tmp_path):
|
|
env = tmp_path / ".env"
|
|
result = provision(ROOT, COMMIT, env, skip_backends=True)
|
|
assert result.ok
|
|
assert result.env_written
|
|
assert not result.postgres
|
|
assert not result.gitea
|
|
|
|
def test_gitea_actions_failure_recorded(self, tmp_path):
|
|
"""gitea-actions failure is appended to errors but does not crash."""
|
|
env = tmp_path / ".env"
|
|
with (
|
|
patch("api.auth.provision.provision_postgres"),
|
|
patch("api.auth.provision.provision_gitea", return_value="tok"),
|
|
patch(
|
|
"api.auth.provision.provision_gitea_actions",
|
|
side_effect=Exception("actions down"),
|
|
),
|
|
patch("api.auth.provision.time.sleep"),
|
|
):
|
|
result = provision(ROOT, COMMIT, env)
|
|
|
|
assert result.env_written
|
|
error_backends = [b for b, _ in result.errors]
|
|
assert "gitea-actions" in error_backends
|
|
|
|
|
|
class TestProvisionGiteaActions:
|
|
def test_raises_without_gitea_token(self) -> None:
|
|
"""Raises ValueError when GITEA_TOKEN is missing."""
|
|
from api.auth.provision import provision_gitea_actions
|
|
|
|
with pytest.raises(ValueError, match="GITEA_TOKEN not available"):
|
|
provision_gitea_actions({})
|
|
|
|
def test_sets_secrets_successfully(self) -> None:
|
|
from api.auth.provision import (
|
|
GITEA_ACTIONS_SECRET_MAP,
|
|
provision_gitea_actions,
|
|
)
|
|
|
|
values = {
|
|
"GITEA_TOKEN": "tok",
|
|
**{env_var: "val" for env_var in GITEA_ACTIONS_SECRET_MAP.values()},
|
|
}
|
|
mock_client = MagicMock()
|
|
mock_client.put.return_value.status_code = 201
|
|
|
|
with patch("api.clients.gitea.GiteaClient", return_value=mock_client):
|
|
provision_gitea_actions(values)
|
|
|
|
assert mock_client.put.call_count > 0
|
|
|
|
def test_warns_on_non_success_status(self) -> None:
|
|
from api.auth.provision import (
|
|
GITEA_ACTIONS_SECRET_MAP,
|
|
provision_gitea_actions,
|
|
)
|
|
|
|
values = {
|
|
"GITEA_TOKEN": "tok",
|
|
**{env_var: "val" for env_var in GITEA_ACTIONS_SECRET_MAP.values()},
|
|
}
|
|
mock_client = MagicMock()
|
|
mock_client.put.return_value.status_code = 500
|
|
|
|
with patch("api.clients.gitea.GiteaClient", return_value=mock_client):
|
|
# Should not raise, just warn
|
|
provision_gitea_actions(values)
|
|
|
|
def test_skips_secrets_with_missing_values(self) -> None:
|
|
"""Secrets whose env var is not in values dict are skipped."""
|
|
from api.auth.provision import provision_gitea_actions
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.put.return_value.status_code = 201
|
|
# Only GITEA_TOKEN is set → maps to DEPLOY_TOKEN
|
|
# GITEA_ADMIN_USER and GITEA_ADMIN_PASSWORD not set → skip
|
|
with patch("api.clients.gitea.GiteaClient", return_value=mock_client):
|
|
provision_gitea_actions({"GITEA_TOKEN": "tok"})
|
|
|
|
# Only 1 call: DEPLOY_TOKEN (the other 2 are skipped)
|
|
assert mock_client.put.call_count == 1
|
|
|
|
|
|
class TestBootstrap:
|
|
def test_skip_backends(self, tmp_path) -> None:
|
|
from api.auth.provision import bootstrap
|
|
|
|
env = tmp_path / ".env"
|
|
result = bootstrap(ROOT, COMMIT, env, skip_backends=True)
|
|
assert result.env_written
|
|
assert result.ok
|
|
|
|
def test_all_backends_fail_records_errors(self, tmp_path) -> None:
|
|
from api.auth.provision import bootstrap
|
|
|
|
env = tmp_path / ".env"
|
|
with (
|
|
patch(
|
|
"api.auth.provision.bootstrap_postgres",
|
|
side_effect=Exception("pg down"),
|
|
),
|
|
patch(
|
|
"api.auth.provision.provision_gitea",
|
|
side_effect=Exception("gitea down"),
|
|
),
|
|
patch(
|
|
"api.auth.provision.provision_gitea_actions",
|
|
side_effect=Exception("actions down"),
|
|
),
|
|
):
|
|
result = bootstrap(ROOT, COMMIT, env)
|
|
|
|
assert result.env_written
|
|
error_backends = [b for b, _ in result.errors]
|
|
assert "postgres" in error_backends
|
|
assert "gitea" in error_backends
|
|
assert "gitea-actions" in error_backends
|
|
|
|
def test_success_path(self, tmp_path) -> None:
|
|
from api.auth.provision import bootstrap
|
|
|
|
env = tmp_path / ".env"
|
|
with (
|
|
patch("api.auth.provision.bootstrap_postgres"),
|
|
patch("api.auth.provision.provision_postgres"),
|
|
patch("api.auth.provision.provision_gitea", return_value="tok"),
|
|
patch("api.auth.provision.provision_gitea_actions"),
|
|
):
|
|
result = bootstrap(ROOT, COMMIT, env)
|
|
|
|
assert result.ok
|
|
assert result.postgres
|
|
assert result.gitea
|