import json from pathlib import Path import os import stat import tarfile import pytest import enroll.manifest as manifest def test_manifest_writes_roles_and_playbook_with_clean_when(tmp_path: Path): bundle = tmp_path / "bundle" out = tmp_path / "ansible" (bundle / "artifacts" / "foo" / "etc").mkdir(parents=True, exist_ok=True) (bundle / "artifacts" / "foo" / "etc" / "foo.conf").write_text( "x", encoding="utf-8" ) state = { "schema_version": 3, "host": {"hostname": "test", "os": "debian", "pkg_backend": "dpkg"}, "inventory": { "packages": { "foo": { "version": "1.0", "arches": [], "installations": [{"version": "1.0", "arch": "amd64"}], "observed_via": [{"kind": "systemd_unit", "ref": "foo.service"}], "roles": ["foo"], }, "curl": { "version": "8.0", "arches": [], "installations": [{"version": "8.0", "arch": "amd64"}], "observed_via": [{"kind": "package_role", "ref": "curl"}], "roles": ["curl"], }, } }, "roles": { "users": { "role_name": "users", "users": [ { "name": "alice", "uid": 1000, "gid": 1000, "gecos": "Alice", "home": "/home/alice", "shell": "/bin/bash", "primary_group": "alice", "supplementary_groups": ["docker", "qubes"], } ], "managed_files": [], "excluded": [], "notes": [], }, "services": [ { "unit": "foo.service", "role_name": "foo", "packages": ["foo"], "active_state": "inactive", "sub_state": "dead", "unit_file_state": "enabled", "condition_result": "no", "managed_files": [ { "path": "/etc/foo.conf", "src_rel": "etc/foo.conf", "owner": "root", "group": "root", "mode": "0644", "reason": "modified_conffile", } ], "excluded": [], "notes": [], } ], "packages": [ { "package": "curl", "role_name": "curl", "managed_files": [], "excluded": [], "notes": [], } ], "apt_config": { "role_name": "apt_config", "managed_files": [], "excluded": [], "notes": [], }, "dnf_config": { "role_name": "dnf_config", "managed_files": [], "excluded": [], "notes": [], }, "etc_custom": { "role_name": "etc_custom", "managed_files": [ { "path": "/etc/default/keyboard", "src_rel": "etc/default/keyboard", "owner": "root", "group": "root", "mode": "0644", "reason": "custom_unowned", } ], "excluded": [], "notes": [], }, "usr_local_custom": { "role_name": "usr_local_custom", "managed_files": [ { "path": "/usr/local/etc/myapp.conf", "src_rel": "usr/local/etc/myapp.conf", "owner": "root", "group": "root", "mode": "0644", "reason": "usr_local_etc_custom", }, { "path": "/usr/local/bin/myscript", "src_rel": "usr/local/bin/myscript", "owner": "root", "group": "root", "mode": "0755", "reason": "usr_local_bin_script", }, ], "excluded": [], "notes": [], }, "extra_paths": { "role_name": "extra_paths", "include_patterns": [], "exclude_patterns": [], "managed_files": [], "excluded": [], "notes": [], }, }, } bundle.mkdir(parents=True, exist_ok=True) (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8") # Create artifact for etc_custom file so copy works (bundle / "artifacts" / "etc_custom" / "etc" / "default").mkdir( parents=True, exist_ok=True ) (bundle / "artifacts" / "etc_custom" / "etc" / "default" / "keyboard").write_text( "kbd", encoding="utf-8" ) # Create artifacts for usr_local_custom files so copy works (bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "etc").mkdir( parents=True, exist_ok=True ) ( bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "etc" / "myapp.conf" ).write_text("myapp=1\n", encoding="utf-8") (bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "bin").mkdir( parents=True, exist_ok=True ) ( bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "bin" / "myscript" ).write_text("#!/bin/sh\necho hi\n", encoding="utf-8") manifest.manifest(str(bundle), str(out)) # Service role: systemd management should be gated on foo_manage_unit and a probe. tasks = (out / "roles" / "foo" / "tasks" / "main.yml").read_text(encoding="utf-8") assert "- name: Probe whether systemd unit exists and is manageable" in tasks assert "when: foo_manage_unit | default(false)" in tasks assert ( "when:\n - foo_manage_unit | default(false)\n - _unit_probe is succeeded\n" in tasks ) # Ensure we didn't emit deprecated/broken '{{ }}' delimiters in when: lines. for line in tasks.splitlines(): if line.lstrip().startswith("when:"): assert "{{" not in line and "}}" not in line defaults = (out / "roles" / "foo" / "defaults" / "main.yml").read_text( encoding="utf-8" ) assert "foo_manage_unit: true" in defaults assert "foo_systemd_enabled: true" in defaults assert "foo_systemd_state: stopped" in defaults # Playbook should include users, etc_custom, packages, and services pb = (out / "playbook.yml").read_text(encoding="utf-8") assert "- users" in pb assert "- etc_custom" in pb assert "- usr_local_custom" in pb assert "- curl" in pb assert "- foo" in pb def test_manifest_site_mode_creates_host_inventory_and_raw_files(tmp_path: Path): """In --fqdn mode, host-specific state goes into inventory/host_vars.""" fqdn = "host1.example.test" bundle = tmp_path / "bundle" out = tmp_path / "ansible" # Artifacts for a service-managed file. (bundle / "artifacts" / "foo" / "etc").mkdir(parents=True, exist_ok=True) (bundle / "artifacts" / "foo" / "etc" / "foo.conf").write_text( "x", encoding="utf-8" ) # Artifacts for etc_custom file so copy works. (bundle / "artifacts" / "etc_custom" / "etc" / "default").mkdir( parents=True, exist_ok=True ) (bundle / "artifacts" / "etc_custom" / "etc" / "default" / "keyboard").write_text( "kbd", encoding="utf-8" ) state = { "schema_version": 3, "host": {"hostname": "test", "os": "debian", "pkg_backend": "dpkg"}, "inventory": { "packages": { "foo": { "version": "1.0", "arches": [], "installations": [{"version": "1.0", "arch": "amd64"}], "observed_via": [{"kind": "systemd_unit", "ref": "foo.service"}], "roles": ["foo"], } } }, "roles": { "users": { "role_name": "users", "users": [], "managed_files": [], "excluded": [], "notes": [], }, "services": [ { "unit": "foo.service", "role_name": "foo", "packages": ["foo"], "active_state": "active", "sub_state": "running", "unit_file_state": "enabled", "condition_result": "yes", "managed_files": [ { "path": "/etc/foo.conf", "src_rel": "etc/foo.conf", "owner": "root", "group": "root", "mode": "0644", "reason": "modified_conffile", } ], "excluded": [], "notes": [], } ], "packages": [], "apt_config": { "role_name": "apt_config", "managed_files": [], "excluded": [], "notes": [], }, "dnf_config": { "role_name": "dnf_config", "managed_files": [], "excluded": [], "notes": [], }, "etc_custom": { "role_name": "etc_custom", "managed_files": [ { "path": "/etc/default/keyboard", "src_rel": "etc/default/keyboard", "owner": "root", "group": "root", "mode": "0644", "reason": "custom_unowned", } ], "excluded": [], "notes": [], }, "usr_local_custom": { "role_name": "usr_local_custom", "managed_files": [ { "path": "/usr/local/etc/myapp.conf", "src_rel": "usr/local/etc/myapp.conf", "owner": "root", "group": "root", "mode": "0644", "reason": "usr_local_etc_custom", } ], "excluded": [], "notes": [], }, "extra_paths": { "role_name": "extra_paths", "include_patterns": [], "exclude_patterns": [], "managed_files": [], "excluded": [], "notes": [], }, }, } bundle.mkdir(parents=True, exist_ok=True) (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8") # Artifacts for usr_local_custom file so copy works. (bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "etc").mkdir( parents=True, exist_ok=True ) ( bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "etc" / "myapp.conf" ).write_text("myapp=1\n", encoding="utf-8") manifest.manifest(str(bundle), str(out), fqdn=fqdn) # Host playbook exists. assert (out / "playbooks" / f"{fqdn}.yml").exists() # Role defaults are safe/host-agnostic in site mode. foo_defaults = (out / "roles" / "foo" / "defaults" / "main.yml").read_text( encoding="utf-8" ) assert "foo_packages: []" in foo_defaults assert "foo_managed_files: []" in foo_defaults assert "foo_manage_unit: false" in foo_defaults # Host vars contain host-specific state. foo_hostvars = (out / "inventory" / "host_vars" / fqdn / "foo.yml").read_text( encoding="utf-8" ) assert "foo_packages" in foo_hostvars assert "foo_managed_files" in foo_hostvars assert "foo_manage_unit: true" in foo_hostvars assert "foo_systemd_state: started" in foo_hostvars # Non-templated raw config is stored per-host under .files. assert ( out / "inventory" / "host_vars" / fqdn / "foo" / ".files" / "etc" / "foo.conf" ).exists() def test_copy2_replace_overwrites_readonly_destination(tmp_path: Path): """Merging into an existing manifest should tolerate read-only files. Some harvested artifacts (e.g. private keys) may be mode 0400. If a previous run copied them into the destination tree, a subsequent run must still be able to update/replace them. """ import os import stat from enroll.manifest import _copy2_replace src = tmp_path / "src" dst = tmp_path / "dst" src.write_text("new", encoding="utf-8") dst.write_text("old", encoding="utf-8") os.chmod(dst, 0o400) _copy2_replace(str(src), str(dst)) assert dst.read_text(encoding="utf-8") == "new" mode = stat.S_IMODE(dst.stat().st_mode) assert mode & stat.S_IWUSR # destination should remain mergeable def test_manifest_includes_dnf_config_role_when_present(tmp_path: Path): bundle = tmp_path / "bundle" out = tmp_path / "ansible" # Create a dnf_config artifact. (bundle / "artifacts" / "dnf_config" / "etc" / "dnf").mkdir( parents=True, exist_ok=True ) (bundle / "artifacts" / "dnf_config" / "etc" / "dnf" / "dnf.conf").write_text( "[main]\n", encoding="utf-8" ) state = { "schema_version": 3, "host": {"hostname": "test", "os": "redhat", "pkg_backend": "rpm"}, "inventory": { "packages": { "dnf": { "version": "4.0", "arches": [], "installations": [{"version": "4.0", "arch": "x86_64"}], "observed_via": [{"kind": "dnf_config"}], "roles": [], } } }, "roles": { "users": { "role_name": "users", "users": [], "managed_files": [], "excluded": [], "notes": [], }, "services": [], "packages": [], "apt_config": { "role_name": "apt_config", "managed_files": [], "excluded": [], "notes": [], }, "dnf_config": { "role_name": "dnf_config", "managed_files": [ { "path": "/etc/dnf/dnf.conf", "src_rel": "etc/dnf/dnf.conf", "owner": "root", "group": "root", "mode": "0644", "reason": "dnf_config", } ], "excluded": [], "notes": [], }, "etc_custom": { "role_name": "etc_custom", "managed_files": [], "excluded": [], "notes": [], }, "usr_local_custom": { "role_name": "usr_local_custom", "managed_files": [], "excluded": [], "notes": [], }, "extra_paths": { "role_name": "extra_paths", "include_patterns": [], "exclude_patterns": [], "managed_files": [], "excluded": [], "notes": [], }, }, } bundle.mkdir(parents=True, exist_ok=True) (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8") manifest.manifest(str(bundle), str(out)) pb = (out / "playbook.yml").read_text(encoding="utf-8") assert "- dnf_config" in pb tasks = (out / "roles" / "dnf_config" / "tasks" / "main.yml").read_text( encoding="utf-8" ) # Ensure the role exists and contains some file deployment logic. assert "Deploy any other managed files" in tasks def test_render_install_packages_tasks_contains_dnf_branch(): from enroll.manifest import _render_install_packages_tasks txt = _render_install_packages_tasks("role", "role") assert "ansible.builtin.apt" in txt assert "ansible.builtin.dnf" in txt assert "ansible.builtin.package" in txt assert "pkg_mgr" in txt def test_manifest_orders_cron_and_logrotate_at_playbook_tail(tmp_path: Path): """Cron/logrotate roles should appear at the end. The cron role may restore per-user crontabs under /var/spool, so it should run after users have been created. """ bundle = tmp_path / "bundle" out = tmp_path / "ansible" state = { "schema_version": 3, "host": {"hostname": "test", "os": "debian", "pkg_backend": "dpkg"}, "inventory": {"packages": {}}, "roles": { "users": { "role_name": "users", "users": [{"name": "alice"}], "managed_files": [], "excluded": [], "notes": [], }, "services": [], "packages": [ { "package": "curl", "role_name": "curl", "managed_files": [], "excluded": [], "notes": [], }, { "package": "cron", "role_name": "cron", "managed_files": [ { "path": "/var/spool/cron/crontabs/alice", "src_rel": "var/spool/cron/crontabs/alice", "owner": "alice", "group": "root", "mode": "0600", "reason": "system_cron", } ], "excluded": [], "notes": [], }, { "package": "logrotate", "role_name": "logrotate", "managed_files": [ { "path": "/etc/logrotate.conf", "src_rel": "etc/logrotate.conf", "owner": "root", "group": "root", "mode": "0644", "reason": "system_logrotate", } ], "excluded": [], "notes": [], }, ], "apt_config": { "role_name": "apt_config", "managed_files": [], "excluded": [], "notes": [], }, "dnf_config": { "role_name": "dnf_config", "managed_files": [], "excluded": [], "notes": [], }, "etc_custom": { "role_name": "etc_custom", "managed_files": [], "excluded": [], "notes": [], }, "usr_local_custom": { "role_name": "usr_local_custom", "managed_files": [], "excluded": [], "notes": [], }, "extra_paths": { "role_name": "extra_paths", "include_patterns": [], "exclude_patterns": [], "managed_files": [], "excluded": [], "notes": [], }, }, } # Minimal artifacts for managed files. (bundle / "artifacts" / "cron" / "var" / "spool" / "cron" / "crontabs").mkdir( parents=True, exist_ok=True ) ( bundle / "artifacts" / "cron" / "var" / "spool" / "cron" / "crontabs" / "alice" ).write_text("@daily echo hi\n", encoding="utf-8") (bundle / "artifacts" / "logrotate" / "etc").mkdir(parents=True, exist_ok=True) (bundle / "artifacts" / "logrotate" / "etc" / "logrotate.conf").write_text( "weekly\n", encoding="utf-8" ) bundle.mkdir(parents=True, exist_ok=True) (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8") manifest.manifest(str(bundle), str(out)) pb = (out / "playbook.yml").read_text(encoding="utf-8").splitlines() # Roles are emitted as indented list items under the `roles:` key. roles = [ ln.strip().removeprefix("- ").strip() for ln in pb if ln.startswith(" - ") ] # Ensure tail ordering. assert roles[-2:] == ["cron", "logrotate"] assert "users" in roles assert roles.index("users") < roles.index("cron") def test_yaml_helpers_fallback_when_yaml_unavailable(monkeypatch): monkeypatch.setattr(manifest, "_try_yaml", lambda: None) assert manifest._yaml_load_mapping("foo: 1\n") == {} out = manifest._yaml_dump_mapping({"b": 2, "a": 1}) # Best-effort fallback is key: repr(value) assert out.splitlines()[0].startswith("a: ") assert out.endswith("\n") def test_copy2_replace_makes_readonly_sources_user_writable( monkeypatch, tmp_path: Path ): src = tmp_path / "src.txt" dst = tmp_path / "dst.txt" src.write_text("hello", encoding="utf-8") # Make source read-only; copy2 preserves mode, so tmp will be read-only too. os.chmod(src, 0o444) manifest._copy2_replace(str(src), str(dst)) st = os.stat(dst, follow_symlinks=False) assert stat.S_IMODE(st.st_mode) & stat.S_IWUSR def test_prepare_bundle_dir_sops_decrypts_and_extracts(monkeypatch, tmp_path: Path): enc = tmp_path / "harvest.tar.gz.sops" enc.write_text("ignored", encoding="utf-8") def fake_require(): return None def fake_decrypt(src: str, dst: str, *, mode: int = 0o600): # Create a minimal tar.gz with a state.json file. with tarfile.open(dst, "w:gz") as tf: p = tmp_path / "state.json" p.write_text("{}", encoding="utf-8") tf.add(p, arcname="state.json") monkeypatch.setattr(manifest, "require_sops_cmd", fake_require) monkeypatch.setattr(manifest, "decrypt_file_binary_to", fake_decrypt) bundle_dir, td = manifest._prepare_bundle_dir(str(enc), sops_mode=True) try: assert (Path(bundle_dir) / "state.json").exists() finally: td.cleanup() def test_prepare_bundle_dir_rejects_non_dir_without_sops(tmp_path: Path): fp = tmp_path / "bundle.tar.gz" fp.write_text("x", encoding="utf-8") with pytest.raises(RuntimeError): manifest._prepare_bundle_dir(str(fp), sops_mode=False) def test_tar_dir_to_with_progress_writes_progress_when_tty(monkeypatch, tmp_path: Path): src = tmp_path / "dir" src.mkdir() (src / "a.txt").write_text("a", encoding="utf-8") (src / "b.txt").write_text("b", encoding="utf-8") out = tmp_path / "out.tar.gz" writes: list[bytes] = [] monkeypatch.setattr(manifest.os, "isatty", lambda fd: True) monkeypatch.setattr(manifest.os, "write", lambda fd, b: writes.append(b) or len(b)) manifest._tar_dir_to_with_progress(str(src), str(out), desc="tarring") assert out.exists() assert writes # progress was written assert writes[-1].endswith(b"\n") def test_encrypt_manifest_out_dir_to_sops_handles_missing_tmp_cleanup( monkeypatch, tmp_path: Path ): src_dir = tmp_path / "manifest" src_dir.mkdir() (src_dir / "x.txt").write_text("x", encoding="utf-8") out = tmp_path / "manifest.tar.gz.sops" monkeypatch.setattr(manifest, "require_sops_cmd", lambda: None) def fake_encrypt(in_fp, out_fp, *args, **kwargs): Path(out_fp).write_text("enc", encoding="utf-8") monkeypatch.setattr(manifest, "encrypt_file_binary", fake_encrypt) # Simulate race where tmp tar is already removed. monkeypatch.setattr( manifest.os, "unlink", lambda p: (_ for _ in ()).throw(FileNotFoundError()) ) res = manifest._encrypt_manifest_out_dir_to_sops(str(src_dir), str(out), ["ABC"]) # type: ignore[arg-type] assert str(res).endswith(".sops") assert out.exists() def test_manifest_applies_jinjaturtle_to_jinjifyable_managed_file( monkeypatch, tmp_path: Path ): # Create a minimal bundle with just an apt_config snapshot. bundle = tmp_path / "bundle" (bundle / "artifacts" / "apt_config" / "etc" / "apt").mkdir(parents=True) (bundle / "artifacts" / "apt_config" / "etc" / "apt" / "foo.ini").write_text( "key=VALUE\n", encoding="utf-8" ) state = { "schema_version": 1, "inventory": {"packages": {}}, "roles": { "services": [], "packages": [], "apt_config": { "role_name": "apt_config", "managed_files": [ { "path": "/etc/apt/foo.ini", "src_rel": "etc/apt/foo.ini", "owner": "root", "group": "root", "mode": "0644", "reason": "apt_config", } ], "managed_dirs": [], "excluded": [], "notes": [], }, }, } (bundle / "state.json").write_text( __import__("json").dumps(state), encoding="utf-8" ) monkeypatch.setattr(manifest, "find_jinjaturtle_cmd", lambda: "jinjaturtle") class _Res: template_text = "key={{ foo }}\n" vars_text = "foo: 123\n" monkeypatch.setattr(manifest, "run_jinjaturtle", lambda *a, **k: _Res()) out_dir = tmp_path / "out" manifest.manifest(str(bundle), str(out_dir), jinjaturtle="on") tmpl = out_dir / "roles" / "apt_config" / "templates" / "etc" / "apt" / "foo.ini.j2" assert tmpl.exists() assert "{{ foo }}" in tmpl.read_text(encoding="utf-8") defaults = out_dir / "roles" / "apt_config" / "defaults" / "main.yml" txt = defaults.read_text(encoding="utf-8") assert "foo: 123" in txt # Non-templated file should not exist under files/. assert not ( out_dir / "roles" / "apt_config" / "files" / "etc" / "apt" / "foo.ini" ).exists()