mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-13 14:16:46 +00:00
feat(roles): pull custom roles from inventory-declared sources
An inventory group web-app-cst-<name> with applications.<id>.custom: true and
a pinned source {url, version, path} is pulled into roles/ before the play,
read raw (Jinja, Ansible tags and credential URLs refused), placed through a
staging directory under an exclusive lock, and owned by a marker that names
its own directory, so a pull or roles-clean never replaces or removes a
directory it did not place and a failed pull leaves roles/ as it was.
Directory inventories, host lists and non-UTF-8 files behave as Ansible
reads them; generated test deploys pass --skip-pull. make roles-pull,
roles-clean and roles list, INFINITO_EXTRA_MOUNTS for local sources, the
web-app-cst category, the reserved prefix lint and a central
repository_files() that also yields the gitignored custom roles let every
lint scan them like in-tree roles. Supersedes the uncommitted 045 External
Roles draft and reuses its fetch, sync and mount design.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 5.5 committed
1 parent
cb4d340de9
commit
d92375922d
101 files changed
+3731
-248
No files matched your search
@@ -1,9 +1,43 @@
|
||||
# cli/administration/deploy/dedicated/tests/test_command_ids.py
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
from cli.administration.deploy.dedicated.command import _normalize_app_ids
|
||||
from cli.administration.deploy.dedicated import command
|
||||
from cli.administration.deploy.dedicated.command import (
|
||||
_normalize_app_ids,
|
||||
pull_custom_roles,
|
||||
)
|
||||
|
||||
|
||||
class TestPullCustomRoles(unittest.TestCase):
|
||||
@mock.patch("cli.administration.deploy.dedicated.command.pull_or_exit")
|
||||
def test_pulls_the_inventory_by_default(self, pull_or_exit):
|
||||
pull_custom_roles(SimpleNamespace(inventory="inv/devices.yml", skip_pull=False))
|
||||
pull_or_exit.assert_called_once_with("inv/devices.yml")
|
||||
|
||||
@mock.patch("cli.administration.deploy.dedicated.command.pull_or_exit")
|
||||
def test_skip_pull_keeps_the_placed_roles(self, pull_or_exit):
|
||||
pull_custom_roles(SimpleNamespace(inventory="inv/devices.yml", skip_pull=True))
|
||||
pull_or_exit.assert_not_called()
|
||||
|
||||
def test_main_pulls_before_anything_reads_roles(self):
|
||||
calls: list[str] = []
|
||||
args = SimpleNamespace(inventory="inv/devices.yml", skip_pull=False)
|
||||
with (
|
||||
mock.patch.object(command, "parse_args", return_value=(args, [], {})),
|
||||
mock.patch.object(
|
||||
command, "pull_or_exit", side_effect=lambda _inv: calls.append("pull")
|
||||
),
|
||||
mock.patch.object(
|
||||
command,
|
||||
"run_from_args",
|
||||
side_effect=lambda *_a: calls.append("deploy") or 0,
|
||||
),
|
||||
):
|
||||
self.assertEqual(0, command.main([]))
|
||||
self.assertEqual(["pull", "deploy"], calls)
|
||||
|
||||
|
||||
class TestNormalizeAppIds(unittest.TestCase):
|
||||
|
||||
@@ -7,6 +7,7 @@ from cli.administration.deploy.dedicated import command as dedicated_command
|
||||
|
||||
|
||||
class TestDedicatedCommandPassthrough(unittest.TestCase):
|
||||
@unittest.mock.patch("cli.administration.deploy.dedicated.command.pull_or_exit")
|
||||
@unittest.mock.patch(
|
||||
"cli.administration.deploy.dedicated.command.validate_application_ids"
|
||||
)
|
||||
@@ -29,6 +30,7 @@ class TestDedicatedCommandPassthrough(unittest.TestCase):
|
||||
mock_load_modes_from_yaml,
|
||||
mock_run_ansible_playbook,
|
||||
mock_validate_application_ids,
|
||||
_mock_pull_or_exit,
|
||||
):
|
||||
mock_load_modes_from_yaml.return_value = []
|
||||
mock_add_dynamic_mode_args.return_value = {}
|
||||
|
||||
@@ -15,6 +15,7 @@ class TestDedicatedMainParser(unittest.TestCase):
|
||||
self.assertIn("limit", opts)
|
||||
self.assertIn("password_file", opts)
|
||||
self.assertIn("skip_build", opts)
|
||||
self.assertIn("skip_pull", opts)
|
||||
self.assertIn("id", opts)
|
||||
self.assertIn("verbose", opts)
|
||||
self.assertIn("diff", opts)
|
||||
|
||||
Whitespace-only changes.
@@ -0,0 +1,34 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
from cli.administration.deploy.development.deploy.run import _run_deploy
|
||||
|
||||
|
||||
class TestRunDeploy(unittest.TestCase):
|
||||
def deploy_cmd(self) -> list[str]:
|
||||
compose = mock.Mock()
|
||||
compose.exec.return_value = SimpleNamespace(returncode=0)
|
||||
_run_deploy(
|
||||
compose,
|
||||
deploy_ids=["web-app-cst-demo"],
|
||||
debug=False,
|
||||
passthrough=[],
|
||||
inventory_dir="/etc/inventories/dev",
|
||||
container_name="infinito",
|
||||
)
|
||||
return compose.exec.call_args.args[0]
|
||||
|
||||
def test_deploys_the_generated_inventory(self):
|
||||
cmd = self.deploy_cmd()
|
||||
self.assertEqual("/etc/inventories/dev/devices.yml", cmd[3])
|
||||
self.assertEqual(["--id", "web-app-cst-demo"], cmd[-2:])
|
||||
|
||||
def test_generated_inventory_keeps_the_placed_custom_roles(self):
|
||||
self.assertIn("--skip-pull", self.deploy_cmd())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -8,7 +8,9 @@ because an unsatisfied `:?` guard inside one of them aborts the whole stack.
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from cli.administration.deploy.development.env import compose_file_args
|
||||
@@ -23,6 +25,7 @@ _LOCAL_PRIMARY = {
|
||||
"INFINITO_CACHE_STACK": "",
|
||||
"INFINITO_GHCR_MIRROR": "",
|
||||
"INFINITO_PUBLISH_PORTS": "",
|
||||
"INFINITO_EXTRA_MOUNTS": "",
|
||||
}
|
||||
|
||||
_WORKTREE = {
|
||||
@@ -99,6 +102,35 @@ class TestComposeFileArgs(unittest.TestCase):
|
||||
compose_file_args()[-2:], ["-f", "compose/noports.override.yml"]
|
||||
)
|
||||
|
||||
def test_extra_mounts_append_their_generated_override(self) -> None:
|
||||
with (
|
||||
tempfile.TemporaryDirectory() as mount,
|
||||
patch.dict(
|
||||
os.environ,
|
||||
{**_LOCAL_PRIMARY, "INFINITO_EXTRA_MOUNTS": mount},
|
||||
clear=False,
|
||||
),
|
||||
patch(
|
||||
"cli.administration.deploy.development.extra_mounts.write_mounts_override",
|
||||
return_value="build/compose/extra-mounts.override.yml",
|
||||
) as write,
|
||||
):
|
||||
args = compose_file_args()
|
||||
self.assertEqual(write.call_args.args[1], [Path(mount)])
|
||||
self.assertEqual(args[-2:], ["-f", "build/compose/extra-mounts.override.yml"])
|
||||
|
||||
def test_an_entry_that_is_no_directory_stops_the_stack_naming_it(self) -> None:
|
||||
with (
|
||||
patch.dict(
|
||||
os.environ,
|
||||
{**_LOCAL_PRIMARY, "INFINITO_EXTRA_MOUNTS": "/nonexistent/gui"},
|
||||
clear=False,
|
||||
),
|
||||
self.assertRaises(SystemExit) as ctx,
|
||||
):
|
||||
compose_file_args()
|
||||
self.assertIn("/nonexistent/gui", str(ctx.exception.code))
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
@@ -0,0 +1,65 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from cli.administration.deploy.development.extra_mounts import (
|
||||
OVERRIDE,
|
||||
parse_mounts,
|
||||
write_mounts_override,
|
||||
)
|
||||
from utils.cache.yaml import load_yaml
|
||||
|
||||
|
||||
class TestParseMounts(unittest.TestCase):
|
||||
def test_existing_absolute_directories_are_returned(self) -> None:
|
||||
with (
|
||||
tempfile.TemporaryDirectory() as first,
|
||||
tempfile.TemporaryDirectory() as second,
|
||||
):
|
||||
self.assertEqual(
|
||||
parse_mounts(f"{first}, {second},"), [Path(first), Path(second)]
|
||||
)
|
||||
|
||||
def test_empty_value_mounts_nothing(self) -> None:
|
||||
self.assertEqual(parse_mounts(""), [])
|
||||
|
||||
def test_relative_entry_is_rejected_naming_it(self) -> None:
|
||||
with self.assertRaises(SystemExit) as ctx:
|
||||
parse_mounts("relative/dir")
|
||||
self.assertIn("relative/dir", str(ctx.exception.code))
|
||||
|
||||
def test_missing_directory_is_rejected_naming_it(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp, self.assertRaises(SystemExit) as ctx:
|
||||
parse_mounts(str(Path(tmp) / "absent"))
|
||||
self.assertIn("absent", str(ctx.exception.code))
|
||||
|
||||
|
||||
class TestWriteMountsOverride(unittest.TestCase):
|
||||
def test_each_directory_is_bound_read_only_at_its_own_path(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as repo:
|
||||
relative = write_mounts_override(Path(repo), [Path("/srv/gui")])
|
||||
override = load_yaml(Path(repo) / relative)
|
||||
self.assertEqual(relative, OVERRIDE)
|
||||
self.assertEqual(
|
||||
override,
|
||||
{
|
||||
"services": {
|
||||
"infinito": {
|
||||
"volumes": [
|
||||
{
|
||||
"type": "bind",
|
||||
"source": "/srv/gui",
|
||||
"target": "/srv/gui",
|
||||
"read_only": True,
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,36 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
from cli.administration.deploy.swarm import command
|
||||
|
||||
|
||||
class TestSwarmMain(unittest.TestCase):
|
||||
def test_custom_roles_are_pulled_before_the_closure_reads_roles(self):
|
||||
calls: list[str] = []
|
||||
args = SimpleNamespace(id=["web-app-cst-demo"], inventory="inv/devices.yml")
|
||||
|
||||
def closure(ids, _inventory):
|
||||
calls.append("closure")
|
||||
return ids
|
||||
|
||||
with (
|
||||
mock.patch.object(command, "parse_args", return_value=(args, [], {})),
|
||||
mock.patch.object(
|
||||
command,
|
||||
"pull_custom_roles",
|
||||
side_effect=lambda _args: calls.append("pull"),
|
||||
),
|
||||
mock.patch.object(command, "swarm_deploy_targets", side_effect=closure),
|
||||
mock.patch.object(command, "run_from_args", return_value=0) as run,
|
||||
):
|
||||
self.assertEqual(0, command.main([]))
|
||||
|
||||
self.assertEqual(["pull", "closure"], calls)
|
||||
run.assert_called_once_with(args, [], {})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -82,6 +82,35 @@ class TestEnsureImage(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestRunInContainer(unittest.TestCase):
|
||||
@unittest.mock.patch("subprocess.run")
|
||||
@unittest.mock.patch(
|
||||
"cli.administration.deploy.container.command.docker_exec",
|
||||
return_value=subprocess.CompletedProcess(args=[], returncode=0),
|
||||
)
|
||||
@unittest.mock.patch(
|
||||
"cli.administration.deploy.container.command.start_ci_container",
|
||||
return_value="infinito-ci-test",
|
||||
)
|
||||
def test_generated_inventory_keeps_the_placed_custom_roles(
|
||||
self, _start, mock_docker_exec, _run
|
||||
):
|
||||
deploy_container.run_in_container(
|
||||
image="infinito:latest",
|
||||
build=False,
|
||||
rebuild=False,
|
||||
no_cache=False,
|
||||
inventory_dir="/tmp/inventory",
|
||||
inventory_args=[],
|
||||
deploy_args=["--debug"],
|
||||
)
|
||||
|
||||
deploy_cmd = mock_docker_exec.call_args_list[-1].args[1]
|
||||
self.assertIn("cli.administration.deploy.dedicated", deploy_cmd)
|
||||
self.assertIn("--skip-pull", deploy_cmd)
|
||||
self.assertEqual("--debug", deploy_cmd[-1])
|
||||
|
||||
|
||||
class TestMain(unittest.TestCase):
|
||||
@unittest.mock.patch("cli.administration.deploy.container.command.run_in_container")
|
||||
def test_main_requires_forwarded_args(self, mock_run_in_container):
|
||||
|
||||
@@ -94,6 +94,24 @@ class TestCompareApplicationKeys(unittest.TestCase):
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn("services.dashbaord", result[0])
|
||||
|
||||
def test_custom_role_declaration_keys_are_legal(self):
|
||||
apps = {
|
||||
"web-app-cst-demo": {
|
||||
"custom": True,
|
||||
"source": {"url": "https://x/y.git", "version": "v1", "path": "r"},
|
||||
"services": {"port": 8080},
|
||||
}
|
||||
}
|
||||
defaults = {"web-app-cst-demo": {"services": {"port": 8080}}}
|
||||
self.assertEqual(compare_application_keys(apps, defaults, "src"), [])
|
||||
|
||||
def test_declaration_keys_on_an_in_tree_role_are_flagged(self):
|
||||
apps = {"app1": {"custom": True, "source": {"url": "https://x/y.git"}}}
|
||||
defaults = {"app1": {"services": {}}}
|
||||
result = compare_application_keys(apps, defaults, "src")
|
||||
self.assertEqual(3, len(result))
|
||||
self.assertTrue(any(err.endswith("app1: custom") for err in result))
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
Whitespace-only changes.
@@ -0,0 +1,149 @@
|
||||
"""Throwaway role repositories, inventories and core roles/ trees for the pull tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from utils.roles.mapping import (
|
||||
ROLE_FILE_META_MAIN,
|
||||
ROLE_FILE_TASKS_MAIN,
|
||||
ROLE_FILE_VARS_MAIN,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
ROLE = "web-app-cst-demo"
|
||||
ROLE_PATH = f"ansible/roles/{ROLE}"
|
||||
|
||||
_ENV = {
|
||||
**{key: value for key, value in os.environ.items() if not key.startswith("GIT_")},
|
||||
"GIT_CONFIG_GLOBAL": os.devnull,
|
||||
}
|
||||
|
||||
|
||||
def git(cwd: Path, *args: str) -> str:
|
||||
"""Run git in *cwd* with a throwaway identity and no signing.
|
||||
|
||||
Args:
|
||||
cwd: the repository.
|
||||
args: the git subcommand and its arguments.
|
||||
"""
|
||||
return subprocess.run(
|
||||
[
|
||||
"git",
|
||||
"-c",
|
||||
"user.name=test",
|
||||
"-c",
|
||||
"user.email=test@example.invalid",
|
||||
"-c",
|
||||
"commit.gpgsign=false",
|
||||
"-c",
|
||||
"tag.gpgsign=false",
|
||||
*args,
|
||||
],
|
||||
cwd=cwd,
|
||||
env=_ENV,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
).stdout
|
||||
|
||||
|
||||
def write(path: Path, text: str) -> Path:
|
||||
"""Write *text* to *path*, creating its parents.
|
||||
|
||||
Args:
|
||||
path: the file.
|
||||
text: its content.
|
||||
"""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(text, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def role_repo(root: Path, application_id: str = ROLE) -> Path:
|
||||
"""A repository whose tag ``v1.0.0`` holds the role and whose branch ``feature`` moved on.
|
||||
|
||||
Args:
|
||||
root: directory the repository is created in.
|
||||
application_id: the ``vars/main.yml`` application_id of the role.
|
||||
"""
|
||||
repo = root / "repo"
|
||||
write(repo / "README.md", "repository root\n")
|
||||
write(repo / ROLE_PATH / ROLE_FILE_META_MAIN, "galaxy_info: {}\n")
|
||||
write(repo / ROLE_PATH / ROLE_FILE_VARS_MAIN, f"application_id: {application_id}\n")
|
||||
write(repo / ROLE_PATH / ROLE_FILE_TASKS_MAIN, "release: v1\n")
|
||||
git(repo, "init", "-q", "-b", "main")
|
||||
git(repo, "add", "-A")
|
||||
git(repo, "commit", "-q", "-m", "v1")
|
||||
git(repo, "tag", "v1.0.0")
|
||||
git(repo, "checkout", "-q", "-b", "feature")
|
||||
write(repo / ROLE_PATH / ROLE_FILE_TASKS_MAIN, "release: feature\n")
|
||||
git(repo, "commit", "-q", "-am", "feature")
|
||||
return repo
|
||||
|
||||
|
||||
def local_role(root: Path, group: str) -> Path:
|
||||
"""A checkout without Git history that holds the role *group* under ansible/roles/.
|
||||
|
||||
Args:
|
||||
root: directory the checkout is created in.
|
||||
group: the role's group and application_id.
|
||||
"""
|
||||
checkout = root / f"{group}-checkout"
|
||||
write(checkout / "ansible/roles" / group / ROLE_FILE_META_MAIN, "galaxy_info: {}\n")
|
||||
write(
|
||||
checkout / "ansible/roles" / group / ROLE_FILE_VARS_MAIN,
|
||||
f"application_id: {group}\n",
|
||||
)
|
||||
return checkout
|
||||
|
||||
|
||||
def inventory(root: Path, groups: list[str], applications: dict) -> Path:
|
||||
"""An inventory whose host joins *groups* and whose host_vars set *applications*.
|
||||
|
||||
Args:
|
||||
root: directory the inventory is created in.
|
||||
groups: inventory groups ``localhost`` joins.
|
||||
applications: the host_vars ``applications`` block.
|
||||
"""
|
||||
write(
|
||||
root / "host_vars/localhost.yml",
|
||||
json.dumps({"applications": applications}),
|
||||
)
|
||||
children = {group: {"hosts": {"localhost": {}}} for group in groups}
|
||||
return write(root / "devices.yml", json.dumps({"all": {"children": children}}))
|
||||
|
||||
|
||||
def declaration(
|
||||
url: str, version: str | None = "v1.0.0", path: str = ROLE_PATH
|
||||
) -> dict:
|
||||
"""The ``applications.<group>`` block that declares a custom role.
|
||||
|
||||
Args:
|
||||
url: ``source.url``.
|
||||
version: ``source.version``; ``None`` leaves it out.
|
||||
path: ``source.path``.
|
||||
"""
|
||||
source = {"url": url, "path": path}
|
||||
if version is not None:
|
||||
source["version"] = version
|
||||
return {"custom": True, "source": source}
|
||||
|
||||
|
||||
def roles_dir(root: Path, *in_tree: str) -> Path:
|
||||
"""A core roles/ holding the given in-tree role directories.
|
||||
|
||||
Args:
|
||||
root: directory the roles/ is created in.
|
||||
in_tree: names of in-tree roles to create.
|
||||
"""
|
||||
roles = root / "roles"
|
||||
roles.mkdir(parents=True)
|
||||
for name in in_tree:
|
||||
(roles / name).mkdir()
|
||||
return roles
|
||||
Whitespace-only changes.
@@ -0,0 +1,39 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import cli.administration.roles.clean.__main__ as clean_main
|
||||
|
||||
|
||||
class TestCleanMain(unittest.TestCase):
|
||||
def test_reports_the_roles_it_removed_and_the_entries_it_left(self):
|
||||
lines = [
|
||||
"web-app-cst-a: removed",
|
||||
"web-app-cst-hand: left in place, the pull did not place it",
|
||||
]
|
||||
with (
|
||||
mock.patch.object(clean_main, "clean", return_value=lines) as clean,
|
||||
contextlib.redirect_stdout(io.StringIO()) as out,
|
||||
):
|
||||
self.assertEqual(0, clean_main.main([]))
|
||||
clean.assert_called_once_with(clean_main.ROLES_DIR, clean_main.WORK_DIR)
|
||||
self.assertEqual(
|
||||
"custom role web-app-cst-a: removed\n"
|
||||
"custom role web-app-cst-hand: left in place, the pull did not place it\n",
|
||||
out.getvalue(),
|
||||
)
|
||||
|
||||
def test_a_filesystem_error_exits_with_it(self):
|
||||
with (
|
||||
mock.patch.object(clean_main, "clean", side_effect=OSError("busy")),
|
||||
self.assertRaises(SystemExit) as exit_,
|
||||
):
|
||||
clean_main.main([])
|
||||
self.assertIn("busy", str(exit_.exception.code))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Whitespace-only changes.
@@ -0,0 +1,36 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import cli.administration.roles.list.__main__ as list_main
|
||||
from cli.administration.roles.marker import MARKER_FILE, marker_text
|
||||
|
||||
ORIGIN = "https://example.invalid/w.git ansible/roles/web-app-cst-a @ v1 (commit abc)"
|
||||
|
||||
|
||||
class TestListMain(unittest.TestCase):
|
||||
def test_prints_each_placed_role_with_source_and_commit(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
roles = Path(tmp)
|
||||
marker = roles / "web-app-cst-a" / MARKER_FILE
|
||||
marker.parent.mkdir()
|
||||
marker.write_text(
|
||||
marker_text("web-app-cst-a", "web-app-cst-a", ORIGIN), encoding="utf-8"
|
||||
)
|
||||
(roles / "web-app-cst-hand").mkdir()
|
||||
(roles / "web-app-nextcloud").mkdir()
|
||||
with (
|
||||
mock.patch.object(list_main, "ROLES_DIR", roles),
|
||||
contextlib.redirect_stdout(io.StringIO()) as out,
|
||||
):
|
||||
self.assertEqual(0, list_main.main([]))
|
||||
self.assertEqual(f"web-app-cst-a: {ORIGIN}\n", out.getvalue())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Whitespace-only changes.
@@ -0,0 +1,85 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import tarfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from yaml.constructor import ConstructorError
|
||||
|
||||
from cli.administration.roles.paths import WORK_DIR
|
||||
from cli.administration.roles.pull import command
|
||||
from utils.cache import ROLES_DIR
|
||||
from utils.roles.custom import CustomRoleError
|
||||
|
||||
|
||||
class TestPullCommand(unittest.TestCase):
|
||||
@mock.patch.object(
|
||||
command, "pull", return_value=["web-app-cst-demo: repo path @ v1 (commit abc)"]
|
||||
)
|
||||
def test_pulls_the_inventory_into_the_core_roles(self, pull):
|
||||
with contextlib.redirect_stdout(io.StringIO()) as out:
|
||||
self.assertEqual(0, command.main(["inv/devices.yml"]))
|
||||
pull.assert_called_once_with(Path("inv/devices.yml"), ROLES_DIR, WORK_DIR)
|
||||
self.assertIn("web-app-cst-demo: repo path @ v1 (commit abc)", out.getvalue())
|
||||
|
||||
@mock.patch.object(command, "pull", return_value=[])
|
||||
def test_nothing_declared_or_placed_prints_nothing(self, _pull):
|
||||
with contextlib.redirect_stdout(io.StringIO()) as out:
|
||||
command.main(["inv/devices.yml"])
|
||||
self.assertEqual("", out.getvalue())
|
||||
|
||||
def test_unpacking_parsing_and_filesystem_errors_exit_1_naming_them(self):
|
||||
for error in (
|
||||
tarfile.AbsoluteLinkError(tarfile.TarInfo(".infinito-custom-role")),
|
||||
ConstructorError(None, None, "found an undefined tag in inv/hosts.yml"),
|
||||
PermissionError(13, "Permission denied", "roles/web-app-cst-demo"),
|
||||
):
|
||||
with self.subTest(error=type(error).__name__):
|
||||
with (
|
||||
mock.patch.object(command, "pull", side_effect=error),
|
||||
contextlib.redirect_stderr(io.StringIO()) as err,
|
||||
self.assertRaises(SystemExit) as exit_,
|
||||
):
|
||||
command.main(["inv/devices.yml"])
|
||||
self.assertEqual(1, exit_.exception.code)
|
||||
self.assertIn(str(error), err.getvalue())
|
||||
|
||||
@mock.patch.object(
|
||||
command,
|
||||
"pull",
|
||||
side_effect=CustomRoleError(
|
||||
"web-app-cst-demo: git fetch of v1 failed: fatal: unable to access "
|
||||
"'https://deploy:s3cr3t@example.invalid/w.git/'"
|
||||
),
|
||||
)
|
||||
def test_credentials_in_a_url_are_never_echoed(self, _pull):
|
||||
with (
|
||||
contextlib.redirect_stderr(io.StringIO()) as err,
|
||||
self.assertRaises(SystemExit),
|
||||
):
|
||||
command.main(["inv/devices.yml"])
|
||||
self.assertNotIn("s3cr3t", err.getvalue())
|
||||
self.assertIn("web-app-cst-demo: git fetch of v1 failed", err.getvalue())
|
||||
|
||||
@mock.patch.object(
|
||||
command, "pull", side_effect=CustomRoleError("web-app-cst-demo: refused")
|
||||
)
|
||||
def test_refused_pull_exits_1_with_the_reason(self, _pull):
|
||||
with (
|
||||
contextlib.redirect_stderr(io.StringIO()) as err,
|
||||
self.assertRaises(SystemExit) as exit_,
|
||||
):
|
||||
command.main(["inv/devices.yml"])
|
||||
self.assertEqual(1, exit_.exception.code)
|
||||
self.assertIn("web-app-cst-demo: refused", err.getvalue())
|
||||
|
||||
def test_inventory_argument_is_required(self):
|
||||
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
||||
command.main([])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,230 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from cli.administration.roles.declaration import Source, read_declarations
|
||||
|
||||
from ._fixtures import ROLE, ROLE_PATH, declaration, inventory, roles_dir, write
|
||||
|
||||
GIT_URL = "https://github.com/acme/widgets.git"
|
||||
|
||||
|
||||
class TestReadDeclarations(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.root = Path(tmp.name)
|
||||
self.roles = roles_dir(self.root, "web-app-nextcloud")
|
||||
|
||||
def read(self, applications: dict, groups: tuple[str, ...] = (ROLE,)):
|
||||
inv = inventory(self.root / "inventory", list(groups), applications)
|
||||
return read_declarations(inv, self.roles)
|
||||
|
||||
def assert_refused(self, applications: dict, *fragments: str, groups=(ROLE,)):
|
||||
declared, errors = self.read(applications, groups)
|
||||
message = "\n".join(errors)
|
||||
for fragment in fragments:
|
||||
self.assertIn(fragment, message)
|
||||
return declared
|
||||
|
||||
def test_complete_declaration_is_read(self):
|
||||
declared, errors = self.read({ROLE: declaration(GIT_URL)})
|
||||
self.assertEqual([], errors)
|
||||
self.assertEqual({ROLE: Source(GIT_URL, "v1.0.0", ROLE_PATH)}, declared)
|
||||
|
||||
def test_ssh_url_is_accepted(self):
|
||||
_declared, errors = self.read(
|
||||
{ROLE: declaration("git@github.com:acme/widgets.git")}
|
||||
)
|
||||
self.assertEqual([], errors)
|
||||
|
||||
def test_ssh_config_host_alias_is_accepted(self):
|
||||
_declared, errors = self.read(
|
||||
{ROLE: declaration("github-store:acme/widgets.git")}
|
||||
)
|
||||
self.assertEqual([], errors)
|
||||
|
||||
def test_plain_http_url_is_refused(self):
|
||||
self.assert_refused(
|
||||
{ROLE: declaration("http://github.com/acme/widgets.git")},
|
||||
ROLE,
|
||||
"source.url is neither",
|
||||
)
|
||||
|
||||
def test_missing_url_names_group_and_field(self):
|
||||
applications = {
|
||||
ROLE: {"custom": True, "source": {"version": "v1", "path": ROLE_PATH}}
|
||||
}
|
||||
declared = self.assert_refused(applications, ROLE, "source.url is missing")
|
||||
self.assertEqual({}, declared)
|
||||
|
||||
def test_missing_version_of_a_git_url_names_the_field(self):
|
||||
self.assert_refused(
|
||||
{ROLE: declaration(GIT_URL, version=None)},
|
||||
ROLE,
|
||||
"source.version is missing",
|
||||
)
|
||||
|
||||
def test_missing_path_names_the_field(self):
|
||||
applications = {
|
||||
ROLE: {"custom": True, "source": {"url": GIT_URL, "version": "v1"}}
|
||||
}
|
||||
self.assert_refused(applications, ROLE, "source.path is missing")
|
||||
|
||||
def test_missing_source_names_the_field(self):
|
||||
self.assert_refused(
|
||||
{ROLE: {"custom": True}}, ROLE, f"applications.{ROLE}.source is missing"
|
||||
)
|
||||
|
||||
def test_local_directory_needs_no_version(self):
|
||||
declared, errors = self.read({ROLE: declaration(str(self.root), version=None)})
|
||||
self.assertEqual([], errors)
|
||||
self.assertTrue(declared[ROLE].local)
|
||||
self.assertIsNone(declared[ROLE].version)
|
||||
|
||||
def test_url_that_is_neither_git_nor_absolute_is_refused(self):
|
||||
self.assert_refused(
|
||||
{ROLE: declaration("checkouts/widgets")}, ROLE, "source.url is neither"
|
||||
)
|
||||
|
||||
def test_url_with_credentials_is_refused_without_echoing_it(self):
|
||||
for url in (
|
||||
"https://deploy:s3cr3t@github.com/acme/widgets.git",
|
||||
"https://ghp_s3cr3t@github.com/acme/widgets.git",
|
||||
"ssh://git:s3cr3t@github.com/acme/widgets.git",
|
||||
):
|
||||
with self.subTest(url=url.split("@", 1)[1]):
|
||||
declared, errors = self.read({ROLE: declaration(url)})
|
||||
message = "\n".join(errors)
|
||||
self.assertIn(
|
||||
f"{ROLE}: applications.{ROLE}.source.url embeds credentials",
|
||||
message,
|
||||
)
|
||||
self.assertNotIn("s3cr3t", message)
|
||||
self.assertEqual({}, declared)
|
||||
|
||||
def test_ssh_user_without_password_is_no_credential(self):
|
||||
_declared, errors = self.read(
|
||||
{ROLE: declaration("ssh://git@github.com/acme/widgets.git")}
|
||||
)
|
||||
self.assertEqual([], errors)
|
||||
|
||||
def test_jinja_in_a_source_field_is_refused(self):
|
||||
applications = {ROLE: declaration("{{ lookup('env', 'ROLE_URL') }}")}
|
||||
self.assert_refused(applications, ROLE, "source.url holds a Jinja expression")
|
||||
|
||||
def test_jinja_in_custom_is_refused(self):
|
||||
applications = {ROLE: {**declaration(GIT_URL), "custom": "{{ true }}"}}
|
||||
declared = self.assert_refused(
|
||||
applications, ROLE, f"{ROLE}.custom holds a Jinja expression"
|
||||
)
|
||||
self.assertEqual({}, declared)
|
||||
|
||||
def test_unknown_source_field_is_refused(self):
|
||||
applications = {ROLE: declaration(GIT_URL)}
|
||||
applications[ROLE]["source"]["tag"] = "v1"
|
||||
self.assert_refused(applications, ROLE, "source.tag is not a source field")
|
||||
|
||||
def test_path_leaving_the_repository_is_refused(self):
|
||||
applications = {ROLE: declaration(GIT_URL, path=f"../{ROLE}")}
|
||||
self.assert_refused(applications, ROLE, "relative path inside the repository")
|
||||
|
||||
def test_directory_name_other_than_the_group_names_both(self):
|
||||
applications = {ROLE: declaration(GIT_URL, path="ansible/roles/web-app-demo")}
|
||||
self.assert_refused(
|
||||
applications, "'web-app-demo'", f"'{ROLE}'", "never renames"
|
||||
)
|
||||
|
||||
def test_group_without_custom_true_is_refused(self):
|
||||
self.assert_refused(
|
||||
{}, f"{ROLE}: inventory group without applications.{ROLE}.custom: true"
|
||||
)
|
||||
|
||||
def test_custom_false_counts_as_undeclared(self):
|
||||
applications = {ROLE: {**declaration(GIT_URL), "custom": False}}
|
||||
declared = self.assert_refused(applications, f"{ROLE}: inventory group without")
|
||||
self.assertEqual({}, declared)
|
||||
|
||||
def test_custom_true_outside_the_prefix_is_refused(self):
|
||||
applications = {"web-app-demo": declaration(GIT_URL)}
|
||||
self.assert_refused(
|
||||
applications, "web-app-demo: custom: true marks a group outside", groups=()
|
||||
)
|
||||
|
||||
def test_entity_name_of_an_in_tree_role_is_refused(self):
|
||||
applications = {
|
||||
"web-app-cst-nextcloud": declaration(
|
||||
GIT_URL, path="roles/web-app-cst-nextcloud"
|
||||
)
|
||||
}
|
||||
self.assert_refused(
|
||||
applications,
|
||||
"web-app-cst-nextcloud: entity name 'nextcloud'",
|
||||
"in-tree role web-app-nextcloud",
|
||||
groups=("web-app-cst-nextcloud",),
|
||||
)
|
||||
|
||||
def test_declarations_that_differ_between_files_are_refused(self):
|
||||
write(
|
||||
self.root / "inventory/group_vars/all.yml",
|
||||
json.dumps(
|
||||
{"applications": {ROLE: declaration(GIT_URL, version="v2.0.0")}}
|
||||
),
|
||||
)
|
||||
self.assert_refused(
|
||||
{ROLE: declaration(GIT_URL)}, f"applications.{ROLE}.source differs between"
|
||||
)
|
||||
|
||||
def read_yaml(self, host_vars: str):
|
||||
base = self.root / "inventory"
|
||||
write(base / "host_vars/localhost.yml", host_vars)
|
||||
devices = {"all": {"children": {ROLE: {"hosts": {"localhost": {}}}}}}
|
||||
return read_declarations(
|
||||
write(base / "devices.yml", json.dumps(devices)), self.roles
|
||||
)
|
||||
|
||||
def test_tagged_custom_is_refused_naming_the_tag(self):
|
||||
declared, errors = self.read_yaml(
|
||||
f"applications:\n {ROLE}:\n custom: !vault |\n"
|
||||
" $ANSIBLE_VAULT;1.1;AES256\n 6162\n"
|
||||
)
|
||||
self.assertEqual({}, declared)
|
||||
self.assertIn(
|
||||
f"{ROLE}: applications.{ROLE}.custom holds a !vault value",
|
||||
"\n".join(errors),
|
||||
)
|
||||
|
||||
def test_tagged_source_field_is_refused_naming_the_tag(self):
|
||||
declared, errors = self.read_yaml(
|
||||
f"applications:\n {ROLE}:\n custom: true\n source:\n"
|
||||
f" url: {GIT_URL}\n version: !unsafe v1.0.0\n path: {ROLE_PATH}\n"
|
||||
)
|
||||
self.assertEqual({}, declared)
|
||||
self.assertIn(
|
||||
f"{ROLE}: applications.{ROLE}.source.version holds a !unsafe value",
|
||||
"\n".join(errors),
|
||||
)
|
||||
|
||||
def test_unparsable_file_declares_nothing_and_is_the_only_error(self):
|
||||
broken = write(self.root / "inventory/group_vars/all.yml", "applications: [\n")
|
||||
declared, errors = self.read({ROLE: declaration(GIT_URL)})
|
||||
self.assertEqual({}, declared)
|
||||
self.assertEqual(1, len(errors))
|
||||
self.assertTrue(errors[0].startswith(f"{broken}: cannot read"), errors[0])
|
||||
|
||||
def test_settings_without_custom_in_another_file_do_not_conflict(self):
|
||||
write(
|
||||
self.root / "inventory/group_vars/all.yml",
|
||||
json.dumps(
|
||||
{"applications": {ROLE: {"services": {"web": {"enabled": True}}}}}
|
||||
),
|
||||
)
|
||||
_declared, errors = self.read({ROLE: declaration(GIT_URL)})
|
||||
self.assertEqual([], errors)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,114 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from cli.administration.roles.marker import (
|
||||
MARKER_FILE,
|
||||
custom_entries,
|
||||
is_placed,
|
||||
marker_text,
|
||||
placed_roles,
|
||||
provenance,
|
||||
write_marker,
|
||||
)
|
||||
|
||||
from ._fixtures import write
|
||||
|
||||
ORIGIN = "https://example.invalid/w.git ansible/roles/web-app-cst-a @ v1 (commit abc)"
|
||||
PLACED = "web-app-cst-a"
|
||||
|
||||
|
||||
class TestMarker(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.roles = Path(tmp.name) / "roles"
|
||||
write(self.roles / PLACED / MARKER_FILE, marker_text(PLACED, PLACED, ORIGIN))
|
||||
write(self.roles / "web-app-cst-hand" / "mine.yml", "mine: true\n")
|
||||
write(self.roles / "web-app-cst-forged" / MARKER_FILE, "something else\n")
|
||||
shutil.copytree(self.roles / PLACED, self.roles / "web-app-cst-copy")
|
||||
(self.roles / "web-app-cst-linked" / "x").mkdir(parents=True)
|
||||
(self.roles / "web-app-cst-linked" / MARKER_FILE).symlink_to(
|
||||
self.roles / PLACED / MARKER_FILE
|
||||
)
|
||||
write(
|
||||
self.roles / "web-app-nextcloud" / MARKER_FILE,
|
||||
marker_text("web-app-nextcloud", "web-app-nextcloud", ORIGIN),
|
||||
)
|
||||
(self.roles / "web-app-cst-link").symlink_to(self.roles / PLACED)
|
||||
|
||||
def test_only_a_directory_whose_marker_names_it_is_placed(self):
|
||||
self.assertTrue(is_placed(self.roles / PLACED))
|
||||
self.assertFalse(is_placed(self.roles / "web-app-cst-hand"))
|
||||
self.assertFalse(is_placed(self.roles / "web-app-cst-forged"))
|
||||
|
||||
def test_a_copy_under_another_name_is_not_placed(self):
|
||||
self.assertFalse(is_placed(self.roles / "web-app-cst-copy"))
|
||||
|
||||
def test_a_marker_that_is_a_symbolic_link_is_not_read(self):
|
||||
self.assertFalse(is_placed(self.roles / "web-app-cst-linked"))
|
||||
|
||||
def test_a_marker_that_is_a_fifo_is_not_placed(self):
|
||||
fifo = self.roles / "web-app-cst-fifo"
|
||||
fifo.mkdir()
|
||||
os.mkfifo(fifo / MARKER_FILE)
|
||||
self.assertFalse(is_placed(fifo))
|
||||
|
||||
def test_a_symbolic_link_to_a_placed_role_is_not_placed(self):
|
||||
self.assertFalse(is_placed(self.roles / "web-app-cst-link"))
|
||||
|
||||
def test_provenance_reads_back_what_the_pull_recorded(self):
|
||||
self.assertEqual(ORIGIN, provenance(self.roles / PLACED))
|
||||
|
||||
def test_custom_entries_are_every_prefixed_entry(self):
|
||||
self.assertEqual(
|
||||
[
|
||||
PLACED,
|
||||
"web-app-cst-copy",
|
||||
"web-app-cst-forged",
|
||||
"web-app-cst-hand",
|
||||
"web-app-cst-link",
|
||||
"web-app-cst-linked",
|
||||
],
|
||||
[entry.name for entry in custom_entries(self.roles)],
|
||||
)
|
||||
|
||||
def test_placed_roles_are_the_custom_entries_whose_marker_names_them(self):
|
||||
self.assertEqual([PLACED], [entry.name for entry in placed_roles(self.roles)])
|
||||
|
||||
|
||||
class TestWriteMarker(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.root = Path(tmp.name)
|
||||
self.role = self.root / PLACED
|
||||
self.role.mkdir()
|
||||
|
||||
def test_marker_names_the_directory_and_the_group(self):
|
||||
write_marker(self.role, PLACED, ORIGIN)
|
||||
self.assertTrue(is_placed(self.role))
|
||||
self.assertEqual(ORIGIN, provenance(self.role))
|
||||
|
||||
def test_a_symbolic_link_under_the_marker_name_is_replaced_not_followed(self):
|
||||
victim = write(self.root / "victim.yml", "victim: true\n")
|
||||
(self.role / MARKER_FILE).symlink_to(victim)
|
||||
write_marker(self.role, PLACED, ORIGIN)
|
||||
text = victim.read_text(encoding="utf-8") # nocheck: cache-read tempdir
|
||||
self.assertEqual("victim: true\n", text)
|
||||
self.assertFalse((self.role / MARKER_FILE).is_symlink())
|
||||
self.assertTrue(is_placed(self.role))
|
||||
|
||||
def test_a_directory_under_the_marker_name_is_replaced(self):
|
||||
write(self.role / MARKER_FILE / "nested.yml", "nested: true\n")
|
||||
write_marker(self.role, PLACED, ORIGIN)
|
||||
self.assertTrue((self.role / MARKER_FILE).is_file())
|
||||
self.assertTrue(is_placed(self.role))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,489 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from cli.administration.roles import placement
|
||||
from cli.administration.roles.marker import MARKER_FILE, is_placed
|
||||
from cli.administration.roles.placement import clean, pull
|
||||
from utils.cache.files import read_text
|
||||
from utils.roles.custom import CustomRoleError
|
||||
from utils.roles.mapping import (
|
||||
ROLE_FILE_META_MAIN,
|
||||
ROLE_FILE_TASKS_MAIN,
|
||||
ROLE_FILE_VARS_MAIN,
|
||||
)
|
||||
|
||||
from ._fixtures import (
|
||||
ROLE,
|
||||
ROLE_PATH,
|
||||
declaration,
|
||||
git,
|
||||
inventory,
|
||||
local_role,
|
||||
role_repo,
|
||||
roles_dir,
|
||||
write,
|
||||
)
|
||||
|
||||
OTHER = "web-app-cst-other"
|
||||
COPY = "web-app-cst-mywork"
|
||||
LEFT = "left in place, the pull did not place it"
|
||||
|
||||
|
||||
class _PullFixture:
|
||||
"""Mixin with a core roles/, a role repository and a pull helper.
|
||||
|
||||
Combined with ``unittest.TestCase`` on each concrete class so the
|
||||
test-discovery lint sees the ``TestCase`` base directly.
|
||||
"""
|
||||
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.root = Path(tmp.name)
|
||||
self.roles = roles_dir(self.root, "web-app-nextcloud")
|
||||
self.work = self.root / "build" / "custom-roles"
|
||||
self.repo = role_repo(self.root)
|
||||
self.inventories = 0
|
||||
|
||||
def next_inventory(self) -> Path:
|
||||
self.inventories += 1
|
||||
return self.root / f"inventory-{self.inventories}"
|
||||
|
||||
def pull(self, applications: dict, groups: tuple[str, ...] = (ROLE,)) -> list[str]:
|
||||
inv = inventory(self.next_inventory(), list(groups), applications)
|
||||
return pull(inv, self.roles, self.work)
|
||||
|
||||
def pull_yaml(self, devices: str, host_vars: str) -> list[str]:
|
||||
base = self.next_inventory()
|
||||
write(base / "host_vars/localhost.yml", host_vars)
|
||||
return pull(write(base / "devices.yml", devices), self.roles, self.work)
|
||||
|
||||
def placed(self) -> list[str]:
|
||||
return sorted(path.name for path in self.roles.iterdir())
|
||||
|
||||
def leftovers(self) -> list[Path]:
|
||||
if not self.work.exists():
|
||||
return []
|
||||
return [path for path in self.work.iterdir() if path.is_dir()]
|
||||
|
||||
def tag(self, name: str) -> None:
|
||||
git(self.repo, "add", "-A")
|
||||
git(self.repo, "commit", "-q", "-m", name)
|
||||
git(self.repo, "tag", name)
|
||||
|
||||
|
||||
class TestPullPlacement(_PullFixture, unittest.TestCase):
|
||||
def test_role_is_placed_at_its_tag_with_the_resolved_commit(self):
|
||||
lines = self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertEqual(
|
||||
"release: v1\n", read_text(str(self.roles / ROLE / ROLE_FILE_TASKS_MAIN))
|
||||
)
|
||||
self.assertTrue(is_placed(self.roles / ROLE))
|
||||
self.assertRegex(
|
||||
lines[0], rf"^{ROLE}: .* @ v1\.0\.0 \(commit [0-9a-f]{{40}}\)$"
|
||||
)
|
||||
|
||||
def test_placed_role_carries_no_git_metadata(self):
|
||||
self.pull({ROLE: declaration(str(self.repo), path=ROLE_PATH)})
|
||||
self.assertFalse((self.roles / ROLE / ".git").exists())
|
||||
|
||||
def test_local_working_tree_is_placed_without_a_version(self):
|
||||
write(self.repo / ROLE_PATH / ROLE_FILE_TASKS_MAIN, "release: uncommitted\n")
|
||||
self.pull({ROLE: declaration(str(self.repo), version=None)})
|
||||
self.assertEqual(
|
||||
"release: uncommitted\n",
|
||||
read_text(str(self.roles / ROLE / ROLE_FILE_TASKS_MAIN)),
|
||||
)
|
||||
|
||||
def test_pull_again_replaces_the_placed_content(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
write(self.roles / ROLE / "stale.yml", "stale: true\n")
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertFalse((self.roles / ROLE / "stale.yml").exists())
|
||||
self.assertTrue((self.roles / ROLE / ROLE_FILE_VARS_MAIN).is_file())
|
||||
|
||||
def test_role_the_inventory_no_longer_declares_is_removed(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
lines = self.pull({}, groups=())
|
||||
self.assertEqual(["web-app-nextcloud"], self.placed())
|
||||
self.assertIn(f"{ROLE}: removed, the inventory no longer declares it", lines)
|
||||
|
||||
def test_in_tree_roles_are_never_touched(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertEqual([ROLE, "web-app-nextcloud"], self.placed())
|
||||
|
||||
def test_no_staging_or_leftover_directory_remains(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertEqual([ROLE, "web-app-nextcloud"], self.placed())
|
||||
self.assertEqual([], self.leftovers())
|
||||
|
||||
def test_undeclared_directory_the_pull_did_not_place_is_left_alone(self):
|
||||
write(self.roles / OTHER / "keep.yml", "keep: true\n")
|
||||
lines = self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertTrue((self.roles / OTHER / "keep.yml").is_file())
|
||||
self.assertIn(f"{OTHER}: {LEFT}", lines)
|
||||
|
||||
|
||||
class TestPullRefusal(_PullFixture, unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
write(self.roles / ROLE / "sentinel.yml", "placed: true\n")
|
||||
|
||||
def assert_untouched(self) -> None:
|
||||
self.assertTrue((self.roles / ROLE / "sentinel.yml").is_file())
|
||||
self.assertEqual([ROLE, "web-app-nextcloud"], self.placed())
|
||||
self.assertEqual([], self.leftovers())
|
||||
|
||||
def test_branch_as_version_is_refused_before_anything_changes(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "'feature' names a branch"):
|
||||
self.pull({ROLE: declaration(str(self.repo), version="feature")})
|
||||
self.assert_untouched()
|
||||
|
||||
def test_application_id_other_than_the_group_names_both(self):
|
||||
other = role_repo(self.root / "other", application_id="web-app-demo")
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError,
|
||||
f"application_id is 'web-app-demo', but the group is '{ROLE}'",
|
||||
):
|
||||
self.pull({ROLE: declaration(str(other))})
|
||||
self.assert_untouched()
|
||||
|
||||
def test_missing_application_id_is_refused(self):
|
||||
write(self.repo / ROLE_PATH / ROLE_FILE_VARS_MAIN, "other: value\n")
|
||||
with self.assertRaisesRegex(CustomRoleError, "application_id is None"):
|
||||
self.pull({ROLE: declaration(str(self.repo), version=None)})
|
||||
self.assert_untouched()
|
||||
|
||||
def test_source_without_meta_main_is_refused(self):
|
||||
(self.repo / ROLE_PATH / ROLE_FILE_META_MAIN).unlink()
|
||||
with self.assertRaisesRegex(CustomRoleError, "no role directory"):
|
||||
self.pull({ROLE: declaration(str(self.repo), version=None)})
|
||||
self.assert_untouched()
|
||||
|
||||
def test_invalid_declaration_is_refused_before_anything_changes(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "source.path is missing"):
|
||||
self.pull(
|
||||
{
|
||||
ROLE: {
|
||||
"custom": True,
|
||||
"source": {"url": str(self.repo), "version": "v1.0.0"},
|
||||
}
|
||||
}
|
||||
)
|
||||
self.assert_untouched()
|
||||
|
||||
def test_path_that_is_no_directory_in_the_source_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "source.path is not a directory"):
|
||||
self.pull({ROLE: declaration(str(self.repo), path=f"absent/{ROLE}")})
|
||||
self.assert_untouched()
|
||||
|
||||
def test_local_source_inside_its_placement_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
|
||||
self.pull(
|
||||
{ROLE: declaration(str(self.root), version=None, path=f"roles/{ROLE}")}
|
||||
)
|
||||
self.assert_untouched()
|
||||
|
||||
def test_a_failing_role_leaves_every_placed_role_unchanged(self):
|
||||
broken = {
|
||||
"custom": True,
|
||||
"source": {"url": str(self.root / "absent"), "path": f"x/{OTHER}"},
|
||||
}
|
||||
with self.assertRaisesRegex(CustomRoleError, OTHER):
|
||||
self.pull(
|
||||
{ROLE: declaration(str(self.repo)), OTHER: broken}, groups=(ROLE, OTHER)
|
||||
)
|
||||
self.assert_untouched()
|
||||
|
||||
def test_missing_inventory_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "neither a file nor a directory"):
|
||||
pull(self.root / "absent/devices.yml", self.roles, self.work)
|
||||
self.assert_untouched()
|
||||
|
||||
def test_unparsable_vars_file_stops_the_pull_naming_it(self):
|
||||
inv = inventory(self.next_inventory(), [], {})
|
||||
broken = write(inv.parent / "group_vars/all.yml", "applications:\n x: [open\n")
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, rf"{re.escape(str(broken))}: cannot read"
|
||||
):
|
||||
pull(inv, self.roles, self.work)
|
||||
self.assert_untouched()
|
||||
|
||||
def test_vars_file_that_is_no_utf8_stops_the_pull_naming_it(self):
|
||||
inv = inventory(self.next_inventory(), [], {})
|
||||
latin = inv.parent / "group_vars/all.yml"
|
||||
latin.parent.mkdir()
|
||||
latin.write_bytes("applications: {caf\xe9: {}}\n".encode("latin-1"))
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, rf"{re.escape(str(latin))}: cannot read"
|
||||
):
|
||||
pull(inv, self.roles, self.work)
|
||||
self.assert_untouched()
|
||||
|
||||
def test_inventory_file_that_is_no_utf8_stops_the_pull_naming_it(self):
|
||||
inv = inventory(self.next_inventory(), [], {})
|
||||
inv.write_bytes("all:\n hosts:\n caf\xe9: {}\n".encode("latin-1"))
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, rf"{re.escape(str(inv))}: cannot read"
|
||||
):
|
||||
pull(inv, self.roles, self.work)
|
||||
self.assert_untouched()
|
||||
|
||||
|
||||
class TestPullInventoryForms(_PullFixture, unittest.TestCase):
|
||||
def test_directory_inventory_is_read_like_a_file_inventory(self):
|
||||
inv = inventory(
|
||||
self.next_inventory(), [ROLE], {ROLE: declaration(str(self.repo))}
|
||||
)
|
||||
pull(inv.parent, self.roles, self.work)
|
||||
self.assertTrue(is_placed(self.roles / ROLE))
|
||||
|
||||
def test_group_of_a_hosts_file_in_a_directory_inventory_needs_custom_true(self):
|
||||
base = self.next_inventory()
|
||||
write(base / "hosts", f"[{ROLE}]\nlocalhost\n")
|
||||
write(base / "README.md", "[web-app-cst-ignored]\n")
|
||||
write(base / "host_vars/localhost.yml", "applications: {}\n")
|
||||
with self.assertRaises(CustomRoleError) as refused:
|
||||
pull(base, self.roles, self.work)
|
||||
self.assertIn(f"{ROLE}: inventory group without", str(refused.exception))
|
||||
self.assertNotIn("web-app-cst-ignored", str(refused.exception))
|
||||
|
||||
def test_host_list_declares_nothing_and_changes_nothing(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertEqual([], pull(Path("localhost,"), self.roles, self.work))
|
||||
self.assertTrue(is_placed(self.roles / ROLE))
|
||||
|
||||
|
||||
class TestPullAnsibleTags(_PullFixture, unittest.TestCase):
|
||||
DEVICES = (
|
||||
"all:\n"
|
||||
" hosts:\n"
|
||||
" localhost:\n"
|
||||
" motd: !unsafe '{{ literal }}'\n"
|
||||
" children:\n"
|
||||
f" {ROLE}:\n"
|
||||
" hosts:\n"
|
||||
" localhost: {}\n"
|
||||
)
|
||||
|
||||
def host_vars(self, url: str) -> str:
|
||||
return (
|
||||
"secret: !vault |\n"
|
||||
" $ANSIBLE_VAULT;1.1;AES256\n"
|
||||
" 6162\n"
|
||||
"applications:\n"
|
||||
f" {ROLE}:\n"
|
||||
" custom: true\n"
|
||||
" source:\n"
|
||||
f" url: {url}\n"
|
||||
" version: v1.0.0\n"
|
||||
f" path: {ROLE_PATH}\n"
|
||||
)
|
||||
|
||||
def test_tagged_values_elsewhere_in_the_inventory_keep_the_pull_working(self):
|
||||
self.pull_yaml(self.DEVICES, self.host_vars(str(self.repo)))
|
||||
self.assertTrue(is_placed(self.roles / ROLE))
|
||||
|
||||
def test_tagged_source_value_is_refused_like_a_jinja_value(self):
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError,
|
||||
f"{ROLE}: applications.{ROLE}.source.url holds a !unsafe value",
|
||||
):
|
||||
self.pull_yaml(self.DEVICES, self.host_vars(f"!unsafe '{self.repo}'"))
|
||||
self.assertEqual(["web-app-nextcloud"], self.placed())
|
||||
|
||||
|
||||
class TestPullOwnership(_PullFixture, unittest.TestCase):
|
||||
def test_symbolic_link_under_a_declared_name_is_refused_and_kept(self):
|
||||
target = self.root / "elsewhere"
|
||||
write(target / "keep.yml", "keep: true\n")
|
||||
(self.roles / ROLE).symlink_to(target)
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: .* is a symbolic link .* remove the link"
|
||||
):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertTrue((self.roles / ROLE).is_symlink())
|
||||
self.assertTrue((target / "keep.yml").is_file())
|
||||
|
||||
def test_hand_placed_directory_under_a_declared_name_is_refused_and_kept(self):
|
||||
write(self.roles / ROLE / "mine.yml", "mine: true\n")
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: .* was not placed by a pull"
|
||||
):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertTrue((self.roles / ROLE / "mine.yml").is_file())
|
||||
self.assertFalse((self.roles / ROLE / MARKER_FILE).exists())
|
||||
|
||||
def test_copy_of_a_placed_role_is_left_in_place_and_reported(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
shutil.copytree(self.roles / ROLE, self.roles / COPY)
|
||||
lines = self.pull({}, groups=())
|
||||
self.assertEqual([COPY, "web-app-nextcloud"], self.placed())
|
||||
self.assertIn(f"{COPY}: {LEFT}", lines)
|
||||
|
||||
def test_declared_name_that_meets_a_copy_is_refused_and_kept(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
shutil.copytree(self.roles / ROLE, self.roles / COPY)
|
||||
write(self.roles / COPY / "mine.yml", "mine: true\n")
|
||||
checkout = local_role(self.root, COPY)
|
||||
mine = declaration(str(checkout), version=None, path=f"ansible/roles/{COPY}")
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{COPY}: .* was not placed by a pull"
|
||||
):
|
||||
self.pull({ROLE: declaration(str(self.repo)), COPY: mine}, (ROLE, COPY))
|
||||
self.assertTrue((self.roles / COPY / "mine.yml").is_file())
|
||||
|
||||
|
||||
class TestPullMarker(_PullFixture, unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
self.in_tree = write(
|
||||
self.roles / "web-app-nextcloud" / ROLE_FILE_TASKS_MAIN, "in-tree: true\n"
|
||||
)
|
||||
|
||||
def assert_in_tree_untouched(self) -> None:
|
||||
text = self.in_tree.read_text(encoding="utf-8") # nocheck: cache-read tempdir
|
||||
self.assertEqual("in-tree: true\n", text)
|
||||
self.assertEqual(["web-app-nextcloud"], self.placed())
|
||||
self.assertEqual([], self.leftovers())
|
||||
|
||||
def test_marker_link_in_a_local_source_is_refused_and_never_followed(self):
|
||||
checkout = local_role(self.root, ROLE)
|
||||
(checkout / "ansible/roles" / ROLE / MARKER_FILE).symlink_to(self.in_tree)
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: .*{MARKER_FILE} is a symbolic link"
|
||||
):
|
||||
self.pull(
|
||||
{
|
||||
ROLE: declaration(
|
||||
str(checkout), version=None, path=f"ansible/roles/{ROLE}"
|
||||
)
|
||||
}
|
||||
)
|
||||
self.assert_in_tree_untouched()
|
||||
|
||||
def test_relative_marker_link_in_a_tag_is_refused(self):
|
||||
(self.repo / ROLE_PATH / MARKER_FILE).symlink_to(ROLE_FILE_TASKS_MAIN)
|
||||
self.tag("v2.0.0")
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: .*{MARKER_FILE} is a symbolic link"
|
||||
):
|
||||
self.pull({ROLE: declaration(str(self.repo), version="v2.0.0")})
|
||||
self.assert_in_tree_untouched()
|
||||
|
||||
def test_absolute_link_in_a_tag_is_refused_naming_the_group(self):
|
||||
(self.repo / ROLE_PATH / MARKER_FILE).symlink_to(self.in_tree)
|
||||
self.tag("v2.0.0")
|
||||
with self.assertRaisesRegex(CustomRoleError, f"^{ROLE}: .*absolute"):
|
||||
self.pull({ROLE: declaration(str(self.repo), version="v2.0.0")})
|
||||
self.assert_in_tree_untouched()
|
||||
|
||||
|
||||
class TestPullCommit(_PullFixture, unittest.TestCase):
|
||||
def test_interrupt_while_deleting_the_old_copy_keeps_the_new_one(self):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
write(self.roles / ROLE / "stale.yml", "stale: true\n")
|
||||
rmtree = shutil.rmtree
|
||||
|
||||
def interrupted(path, *args, **kwargs):
|
||||
if ".retired." in str(path):
|
||||
raise KeyboardInterrupt
|
||||
return rmtree(path, *args, **kwargs)
|
||||
|
||||
with (
|
||||
mock.patch.object(placement.shutil, "rmtree", interrupted),
|
||||
self.assertRaises(KeyboardInterrupt),
|
||||
):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertTrue(is_placed(self.roles / ROLE))
|
||||
self.assertFalse((self.roles / ROLE / "stale.yml").exists())
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertEqual([], self.leftovers())
|
||||
|
||||
def test_directory_that_appears_at_a_declared_name_during_the_pull_is_kept(self):
|
||||
stage = placement.stage
|
||||
|
||||
def stage_beside_a_hand_placed_copy(group, source, target):
|
||||
write(self.roles / group / "mine.yml", "mine: true\n")
|
||||
return stage(group, source, target)
|
||||
|
||||
with (
|
||||
mock.patch.object(placement, "stage", stage_beside_a_hand_placed_copy),
|
||||
self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: .* was not placed by a pull"
|
||||
),
|
||||
):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
self.assertTrue((self.roles / ROLE / "mine.yml").is_file())
|
||||
self.assertFalse(is_placed(self.roles / ROLE))
|
||||
self.assertEqual([], self.leftovers())
|
||||
|
||||
def test_clean_waits_until_a_running_pull_has_finished(self):
|
||||
events: list[tuple[str, object]] = []
|
||||
cleaner = threading.Thread(
|
||||
target=lambda: events.append(("cleaned", clean(self.roles, self.work)))
|
||||
)
|
||||
stage = placement.stage
|
||||
|
||||
def stage_while_a_clean_starts(group, source, target):
|
||||
cleaner.start()
|
||||
cleaner.join(timeout=0.5)
|
||||
events.append(("clean still waiting", cleaner.is_alive()))
|
||||
return stage(group, source, target)
|
||||
|
||||
with mock.patch.object(placement, "stage", stage_while_a_clean_starts):
|
||||
self.pull({ROLE: declaration(str(self.repo))})
|
||||
cleaner.join(timeout=30)
|
||||
self.assertEqual(("clean still waiting", True), events[0])
|
||||
self.assertEqual("cleaned", events[1][0])
|
||||
self.assertEqual(["web-app-nextcloud"], self.placed())
|
||||
|
||||
|
||||
class TestClean(_PullFixture, unittest.TestCase):
|
||||
def test_removes_only_the_roles_the_pull_placed_and_reports_the_rest(self):
|
||||
checkout = local_role(self.root, OTHER)
|
||||
self.pull(
|
||||
{
|
||||
ROLE: declaration(str(self.repo)),
|
||||
OTHER: declaration(
|
||||
str(checkout), version=None, path=f"ansible/roles/{OTHER}"
|
||||
),
|
||||
},
|
||||
groups=(ROLE, OTHER),
|
||||
)
|
||||
shutil.copytree(self.roles / ROLE, self.roles / COPY)
|
||||
write(self.roles / "web-app-cst-hand" / "mine.yml", "mine: true\n")
|
||||
target = self.root / "elsewhere"
|
||||
write(target / "keep.yml", "keep: true\n")
|
||||
(self.roles / "web-app-cst-link").symlink_to(target)
|
||||
lines = clean(self.roles, self.work)
|
||||
self.assertEqual(
|
||||
[
|
||||
f"{ROLE}: removed",
|
||||
f"web-app-cst-hand: {LEFT}",
|
||||
f"web-app-cst-link: {LEFT}",
|
||||
f"{COPY}: {LEFT}",
|
||||
f"{OTHER}: removed",
|
||||
],
|
||||
lines,
|
||||
)
|
||||
self.assertEqual(
|
||||
["web-app-cst-hand", "web-app-cst-link", COPY, "web-app-nextcloud"],
|
||||
self.placed(),
|
||||
)
|
||||
self.assertTrue((target / "keep.yml").is_file())
|
||||
self.assertEqual([], self.leftovers())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,186 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from cli.administration.roles import source as source_module
|
||||
from cli.administration.roles.declaration import Source
|
||||
from cli.administration.roles.source import check, stage
|
||||
from utils.cache.files import read_text
|
||||
from utils.roles.custom import CustomRoleError
|
||||
from utils.roles.mapping import ROLE_FILE_TASKS_MAIN
|
||||
|
||||
from ._fixtures import ROLE, ROLE_PATH, git, role_repo, roles_dir, write
|
||||
|
||||
|
||||
class _SourceFixture:
|
||||
"""Mixin with a core roles/ and a role repository.
|
||||
|
||||
Combined with ``unittest.TestCase`` on each concrete class so the
|
||||
test-discovery lint sees the ``TestCase`` base directly.
|
||||
"""
|
||||
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.root = Path(tmp.name)
|
||||
self.roles = roles_dir(self.root)
|
||||
self.repo = role_repo(self.root)
|
||||
|
||||
def check(self, source: Source, placements: tuple[Path, ...] = ()) -> None:
|
||||
check(ROLE, source, list(placements), self.roles)
|
||||
|
||||
|
||||
class TestCheckVersion(_SourceFixture, unittest.TestCase):
|
||||
def test_tag_is_accepted(self):
|
||||
self.check(Source(str(self.repo), "v1.0.0", ROLE_PATH))
|
||||
|
||||
def test_full_commit_sha_is_accepted(self):
|
||||
sha = git(self.repo, "rev-parse", "v1.0.0").strip()
|
||||
self.check(Source(str(self.repo), sha, ROLE_PATH))
|
||||
|
||||
def test_full_commit_sha_needs_no_ref_lookup(self):
|
||||
self.check(Source("https://example.invalid/widgets.git", "a" * 40, ROLE_PATH))
|
||||
|
||||
def test_branch_is_refused(self):
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: source.version 'feature' names a branch"
|
||||
):
|
||||
self.check(Source(str(self.repo), "feature", ROLE_PATH))
|
||||
|
||||
def test_abbreviated_sha_is_refused(self):
|
||||
sha = git(self.repo, "rev-parse", "--short", "v1.0.0").strip()
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, "neither a tag nor a full commit SHA"
|
||||
):
|
||||
self.check(Source(str(self.repo), sha, ROLE_PATH))
|
||||
|
||||
def test_unknown_tag_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "'v9.9.9' is neither a tag"):
|
||||
self.check(Source(str(self.repo), "v9.9.9", ROLE_PATH))
|
||||
|
||||
|
||||
class TestCheckLocalSource(_SourceFixture, unittest.TestCase):
|
||||
def test_missing_directory_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "is not a directory"):
|
||||
self.check(Source(str(self.root / "absent"), None, ROLE_PATH))
|
||||
|
||||
def test_own_placement_directory_is_refused_naming_the_path(self):
|
||||
(self.roles / ROLE).mkdir()
|
||||
placement = re.escape(str((self.roles / ROLE).resolve()))
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{placement} lies in the placement directory"
|
||||
):
|
||||
self.check(Source(str(self.root), None, f"roles/{ROLE}"))
|
||||
|
||||
def test_another_placement_is_refused(self):
|
||||
other = self.roles / "web-app-cst-other"
|
||||
other.mkdir()
|
||||
with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
|
||||
self.check(
|
||||
Source(str(self.root), None, f"roles/{other.name}/{ROLE}"), (other,)
|
||||
)
|
||||
|
||||
def test_placement_reached_through_a_symlink_is_refused(self):
|
||||
(self.roles / ROLE).symlink_to(self.repo / ROLE_PATH)
|
||||
with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
|
||||
self.check(Source(str(self.repo), None, ROLE_PATH))
|
||||
|
||||
def test_source_through_a_symlinked_checkout_is_refused(self):
|
||||
alias = self.root / "alias"
|
||||
alias.symlink_to(self.root)
|
||||
with self.assertRaisesRegex(CustomRoleError, "never deletes its own source"):
|
||||
self.check(Source(str(alias), None, f"roles/{ROLE}"))
|
||||
|
||||
def test_checkout_outside_roles_is_accepted(self):
|
||||
self.check(Source(str(self.repo), None, ROLE_PATH))
|
||||
|
||||
|
||||
class TestStage(_SourceFixture, unittest.TestCase):
|
||||
def test_tag_stages_only_the_role_path_at_that_tag(self):
|
||||
target = self.root / "staged"
|
||||
origin = stage(ROLE, Source(str(self.repo), "v1.0.0", ROLE_PATH), target)
|
||||
sha = git(self.repo, "rev-parse", "v1.0.0^{commit}").strip()
|
||||
self.assertEqual("release: v1\n", read_text(str(target / ROLE_FILE_TASKS_MAIN)))
|
||||
self.assertFalse((target / "README.md").exists())
|
||||
self.assertIn(f"(commit {sha})", origin)
|
||||
|
||||
def test_full_commit_sha_is_staged(self):
|
||||
sha = git(self.repo, "rev-parse", "v1.0.0^{commit}").strip()
|
||||
target = self.root / "staged"
|
||||
stage(ROLE, Source(str(self.repo), sha, ROLE_PATH), target)
|
||||
self.assertEqual("release: v1\n", read_text(str(target / ROLE_FILE_TASKS_MAIN)))
|
||||
|
||||
def test_local_directory_without_version_is_its_working_tree(self):
|
||||
write(self.repo / ROLE_PATH / ROLE_FILE_TASKS_MAIN, "release: uncommitted\n")
|
||||
(self.repo / ROLE_PATH / "__pycache__").mkdir()
|
||||
target = self.root / "staged"
|
||||
origin = stage(ROLE, Source(str(self.repo), None, ROLE_PATH), target)
|
||||
self.assertEqual(
|
||||
"release: uncommitted\n", read_text(str(target / ROLE_FILE_TASKS_MAIN))
|
||||
)
|
||||
self.assertFalse((target / "__pycache__").exists())
|
||||
self.assertIn("(working tree)", origin)
|
||||
|
||||
def test_path_that_is_a_file_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "source.path is not a directory"):
|
||||
stage(ROLE, Source(str(self.repo), "v1.0.0", "README.md"), self.root / "s")
|
||||
|
||||
def test_missing_path_is_refused(self):
|
||||
with self.assertRaisesRegex(CustomRoleError, "source.path is not a directory"):
|
||||
stage(
|
||||
ROLE,
|
||||
Source(str(self.repo), "v1.0.0", f"absent/{ROLE}"),
|
||||
self.root / "s",
|
||||
)
|
||||
|
||||
def test_failed_fetch_names_the_group(self):
|
||||
with self.assertRaisesRegex(
|
||||
CustomRoleError, f"{ROLE}: git fetch of v1.0.0 failed"
|
||||
):
|
||||
stage(
|
||||
ROLE,
|
||||
Source(str(self.root / "absent"), "v1.0.0", ROLE_PATH),
|
||||
self.root / "staged",
|
||||
)
|
||||
|
||||
|
||||
class TestGitNeverPrompts(unittest.TestCase):
|
||||
def test_git_runs_detached_without_a_prompt_or_askpass(self):
|
||||
failed = subprocess.CompletedProcess(
|
||||
["git"], 128, stdout=b"", stderr=b"fatal: could not read Username"
|
||||
)
|
||||
source = Source("git@example.invalid:acme/widgets.git", "v1.0.0", ROLE_PATH)
|
||||
with (
|
||||
tempfile.TemporaryDirectory() as tmp,
|
||||
mock.patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"GIT_ASKPASS": "/bin/askpass",
|
||||
"SSH_ASKPASS": "/bin/askpass",
|
||||
"SSH_AUTH_SOCK": "/run/agent.sock",
|
||||
},
|
||||
),
|
||||
mock.patch.object(
|
||||
source_module.subprocess, "run", return_value=failed
|
||||
) as run,
|
||||
self.assertRaisesRegex(CustomRoleError, "could not read Username"),
|
||||
):
|
||||
stage(ROLE, source, Path(tmp) / "staged")
|
||||
kwargs = run.call_args.kwargs
|
||||
self.assertEqual("0", kwargs["env"]["GIT_TERMINAL_PROMPT"])
|
||||
self.assertEqual("never", kwargs["env"]["SSH_ASKPASS_REQUIRE"])
|
||||
self.assertNotIn("GIT_ASKPASS", kwargs["env"])
|
||||
self.assertNotIn("SSH_ASKPASS", kwargs["env"])
|
||||
self.assertEqual("/run/agent.sock", kwargs["env"]["SSH_AUTH_SOCK"])
|
||||
self.assertTrue(kwargs["start_new_session"])
|
||||
self.assertEqual(subprocess.DEVNULL, kwargs["stdin"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user