computer-playbook/tests/unit/python/utils/cache/users/test_module.py

"""Focused unit tests for ``utils.cache.users``.

Pins the public API (`get_user_defaults`, `get_merged_users`) plus the
internal helpers that only this module owns: `_load_user_defs`,
`_build_users`, `_validate_forwards`, `_compute_reserved_usernames`, `_load_store_users`,
`_resolve_tokens_file`, `_hydrate_users_tokens`, `_merge_users`,
`_materialize_domain_label_reservations`. Module name is `test_users_module`
to disambiguate from the existing `tests/unit/python/plugins/lookup/test_users.py`
which targets the lookup plugin.
"""

from __future__ import annotations

import tempfile
import textwrap
import unittest
import unittest.mock as mock
from collections import OrderedDict
from pathlib import Path

from utils.cache import _reset_cache_for_tests
from utils.cache import base as cache_base
from utils.cache import users as cache_users
from utils.cache.carrier import USERS_RENDERED_FACT, merged_users_cache_key
from utils.cache.users import store as cache_users_store
from utils.cache.yaml import dump_yaml_str
from utils.domains.default_primary import default_domain_primary
from utils.roles.mapping import ROLE_FILE_META_SERVICES, ROLE_FILE_META_USERS

DOMAIN = default_domain_primary()


def _write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(textwrap.dedent(content), encoding="utf-8")


def _seed_minimal_user_role(tmp: Path, role_name: str = "web-app-foo") -> Path:
    roles = tmp / "roles"
    role = roles / role_name
    _write(role / ROLE_FILE_META_SERVICES, "{}\n")
    _write(
        role / ROLE_FILE_META_USERS,
        f"""
        {role_name.rsplit("-", maxsplit=1)[-1]}:
          description: "test-only user"
        """,
    )
    return roles


class TestComputeReservedUsernames(unittest.TestCase):
    def test_extracts_lowercase_alnum_suffix(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = Path(tmp)
            (roles / "web-app-postgres").mkdir(parents=True)
            (roles / "web-app-mailu").mkdir(parents=True)
            (roles / "svc-db-mariadb").mkdir(parents=True)
            (roles / "not-a-role.txt").write_text("", encoding="utf-8")
            self.assertEqual(
                cache_users._compute_reserved_usernames(roles),
                ["mailu", "mariadb", "postgres"],
            )


class TestLoadUserDefs(unittest.TestCase):
    def test_aggregates_users_across_roles(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-alpha")
            _write(
                roles / "web-app-beta" / ROLE_FILE_META_USERS,
                """
                beta:
                  description: "beta user"
                """,
            )
            defs = cache_users._load_user_defs(roles)
            self.assertIn("alpha", defs)
            self.assertIn("beta", defs)

    def test_conflicting_definitions_raise(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = Path(tmp) / "roles"
            _write(
                roles / "web-app-a" / ROLE_FILE_META_USERS,
                """
                shared: {description: from-a}
                """,
            )
            _write(
                roles / "web-app-b" / ROLE_FILE_META_USERS,
                """
                shared: {description: from-b}
                """,
            )
            with self.assertRaisesRegex(ValueError, "Conflict for user 'shared'"):
                cache_users._load_user_defs(roles)

    def test_two_roles_register_a_token_for_the_same_user(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = Path(tmp) / "roles"
            _write(
                roles / "web-app-a" / ROLE_FILE_META_USERS,
                """
                administrator: {tokens: {web-app-a: ""}}
                """,
            )
            _write(
                roles / "web-app-b" / ROLE_FILE_META_USERS,
                """
                administrator: {tokens: {web-app-b: ""}}
                """,
            )
            defs = cache_users._load_user_defs(roles)
            self.assertEqual(
                sorted(defs["administrator"]["tokens"]), ["web-app-a", "web-app-b"]
            )

    def test_the_same_token_key_with_two_values_still_conflicts(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = Path(tmp) / "roles"
            _write(
                roles / "web-app-a" / ROLE_FILE_META_USERS,
                """
                administrator: {tokens: {shared-app: from-a}}
                """,
            )
            _write(
                roles / "web-app-b" / ROLE_FILE_META_USERS,
                """
                administrator: {tokens: {shared-app: from-b}}
                """,
            )
            with self.assertRaisesRegex(ValueError, "tokens.shared-app"):
                cache_users._load_user_defs(roles)


class TestBuildUsers(unittest.TestCase):
    def test_allocates_distinct_uids_for_users_without_explicit_uid(self):
        defs: OrderedDict[str, dict] = OrderedDict(
            [
                ("alice", {}),
                ("bob", {}),
            ]
        )
        users = cache_users._build_users(
            defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
        )
        self.assertEqual(users["alice"]["uid"], 1001)
        self.assertEqual(users["bob"]["uid"], 1002)

    def test_explicit_uid_is_honoured(self):
        defs = OrderedDict(
            [
                ("alice", {"uid": 5000}),
                ("bob", {}),
            ]
        )
        users = cache_users._build_users(
            defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
        )
        self.assertEqual(users["alice"]["uid"], 5000)
        self.assertEqual(users["bob"]["uid"], 1001)

    def test_duplicate_explicit_uid_raises(self):
        defs = OrderedDict(
            [
                ("alice", {"uid": 1500}),
                ("bob", {"uid": 1500}),
            ]
        )
        with self.assertRaisesRegex(ValueError, "Duplicate uid 1500"):
            cache_users._build_users(
                defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
            )

    def test_duplicate_username_after_overrides_raises(self):
        defs = OrderedDict(
            [
                ("alice", {"username": "shared"}),
                ("bob", {"username": "shared"}),
            ]
        )
        with self.assertRaisesRegex(ValueError, "Duplicate username 'shared'"):
            cache_users._build_users(
                defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
            )

    def test_default_email_uses_username_at_primary_domain(self):
        users = cache_users._build_users(
            OrderedDict([("alice", {})]),
            primary_domain="example.org",
            start_id=1001,
            become_pwd="pw",
        )
        self.assertEqual(users["alice"]["email"], "alice@example.org")

    def test_accounts_derivation_and_forward_defaults(self):
        defs = OrderedDict(
            [
                ("plain", {}),
                ("botuser", {"roles": ["bot"]}),
                ("adminuser", {"roles": ["administrator"]}),
                ("pinned", {"accounts": ["host"], "forward": "adminuser"}),
                ("empty", {"accounts": []}),
            ]
        )
        users = cache_users._build_users(
            defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
        )
        self.assertEqual(users["plain"]["accounts"], ["identity"])
        self.assertEqual(users["botuser"]["accounts"], ["mailbox", "identity"])
        self.assertEqual(users["adminuser"]["accounts"], ["mailbox", "identity"])
        self.assertEqual(users["pinned"]["accounts"], ["host"])
        self.assertEqual(users["empty"]["accounts"], [])
        self.assertEqual(users["plain"]["forward"], "")
        self.assertEqual(users["pinned"]["forward"], "adminuser")

    def test_forward_to_an_unknown_user_raises(self):
        defs = OrderedDict([("pinned", {"accounts": [], "forward": "nobody"})])
        with self.assertRaisesRegex(ValueError, "'pinned' forwards to unknown user"):
            cache_users._build_users(
                defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
            )


class TestValidateForwards(unittest.TestCase):
    @staticmethod
    def _user(forward="", accounts=("mailbox",)):
        return {"forward": forward, "accounts": list(accounts)}

    def test_users_without_forward_pass(self):
        cache_users._validate_forwards({"a": self._user(), "b": self._user()})

    def test_forward_to_a_known_user_passes(self):
        cache_users._validate_forwards(
            {"a": self._user("b", accounts=()), "b": self._user()}
        )

    def test_chain_of_forwards_passes(self):
        cache_users._validate_forwards(
            {"a": self._user("b", accounts=()), "b": self._user("c"), "c": self._user()}
        )

    def test_forward_to_a_user_without_mailbox_passes(self):
        cache_users._validate_forwards(
            {"a": self._user("b"), "b": self._user(accounts=("host", "identity"))}
        )

    def test_forward_to_unknown_user_raises(self):
        with self.assertRaisesRegex(ValueError, "'a' forwards to unknown user 'x'"):
            cache_users._validate_forwards({"a": self._user("x")})

    def test_unrendered_forward_is_left_to_the_consumer(self):
        cache_users._validate_forwards({"a": self._user("{{ 'nobody' }}")})

    def test_unrendered_statement_forward_is_left_to_the_consumer(self):
        cache_users._validate_forwards({"a": self._user("{% if x %}nobody{% endif %}")})

    def test_unrendered_forward_ends_the_chain(self):
        cache_users._validate_forwards(
            {"a": self._user("b"), "b": self._user("{{ 'a' }}")}
        )

    def test_forward_to_a_non_mapping_entry_raises_a_clear_error(self):
        with self.assertRaisesRegex(
            TypeError, "'a' forwards to 'b', whose entry is None"
        ):
            cache_users._validate_forwards({"a": self._user("b"), "b": None})

    def test_forward_to_itself_raises(self):
        with self.assertRaisesRegex(ValueError, "Forward loop: a -> a"):
            cache_users._validate_forwards({"a": self._user("a")})

    def test_forward_cycle_raises(self):
        with self.assertRaisesRegex(ValueError, "Forward loop: a -> b -> a"):
            cache_users._validate_forwards({"a": self._user("b"), "b": self._user("a")})

    def test_non_string_forward_raises(self):
        with self.assertRaisesRegex(TypeError, "a user key is required"):
            cache_users._validate_forwards({"a": self._user(["b"]), "b": self._user()})

    def test_non_mapping_user_is_skipped(self):
        cache_users._validate_forwards({"a": None, "b": self._user()})


class TestMergeUsers(unittest.TestCase):
    def test_overrides_extend_defaults(self):
        defaults = {"alice": {"description": "from-defaults"}}
        overrides = {"alice": {"username": "alice-override"}, "bob": {}}
        merged = cache_users._merge_users(defaults, overrides)
        self.assertEqual(merged["alice"]["username"], "alice-override")
        self.assertEqual(merged["alice"]["description"], "from-defaults")
        self.assertIn("bob", merged)

    def test_apply_account_defaults_covers_inventory_only_users(self):
        merged = cache_users._apply_account_defaults(
            {
                "invuser": {"username": "invuser"},
                "invbot": {"username": "invbot", "roles": ["bot"]},
                "pinned": {
                    "username": "pinned",
                    "accounts": [],
                    "forward": "x",
                    "email": "pinned@example.org",
                },
            }
        )
        self.assertEqual(merged["invuser"]["email"], "invuser@{{ DOMAIN_PRIMARY }}")
        self.assertEqual(merged["pinned"]["email"], "pinned@example.org")
        self.assertEqual(merged["invuser"]["accounts"], ["identity"])
        self.assertEqual(merged["invuser"]["forward"], "")
        self.assertEqual(merged["invbot"]["accounts"], ["mailbox", "identity"])
        self.assertEqual(merged["pinned"]["accounts"], [])
        self.assertEqual(merged["pinned"]["forward"], "x")


class TestHydrateUsersTokens(unittest.TestCase):
    def test_store_tokens_fill_empty_in_users_only(self):
        users = {
            "alice": {"tokens": {"web-app-x": ""}},
            "bob": {"tokens": {"web-app-y": "from-user"}},
        }
        store = {
            "alice": {"tokens": {"web-app-x": "from-store"}},
            "bob": {"tokens": {"web-app-y": "store-attempt"}},
        }
        merged = cache_users._hydrate_users_tokens(users, store)
        self.assertEqual(merged["alice"]["tokens"]["web-app-x"], "from-store")
        self.assertEqual(merged["bob"]["tokens"]["web-app-y"], "from-user")

    def test_returns_deep_copy_independent_of_input(self):
        users = {"alice": {"tokens": {}}}
        store = {"alice": {"tokens": {"x": "y"}}}
        merged = cache_users._hydrate_users_tokens(users, store)
        merged["alice"]["tokens"]["x"] = "mutated"
        self.assertEqual(store["alice"]["tokens"]["x"], "y")


class TestResolveTokensFile(unittest.TestCase):
    def test_falls_back_to_default_when_no_overrides(self):
        previous = cache_base.DEFAULT_TOKENS_FILE
        with tempfile.TemporaryDirectory() as tmp:
            sentinel = Path(tmp) / "sentinel.yml"
            sentinel.write_text("users: {}\n", encoding="utf-8")
            cache_base.DEFAULT_TOKENS_FILE = sentinel
            try:
                result = cache_users._resolve_tokens_file(variables={})
                self.assertEqual(result, sentinel)
            finally:
                cache_base.DEFAULT_TOKENS_FILE = previous

    def test_explicit_file_tokens_wins(self):
        with tempfile.TemporaryDirectory() as tmp:
            explicit = Path(tmp) / "tokens.yml"
            explicit.write_text("users: {}\n", encoding="utf-8")
            result = cache_users._resolve_tokens_file(
                variables={"FILE_TOKENS": str(explicit)}
            )
            self.assertEqual(result, explicit)


class TestLoadStoreUsers(unittest.TestCase):
    def test_missing_path_returns_empty(self):
        self.assertEqual(cache_users._load_store_users(None), {})
        self.assertEqual(cache_users._load_store_users(""), {})

    def test_yaml_users_block_passes_through(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "tokens.yml"
            path.write_text(
                dump_yaml_str({"users": {"alice": {"tokens": {"x": "y"}}}}),
                encoding="utf-8",
            )
            result = cache_users._load_store_users(path)
            self.assertEqual(result, {"alice": {"tokens": {"x": "y"}}})


class TestMaterializeBuiltinUserAliases(unittest.TestCase):
    def test_no_primary_domain_returns_users_unchanged(self):
        users = {"alice": {"username": "alice"}}
        out = cache_users._materialize_domain_label_reservations(
            users, variables={}, templar=None
        )
        self.assertEqual(out, users)

    def test_every_domain_label_becomes_a_reserved_username(self):
        out = cache_users._materialize_domain_label_reservations(
            {},
            variables={"DOMAIN_PRIMARY": DOMAIN},
            templar=None,
        )
        self.assertEqual(
            [out[key]["username"] for key in sorted(out)],
            DOMAIN.split("."),
        )
        self.assertTrue(all(out[key]["accounts"] == [] for key in out))


class TestGetUserDefaults(unittest.TestCase):
    def setUp(self) -> None:
        _reset_cache_for_tests()

    def test_returns_users_per_role(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            defaults = cache_users.get_user_defaults(roles_dir=roles)
            self.assertIn("foo", defaults)
            self.assertEqual(defaults["foo"]["description"], "test-only user")

    def test_caches_per_roles_dir(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            first = cache_users.get_user_defaults(roles_dir=roles)
            first["foo"]["mutated"] = True
            second = cache_users.get_user_defaults(roles_dir=roles)
            self.assertNotIn("mutated", second["foo"])

    def test_reserved_usernames_added_when_missing(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            (roles / "svc-db-mariadb").mkdir(parents=True)
            defaults = cache_users.get_user_defaults(roles_dir=roles)
            self.assertIn("mariadb", defaults)
            self.assertEqual(defaults["mariadb"]["accounts"], [])


class TestGetMergedUsersWithOverrides(unittest.TestCase):
    def setUp(self) -> None:
        _reset_cache_for_tests()

    def test_runtime_override_wins_for_username(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            merged = cache_users.get_merged_users(
                variables={
                    "users": {
                        "foo": {"username": "override-user"},
                    }
                },
                roles_dir=roles,
                templar=None,
            )
            self.assertEqual(merged["foo"]["username"], "override-user")

    def test_runtime_forward_to_an_unknown_user_raises(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            with self.assertRaisesRegex(ValueError, "'foo' forwards to unknown user"):
                cache_users.get_merged_users(
                    variables={"users": {"foo": {"forward": "nobody"}}},
                    roles_dir=roles,
                    templar=None,
                )

    def test_runtime_forward_to_an_inventory_mailbox_is_kept(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            merged = cache_users.get_merged_users(
                variables={
                    "users": {
                        "foo": {"forward": "inbox"},
                        "inbox": {"accounts": ["mailbox"]},
                    }
                },
                roles_dir=roles,
                templar=None,
            )
            self.assertEqual(merged["foo"]["forward"], "inbox")

    def test_runtime_target_without_mailbox_keeps_the_lookup(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            merged = cache_users.get_merged_users(
                variables={
                    "users": {
                        "foo": {"forward": "workstation"},
                        "workstation": {"accounts": ["host", "identity"]},
                    }
                },
                roles_dir=roles,
                templar=None,
            )
            self.assertEqual(merged["workstation"]["accounts"], ["host", "identity"])

    def test_runtime_templated_forward_is_not_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            merged = cache_users.get_merged_users(
                variables={"users": {"foo": {"forward": "{{ 'inbox' }}"}}},
                roles_dir=roles,
                templar=None,
            )
            self.assertEqual(merged["foo"]["forward"], "{{ 'inbox' }}")

    def test_runtime_null_target_raises_a_clear_error(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
            with self.assertRaisesRegex(
                TypeError, "'foo' forwards to 'inbox', whose entry"
            ):
                cache_users.get_merged_users(
                    variables={"users": {"foo": {"forward": "inbox"}, "inbox": None}},
                    roles_dir=roles,
                    templar=None,
                )


class TestGetMergedUsersTokenSource(unittest.TestCase):
    def setUp(self) -> None:
        _reset_cache_for_tests()

    def _merged(self, roles: Path, tokens_file: Path, **extra) -> dict:
        return cache_users.get_merged_users(
            variables={"FILE_TOKENS": str(tokens_file), **extra},
            roles_dir=roles,
            templar=None,
        )

    def _seed(self, tmp: Path) -> tuple[Path, Path]:
        roles = _seed_minimal_user_role(tmp, "web-app-foo")
        tokens_file = tmp / "tokens.yml"
        tokens_file.write_text(
            dump_yaml_str({"users": {"foo": {"tokens": {"web-app-foo": "from-file"}}}}),
            encoding="utf-8",
        )
        return roles, tokens_file

    def test_the_store_fact_of_the_play_wins_over_the_controller_file(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles, tokens_file = self._seed(Path(tmp))
            merged = self._merged(
                roles,
                tokens_file,
                sys_token_store_users={
                    "foo": {"tokens": {"web-app-foo": "from-stack-host"}}
                },
            )
            self.assertEqual(merged["foo"]["tokens"]["web-app-foo"], "from-stack-host")

    def test_without_the_fact_the_controller_file_is_read(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles, tokens_file = self._seed(Path(tmp))
            merged = self._merged(roles, tokens_file)
            self.assertEqual(merged["foo"]["tokens"]["web-app-foo"], "from-file")

    def test_a_written_token_reaches_the_next_lookup(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles, tokens_file = self._seed(Path(tmp))
            before = self._merged(roles, tokens_file, sys_token_store_users={})
            after = self._merged(
                roles,
                tokens_file,
                sys_token_store_users={"foo": {"tokens": {"web-app-foo": "minted"}}},
            )
            self.assertNotIn("web-app-foo", before["foo"].get("tokens", {}))
            self.assertEqual(after["foo"]["tokens"]["web-app-foo"], "minted")


class _Templar:
    def __init__(self) -> None:
        self.available_variables: dict = {}

    def template(self, value, fail_on_undefined=False):
        return value


class TestGetMergedUsersCarrierFact(unittest.TestCase):
    def setUp(self) -> None:
        _reset_cache_for_tests()

    @staticmethod
    def _variables(host: str = "h1") -> dict:
        return {
            "inventory_hostname": host,
            "users": {"foo": {"email": "foo@example.org"}},
            cache_users_store.STORE_FACT: {
                "foo": {"tokens": {"web-app-foo": "minted"}}
            },
        }

    @staticmethod
    def _park(roles: Path, variables: dict, payload: dict) -> dict:
        key = merged_users_cache_key(variables, roles_dir=roles)
        variables[USERS_RENDERED_FACT] = {
            "key": [key[0], list(key[1])],
            "users": payload,
        }
        return payload

    @staticmethod
    def _merged(roles: Path, variables: dict, templar=None) -> tuple:
        with mock.patch.object(
            cache_users,
            "_render_with_templar",
            side_effect=lambda value, **_: {"rendered": value},
        ) as render:
            merged = cache_users.get_merged_users(
                variables=variables,
                roles_dir=roles,
                templar=_Templar() if templar is None else templar,
            )
        return merged, render.call_count

    def test_matching_key_serves_the_carried_payload_without_rendering(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables()
            carried = self._park(roles, variables, {"foo": {"carried": True}})
            merged, renders = self._merged(roles, variables)
            self.assertIs(merged, carried)
            self.assertEqual(renders, 0)

    def test_the_carried_payload_answers_the_rest_of_the_process(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables()
            carried = self._park(roles, variables, {"foo": {"carried": True}})
            self._merged(roles, variables)
            del variables[USERS_RENDERED_FACT]
            merged, renders = self._merged(roles, variables)
            self.assertIs(merged, carried)
            self.assertEqual(renders, 0)

    def test_stale_key_falls_back_to_a_fresh_render(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables()
            key = merged_users_cache_key(variables, roles_dir=roles)
            variables[USERS_RENDERED_FACT] = {
                "key": [key[0], ["stale"] * len(key[1])],
                "users": {"foo": {"carried": True}},
            }
            merged, renders = self._merged(roles, variables)
            self.assertEqual(renders, 1)
            self.assertEqual(merged["rendered"]["foo"]["email"], "foo@example.org")

    def test_another_hosts_carrier_is_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables("h1")
            self._park(roles, variables, {"foo": {"carried": True}})
            variables["inventory_hostname"] = "h2"
            merged, renders = self._merged(roles, variables)
            self.assertEqual(renders, 1)
            self.assertNotIn("carried", merged["rendered"]["foo"])

    def test_a_rotated_store_token_rejects_the_carrier(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables()
            self._park(roles, variables, {"foo": {"carried": True}})
            variables[cache_users_store.STORE_FACT] = {
                "foo": {"tokens": {"web-app-foo": "rotated"}}
            }
            merged, renders = self._merged(roles, variables)
            self.assertEqual(renders, 1)
            self.assertEqual(
                merged["rendered"]["foo"]["tokens"]["web-app-foo"], "rotated"
            )

    def _park_from_the_tokens_file(self, tmp: Path, token: str) -> tuple:
        roles = _seed_minimal_user_role(tmp)
        tokens = tmp / "tokens.yml"
        tokens.write_text(
            f"users: {{foo: {{tokens: {{web-app-foo: {token}}}}}}}\n",
            encoding="utf-8",
        )
        variables = self._variables()
        del variables[cache_users_store.STORE_FACT]
        variables["FILE_TOKENS"] = str(tokens)
        carried = self._park(roles, variables, {"foo": {"carried": True}})
        return roles, variables, tokens, carried

    def test_a_changed_tokens_file_rejects_the_carrier(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles, variables, tokens, _ = self._park_from_the_tokens_file(
                Path(tmp), "minted"
            )
            tokens.write_text(
                "users: {foo: {tokens: {web-app-foo: rotated-token}}}\n",
                encoding="utf-8",
            )
            merged, renders = self._merged(roles, variables)
            self.assertEqual(renders, 1)
            self.assertEqual(
                merged["rendered"]["foo"]["tokens"]["web-app-foo"], "rotated-token"
            )

    def test_the_store_fact_with_the_file_tokens_keeps_the_carrier(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles, variables, _, carried = self._park_from_the_tokens_file(
                Path(tmp), "minted"
            )
            variables[cache_users_store.STORE_FACT] = {
                "foo": {"tokens": {"web-app-foo": "minted"}}
            }
            merged, renders = self._merged(roles, variables)
            self.assertIs(merged, carried)
            self.assertEqual(renders, 0)

    def test_a_rotated_inventory_password_rejects_the_carrier(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables()
            variables["users"]["foo"]["password"] = "before"
            self._park(roles, variables, {"foo": {"carried": True}})
            variables["users"] = {"foo": {"password": "after"}}
            merged, renders = self._merged(roles, variables)
            self.assertEqual(renders, 1)
            self.assertEqual(merged["rendered"]["foo"]["password"], "after")

    def test_without_a_templar_the_carrier_is_ignored(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            variables = self._variables()
            self._park(roles, variables, {"foo": {"carried": True}})
            merged = cache_users.get_merged_users(
                variables=variables, roles_dir=roles, templar=None
            )
            self.assertNotIn("carried", merged["foo"])
            self.assertEqual(merged["foo"]["email"], "foo@example.org")

    def test_malformed_carrier_is_ignored(self):
        with tempfile.TemporaryDirectory() as tmp:
            roles = _seed_minimal_user_role(Path(tmp))
            key = merged_users_cache_key(self._variables(), roles_dir=roles)
            for carrier in (
                "not-a-mapping",
                {"key": "x", "users": {}},
                {"key": [key[0], list(key[1])], "users": "not-a-mapping"},
                {},
            ):
                with self.subTest(carrier=carrier):
                    _reset_cache_for_tests()
                    variables = self._variables()
                    variables[USERS_RENDERED_FACT] = carrier
                    merged, renders = self._merged(roles, variables)
                    self.assertEqual(renders, 1)
                    self.assertEqual(
                        merged["rendered"]["foo"]["email"], "foo@example.org"
                    )


class TestStoreSignature(unittest.TestCase):
    def setUp(self) -> None:
        _reset_cache_for_tests()

    @staticmethod
    def _signature(store: dict) -> tuple:
        with tempfile.TemporaryDirectory() as tmp:
            return cache_users._store_signature(
                {cache_users_store.STORE_FACT: store}, Path(tmp) / "tokens.yml"
            )

    @staticmethod
    def _hydrated(store: dict) -> str:
        users = {"foo": {"tokens": {"web-app-foo": ""}}}
        return repr(cache_users._hydrate_users_tokens(users, store))

    def test_the_tokens_file_signs_like_the_store_fact_with_the_same_tokens(self):
        store = {"foo": {"tokens": {"web-app-foo": "minted"}}}
        with tempfile.TemporaryDirectory() as tmp:
            tokens_file = Path(tmp) / "tokens.yml"
            tokens_file.write_text(dump_yaml_str({"users": store}), encoding="utf-8")
            self.assertEqual(
                cache_users._store_signature({}, tokens_file),
                cache_users._store_signature(
                    {cache_users_store.STORE_FACT: store}, tokens_file
                ),
            )

    def test_a_missing_tokens_file_signs_like_an_empty_store_fact(self):
        with tempfile.TemporaryDirectory() as tmp:
            absent = Path(tmp) / "tokens.yml"
            self.assertEqual(
                cache_users._store_signature({}, absent),
                cache_users._store_signature(
                    {cache_users_store.STORE_FACT: {}}, absent
                ),
            )

    def test_fields_and_padding_the_hydration_ignores_keep_the_signature(self):
        plain = {"foo": {"tokens": {"web-app-foo": "minted"}}}
        for variant in (
            {"foo": {"tokens": {"web-app-foo": " minted "}}},
            {"foo": {"tokens": {"web-app-foo": "minted"}, "note": "unread"}},
            {"foo": {"tokens": {"web-app-foo": "minted", "web-app-bar": None}}},
            {"foo": {"tokens": {"web-app-foo": "minted"}}, "bar": "not-a-mapping"},
        ):
            with self.subTest(variant=variant):
                self.assertEqual(self._hydrated(variant), self._hydrated(plain))
                self.assertEqual(self._signature(variant), self._signature(plain))

    def test_the_store_order_enters_the_signature_because_the_hydration_keeps_it(
        self,
    ):
        first = {"bar": {"tokens": {"x": "1"}}, "baz": {"tokens": {"y": "2"}}}
        swapped = {"baz": {"tokens": {"y": "2"}}, "bar": {"tokens": {"x": "1"}}}
        self.assertNotEqual(self._hydrated(first), self._hydrated(swapped))
        self.assertNotEqual(self._signature(first), self._signature(swapped))

    def test_an_equal_signature_means_an_equal_hydration(self):
        stores = [
            {"foo": {"tokens": {"web-app-foo": "minted"}}},
            {"foo": {"tokens": {"web-app-foo": " minted "}}},
            {"foo": {"tokens": {"web-app-foo": "rotated"}}},
            {"foo": {"tokens": {"web-app-foo": None}}},
            {"foo": {"tokens": {"web-app-foo": ""}}},
            {"foo": {"tokens": {}}},
            {"foo": {"tokens": "not-a-mapping"}},
            {"foo": {}},
            {"foo": "not-a-mapping"},
            {},
            {"bar": {"tokens": {"x": "1"}}},
            {"bar": {"tokens": {"x": 1}}},
            {"bar": {"tokens": {}}},
            {1: {"tokens": {"x": "1"}}},
            {"1": {"tokens": {"x": "1"}}},
        ]
        for first in stores:
            for second in stores:
                if self._signature(first) != self._signature(second):
                    continue
                with self.subTest(first=first, second=second):
                    self.assertEqual(self._hydrated(first), self._hydrated(second))


if __name__ == "__main__":
    unittest.main()