computer-playbook/tests/unit/python/utils/networks/test_render.py

"""Unit tests for the pure-Python networks renderer that backs the
``compose_networks`` and ``container_networks`` lookup plugins.

The integration test in ``tests/integration/infrastructure/compose/
test_networks_render.py`` pins the full rendered YAML for representative
scenarios; these unit tests cover the helper predicates and edge cases
(empty registries, missing modes, alias defaults, canonical aliases) so
regressions surface with a precise failure rather than a snapshot diff.
"""

from __future__ import annotations

import unittest
from typing import ClassVar

from utils.networks.render import (
    _coerce_bool,
    _compute_attachments,
    _is_consumer,
    _own_shared_net_provider,
    _suppress_default,
    compute_external_network_roles,
    render_compose_networks,
    render_container_networks,
    shared_network_compose_key,
)


def _entity_name(role: str) -> str:
    return role.rsplit("-", 1)[-1]


def _const_lookup_config(**values):
    def _lookup(_app, path, default):
        return values.get(path, default)

    return _lookup


def _role_lookup_config(per_role, **values):
    """Answer per (role, path) so a provider override is expressible.

    Args:
        per_role: ``{role: {path: value}}`` consulted before the shared map.
        values: paths every role answers identically.
    """

    def _lookup(app, path, default):
        scoped = per_role.get(app) or {}
        if path in scoped:
            return scoped[path]
        return values.get(path, default)

    return _lookup


def _const_lookup_database(**values):
    def _lookup(_app, key):
        return values.get(key, "")

    return _lookup


class TestCoerceBool(unittest.TestCase):
    def test_bool_passthrough(self):
        self.assertTrue(_coerce_bool(True))
        self.assertFalse(_coerce_bool(False))

    def test_string_truthy_set(self):
        for v in ("true", "True", "TRUE", "1", "yes", "Yes"):
            self.assertTrue(_coerce_bool(v), msg=v)

    def test_string_falsy_set(self):
        for v in ("false", "no", "0", "", "anything-else"):
            self.assertFalse(_coerce_bool(v), msg=v)

    def test_none_is_false(self):
        self.assertFalse(_coerce_bool(None))

    def test_other_truthy(self):
        self.assertTrue(_coerce_bool(1))
        self.assertTrue(_coerce_bool([0]))
        self.assertFalse(_coerce_bool(0))
        self.assertFalse(_coerce_bool([]))


class TestMcpClientConsumer(unittest.TestCase):
    """Only a client the provider admitted joins its network."""

    ADMITTED: ClassVar[dict] = {"services.qdrant.mcp_consumer": True}

    def _consumer(self, **flags):
        entry = {
            "role": "web-app-homeassistant",
            "overlay": {"consumer": {"kind": "mcp_client"}},
        }
        return _is_consumer(
            entry,
            "svc-db-qdrant",
            _const_lookup_config(**flags),
            _const_lookup_database(),
        )

    def test_client_joins(self):
        self.assertTrue(
            self._consumer(
                **{
                    "mcp.enabled": True,
                    "mcp.shared": True,
                    "mcp.direction": "client",
                    **self.ADMITTED,
                }
            )
        )

    def test_both_joins(self):
        self.assertTrue(
            self._consumer(
                **{
                    "mcp.enabled": True,
                    "mcp.shared": True,
                    "mcp.direction": "both",
                    **self.ADMITTED,
                }
            )
        )

    def test_server_stays_out(self):
        self.assertFalse(
            self._consumer(
                **{
                    "mcp.enabled": True,
                    "mcp.shared": True,
                    "mcp.direction": "server",
                }
            ),
            "a provider on another provider's network is the mesh this kind removes",
        )

    def test_disabled_stays_out(self):
        self.assertFalse(self._consumer(**{"mcp.direction": "client"}))

    def test_a_client_that_never_declared_itself_stays_out(self):
        self.assertFalse(
            self._consumer(
                **{
                    "mcp.enabled": True,
                    "mcp.shared": True,
                    "mcp.direction": "client",
                }
            ),
            "being a client is not an admission; the self-declaration decides",
        )

    def test_a_provider_override_refuses_a_declared_client(self):
        entry = {
            "role": "web-app-homeassistant",
            "overlay": {"consumer": {"kind": "mcp_client"}},
        }
        self.assertFalse(
            _is_consumer(
                entry,
                "svc-db-qdrant",
                _role_lookup_config(
                    {"web-app-homeassistant": {"services.qdrant.mcp_consumer": False}},
                    **{
                        "mcp.enabled": True,
                        "mcp.shared": True,
                        "mcp.direction": "client",
                        "services.qdrant.mcp_consumer": True,
                    },
                ),
                _const_lookup_database(),
            )
        )


class TestConsumerKindList(unittest.TestCase):
    """A provider serving two kinds of consumer declares both on one overlay.

    Qdrant is the case: applications consume it as a vector database through the
    services flags, and MCP clients must reach its adapter. One overlay per role
    means a single `kind` cannot express both, and the second consumer silently
    has no route.
    """

    ENTRY: ClassVar[dict] = {
        "role": "svc-db-qdrant",
        "entity_name": "qdrant",
        "overlay": {
            "consumer": {"kind": ["services_flags", "mcp_client"], "key": "qdrant"}
        },
    }

    def _consumer(self, application_id, **flags):
        return _is_consumer(
            self.ENTRY,
            application_id,
            _const_lookup_config(**flags),
            _const_lookup_database(),
        )

    def test_a_service_consumer_still_joins(self):
        self.assertTrue(
            self._consumer(
                "web-app-openwebui",
                **{"services.qdrant.enabled": True, "services.qdrant.shared": True},
            )
        )

    def test_an_admitted_mcp_client_joins(self):
        self.assertTrue(
            self._consumer(
                "web-app-hermes",
                **{
                    "mcp.enabled": True,
                    "mcp.shared": True,
                    "mcp.direction": "client",
                    "services.hermes.mcp_consumer": True,
                },
            )
        )

    def test_a_role_matching_neither_kind_stays_out(self):
        self.assertFalse(self._consumer("web-app-unrelated"))


class TestSuppressDefault(unittest.TestCase):
    def test_svc_db_prefix_suppresses(self):
        self.assertTrue(_suppress_default("svc-db-mariadb"))
        self.assertTrue(_suppress_default("svc-db-postgres"))

    def test_svc_ai_prefix_suppresses(self):
        self.assertTrue(_suppress_default("svc-ai-ollama"))

    def test_other_prefixes_do_not_suppress(self):
        self.assertFalse(_suppress_default("svc-prx-openresty"))
        self.assertFalse(_suppress_default("web-app-baserow"))
        self.assertFalse(_suppress_default("web-svc-collabora"))


class TestIsConsumer(unittest.TestCase):
    def test_database_kind_all_conditions(self):
        entry = {
            "role": "svc-db-postgres",
            "overlay": {"consumer": {"kind": "database"}},
        }
        lookup_db = _const_lookup_database(
            enabled=True, shared=True, id="svc-db-postgres"
        )
        self.assertTrue(
            _is_consumer(entry, "web-app-baserow", _const_lookup_config(), lookup_db)
        )

    def test_database_kind_wrong_provider(self):
        entry = {
            "role": "svc-db-mariadb",
            "overlay": {"consumer": {"kind": "database"}},
        }
        lookup_db = _const_lookup_database(
            enabled=True, shared=True, id="svc-db-postgres"
        )
        self.assertFalse(
            _is_consumer(entry, "web-app-baserow", _const_lookup_config(), lookup_db)
        )

    def test_database_kind_not_shared(self):
        entry = {
            "role": "svc-db-mariadb",
            "overlay": {"consumer": {"kind": "database"}},
        }
        lookup_db = _const_lookup_database(
            enabled=True, shared=False, id="svc-db-mariadb"
        )
        self.assertFalse(
            _is_consumer(entry, "web-app-baserow", _const_lookup_config(), lookup_db)
        )

    def test_services_flags_explicit_key_and_flags(self):
        entry = {
            "role": "svc-prx-openresty",
            "entity_name": "openresty",
            "overlay": {
                "consumer": {
                    "kind": "services_flags",
                    "key": "sso",
                    "flags": ["enabled"],
                }
            },
        }
        cfg = _const_lookup_config(**{"services.sso.enabled": True})
        self.assertTrue(
            _is_consumer(entry, "web-app-baserow", cfg, _const_lookup_database())
        )

    def test_services_flags_default_key_from_provides(self):
        entry = {
            "role": "svc-db-openldap",
            "entity_name": "openldap",
            "provides": "ldap",
            "overlay": {},
        }
        cfg = _const_lookup_config(
            **{"services.ldap.enabled": True, "services.ldap.shared": True}
        )
        self.assertTrue(
            _is_consumer(entry, "web-app-bookwyrm", cfg, _const_lookup_database())
        )

    def test_services_flags_default_key_from_entity_name(self):
        entry = {
            "role": "svc-ai-ollama",
            "entity_name": "ollama",
            "overlay": {},
        }
        cfg = _const_lookup_config(
            **{"services.ollama.enabled": True, "services.ollama.shared": True}
        )
        self.assertTrue(
            _is_consumer(entry, "web-app-openwebui", cfg, _const_lookup_database())
        )

    def test_services_flags_default_flags_require_enabled_and_shared(self):
        entry = {
            "role": "svc-db-openldap",
            "entity_name": "openldap",
            "provides": "ldap",
            "overlay": {},
        }
        cfg = _const_lookup_config(**{"services.ldap.enabled": True})
        self.assertFalse(
            _is_consumer(entry, "web-app-x", cfg, _const_lookup_database())
        )

    def test_unknown_kind_returns_false(self):
        entry = {"role": "x", "overlay": {"consumer": {"kind": "future_thing"}}}
        self.assertFalse(
            _is_consumer(
                entry, "web-app-x", _const_lookup_config(), _const_lookup_database()
            )
        )

    def test_web_facing_matches_web_app_and_web_svc(self):
        entry = {
            "role": "svc-prx-openresty",
            "overlay": {"consumer": {"kind": "web_facing"}},
        }
        for app in ("web-app-baserow", "web-svc-logout"):
            self.assertTrue(
                _is_consumer(
                    entry, app, _const_lookup_config(), _const_lookup_database()
                ),
                msg=app,
            )

    def test_web_facing_rejects_non_web_roles(self):
        entry = {
            "role": "svc-prx-openresty",
            "overlay": {"consumer": {"kind": "web_facing"}},
        }
        for app in ("svc-db-mariadb", "svc-ai-ollama", "sys-svc-x"):
            self.assertFalse(
                _is_consumer(
                    entry, app, _const_lookup_config(), _const_lookup_database()
                ),
                msg=app,
            )


class TestIsConsumerOnionSso(unittest.TestCase):
    ENTRY: ClassVar[dict] = {
        "role": "svc-net-tor",
        "overlay": {
            "consumer": {
                "kind": "onion_sso",
                "key": "sso",
                "provider": "web-app-keycloak",
                "provider_key": "tor",
            }
        },
    }

    @staticmethod
    def _lookup(*, sso, provider_tor):
        def _inner(app, path, default):
            if app == "web-app-keycloak" and path == "services.tor.enabled":
                return provider_tor
            if path == "services.sso.enabled":
                return sso
            return default

        return _inner

    def _consumer(self, *, sso, provider_tor, entry=None):
        return _is_consumer(
            entry or self.ENTRY,
            "web-app-nextcloud",
            self._lookup(sso=sso, provider_tor=provider_tor),
            _const_lookup_database(),
        )

    def test_attaches_only_when_sso_consumer_meets_onion_provider(self):
        cases = [
            (True, True, True),
            (True, False, False),
            (False, True, False),
            (False, False, False),
        ]
        for sso, provider_tor, expected in cases:
            with self.subTest(sso=sso, provider_tor=provider_tor):
                self.assertIs(
                    self._consumer(sso=sso, provider_tor=provider_tor), expected
                )

    def test_provider_is_mandatory(self):
        entry = {"role": "svc-net-tor", "overlay": {"consumer": {"kind": "onion_sso"}}}
        self.assertFalse(self._consumer(sso=True, provider_tor=True, entry=entry))

    def test_provider_key_defaults_to_tor(self):
        entry = {
            "role": "svc-net-tor",
            "overlay": {
                "consumer": {"kind": "onion_sso", "provider": "web-app-keycloak"}
            },
        }
        self.assertTrue(self._consumer(sso=True, provider_tor=True, entry=entry))
        self.assertFalse(self._consumer(sso=True, provider_tor=False, entry=entry))


class TestComputeAttachments(unittest.TestCase):
    def test_skips_canonical_aliases(self):
        registry = {
            "css": {
                "role": "web-svc-cdn",
                "entity_name": "cdn",
                "canonical": "cdn",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "shared_net",
                },
            },
        }
        att, default_aliases = _compute_attachments(
            registry,
            "web-app-x",
            "swarm",
            _const_lookup_config(),
            _const_lookup_database(),
        )
        self.assertEqual(att, [])
        self.assertEqual(default_aliases, [])

    def test_skips_mode_mismatch(self):
        registry = {
            "ldap": {
                "role": "svc-db-openldap",
                "entity_name": "openldap",
                "provides": "ldap",
                "overlay": {
                    "modes": ["compose"],
                    "topology": "shared_net",
                },
            },
        }
        att, _ = _compute_attachments(
            registry,
            "svc-db-openldap",
            "swarm",
            _const_lookup_config(),
            _const_lookup_database(),
        )
        self.assertEqual(att, [])

    def test_skips_beacon_entries_for_non_provider(self):
        registry = {
            "sso": {
                "role": "web-app-keycloak",
                "entity_name": "keycloak",
                "provides": "sso",
                "overlay": {
                    "modes": ["swarm"],
                    "proxy_resolvable": True,
                    "aliases": ["auth.example.com"],
                },
            },
        }
        att, default_aliases = _compute_attachments(
            registry,
            "web-app-baserow",
            "swarm",
            _const_lookup_config(**{"services.sso.enabled": True}),
            _const_lookup_database(),
        )
        self.assertEqual(att, [])
        self.assertEqual(default_aliases, [])

    def test_provider_self_render_uses_default_alias_from_entity_name(self):
        registry = {
            "ldap": {
                "role": "svc-db-openldap",
                "entity_name": "openldap",
                "provides": "ldap",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "shared_net",
                },
            },
        }
        att, _ = _compute_attachments(
            registry,
            "svc-db-openldap",
            "swarm",
            _const_lookup_config(),
            _const_lookup_database(),
        )
        self.assertEqual(len(att), 1)
        self.assertEqual(att[0]["aliases"], ["openldap"])

    def test_proxy_aliases_override_topology_aliases_in_harvest(self):
        """A shared_net beacon exposes only proxy_aliases to the harvest; its
        own topology aliases (the entity name) must stay off the proxy."""
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "shared_net",
                    "collect_proxy_resolvable": True,
                },
            },
            "seaweedfs": {
                "role": "web-app-seaweedfs",
                "entity_name": "seaweedfs",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "shared_net",
                    "proxy_resolvable": True,
                    "proxy_aliases": ["api.s3.example.com"],
                    "aliases": ["seaweedfs"],
                },
            },
        }
        att, _ = _compute_attachments(
            registry,
            "svc-prx-openresty",
            "swarm",
            _const_lookup_config(),
            _const_lookup_database(),
        )
        provider = next(a for a in att if a["is_provider"])
        self.assertIn("api.s3.example.com", provider["aliases"])
        self.assertNotIn("seaweedfs", provider["aliases"])

    def test_proxy_resolvable_sweep_skips_canonical_clones(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {"modes": ["swarm"], "topology": "default_net"},
            },
            "real_beacon": {
                "role": "web-app-keycloak",
                "entity_name": "keycloak",
                "provides": "sso",
                "overlay": {
                    "modes": ["swarm"],
                    "proxy_resolvable": True,
                    "aliases": ["auth.example.com"],
                },
            },
            "clone_alias": {
                "role": "web-app-keycloak",
                "entity_name": "keycloak",
                "canonical": "sso",
                "overlay": {
                    "modes": ["swarm"],
                    "proxy_resolvable": True,
                    "aliases": ["auth.example.com"],
                },
            },
        }
        _, default_aliases = _compute_attachments(
            registry,
            "svc-prx-openresty",
            "swarm",
            _const_lookup_config(),
            _const_lookup_database(),
        )
        self.assertEqual(default_aliases, ["auth.example.com"])

    def test_proxy_resolvable_sweep_skips_self_role(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "default_net",
                    "proxy_resolvable": True,
                    "aliases": ["self-mistake"],
                },
            },
            "sso": {
                "role": "web-app-keycloak",
                "entity_name": "keycloak",
                "provides": "sso",
                "overlay": {
                    "modes": ["swarm"],
                    "proxy_resolvable": True,
                    "aliases": ["auth.example.com"],
                },
            },
        }
        _, default_aliases = _compute_attachments(
            registry,
            "svc-prx-openresty",
            "swarm",
            _const_lookup_config(),
            _const_lookup_database(),
        )
        self.assertEqual(default_aliases.count("self-mistake"), 1)
        self.assertIn("auth.example.com", default_aliases)


class TestRenderComposeNetworks(unittest.TestCase):
    def test_provider_swarm_emits_only_default(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {"modes": ["swarm"], "topology": "default_net"},
            },
        }
        rendered = render_compose_networks(
            application_id="svc-prx-openresty",
            deployment_mode="swarm",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("networks:", rendered)
        self.assertIn("default:", rendered)
        self.assertIn("driver: overlay", rendered)
        self.assertNotIn("openresty:\n    external", rendered)

    def test_provider_compose_falls_back_to_bridge(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {"modes": ["swarm"], "topology": "default_net"},
            },
        }
        rendered = render_compose_networks(
            application_id="svc-prx-openresty",
            deployment_mode="compose",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(
                **{"networks.local.subnet": "10.0.0.0/24"}
            ),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("driver: bridge", rendered)
        self.assertIn("subnet: 10.0.0.0/24", rendered)

    def test_svc_db_provider_suppresses_default_block(self):
        registry = {
            "mariadb": {
                "role": "svc-db-mariadb",
                "entity_name": "mariadb",
                "overlay": {"modes": ["swarm"], "topology": "shared_net"},
            },
        }
        rendered = render_compose_networks(
            application_id="svc-db-mariadb",
            deployment_mode="swarm",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("mariadb:", rendered)
        self.assertIn("external: true", rendered)
        self.assertNotIn("default:", rendered)

    def test_empty_own_entity_swarm_omits_name(self):
        rendered = render_compose_networks(
            application_id="svc-runner",
            deployment_mode="swarm",
            registry={},
            get_entity_name=lambda _role: "",
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("driver: overlay", rendered)
        self.assertNotIn("name:", rendered)

    def test_node_local_swarm_container_networks_compose_shape(self):
        rendered = render_container_networks(
            application_id="svc-runner",
            deployment_mode="swarm",
            registry={},
            get_entity_name=lambda _role: "",
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
            node_local=True,
        )
        expected = render_container_networks(
            application_id="svc-runner",
            deployment_mode="compose",
            registry={},
            get_entity_name=lambda _role: "",
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertEqual(rendered, expected)

    def test_a_role_with_nothing_to_attach_emits_no_networks_key(self):
        """A bare `networks:` parses as null, and compose refuses it as not an array."""
        for render in (render_container_networks, render_compose_networks):
            for role in ("svc-ai-mcp-adapter", "svc-db-nothing"):
                with self.subTest(render=render.__name__, role=role):
                    self.assertEqual(
                        "",
                        render(
                            application_id=role,
                            deployment_mode="compose",
                            registry={},
                            get_entity_name=_entity_name,
                            lookup_config=_const_lookup_config(),
                            lookup_database=_const_lookup_database(),
                        ),
                    )

    def test_a_role_that_keeps_the_default_still_renders_it(self):
        """The guard must not swallow the default attachment it sits behind."""
        for render in (render_container_networks, render_compose_networks):
            with self.subTest(render=render.__name__):
                self.assertIn(
                    "default:",
                    render(
                        application_id="web-app-anything",
                        deployment_mode="compose",
                        registry={},
                        get_entity_name=_entity_name,
                        lookup_config=_const_lookup_config(),
                        lookup_database=_const_lookup_database(),
                    ),
                )

    def test_own_network_only_drops_the_consumer_networks(self):
        registry = {
            "qdrant": {
                "role": "svc-db-qdrant",
                "entity_name": "qdrant",
                "overlay": {"modes": ["compose"], "topology": "shared_net"},
            },
            "moodle": {
                "role": "web-app-moodle",
                "entity_name": "moodle",
                "overlay": {"modes": ["compose"], "topology": "shared_net"},
            },
        }
        lookup_config = _const_lookup_config(
            **{"services.moodle.enabled": True, "services.moodle.shared": True}
        )
        full = render_container_networks(
            application_id="svc-db-qdrant",
            deployment_mode="compose",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=lookup_config,
            lookup_database=_const_lookup_database(),
        )
        restricted = render_container_networks(
            application_id="svc-db-qdrant",
            deployment_mode="compose",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=lookup_config,
            lookup_database=_const_lookup_database(),
            own_network_only=True,
            provider_self_alias=False,
        )
        self.assertIn("moodle:", full)
        self.assertIn("qdrant:", restricted)
        self.assertNotIn("moodle:", restricted)
        self.assertNotIn("aliases:", restricted)

    def test_own_network_only_emits_nothing_without_an_attachment(self):
        rendered = render_container_networks(
            application_id="web-app-prometheus",
            deployment_mode="compose",
            registry={},
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
            own_network_only=True,
        )
        self.assertEqual(
            rendered,
            "",
            "a bare 'networks:' key is not a list and compose refuses the file",
        )

    def test_node_local_swarm_renders_bridge(self):
        rendered = render_compose_networks(
            application_id="svc-runner",
            deployment_mode="swarm",
            registry={},
            get_entity_name=lambda _role: "",
            lookup_config=_const_lookup_config(
                **{"networks.local.subnet": "10.0.0.0/24"}
            ),
            lookup_database=_const_lookup_database(),
            node_local=True,
        )
        self.assertNotIn("overlay", rendered)
        self.assertIn("driver: bridge", rendered)

    def test_empty_own_entity_compose_omits_name(self):
        rendered = render_compose_networks(
            application_id="svc-runner",
            deployment_mode="compose",
            registry={},
            get_entity_name=lambda _role: "",
            lookup_config=_const_lookup_config(
                **{"networks.local.subnet": "10.0.0.0/24"}
            ),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("driver: bridge", rendered)
        self.assertIn("subnet: 10.0.0.0/24", rendered)
        self.assertNotIn("name:", rendered)


class TestComputeExternalNetworkRoles(unittest.TestCase):
    def test_returns_consumer_provider_role(self):
        registry = {
            "redis": {
                "role": "svc-db-redis",
                "entity_name": "redis",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "shared_net",
                    "consumer": {"kind": "services_flags", "key": "redis"},
                },
            },
        }
        cfg = _const_lookup_config(
            **{"services.redis.enabled": True, "services.redis.shared": True}
        )
        roles = compute_external_network_roles(
            application_id="web-app-gitea",
            deployment_mode="swarm",
            registry=registry,
            lookup_config=cfg,
            lookup_database=_const_lookup_database(),
        )
        self.assertEqual(roles, ["svc-db-redis"])

    def test_matches_external_true_entries_in_rendered_compose(self):
        registry = {
            "redis": {
                "role": "svc-db-redis",
                "entity_name": "redis",
                "overlay": {
                    "modes": ["swarm"],
                    "topology": "shared_net",
                    "consumer": {"kind": "services_flags", "key": "redis"},
                },
            },
        }
        cfg = _const_lookup_config(
            **{"services.redis.enabled": True, "services.redis.shared": True}
        )
        rendered = render_compose_networks(
            application_id="web-app-gitea",
            deployment_mode="swarm",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=cfg,
            lookup_database=_const_lookup_database(),
        )
        roles = compute_external_network_roles(
            application_id="web-app-gitea",
            deployment_mode="swarm",
            registry=registry,
            lookup_config=cfg,
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("redis:\n    external: true", rendered)
        self.assertEqual([_entity_name(r) for r in roles], ["redis"])

    def test_own_default_net_provider_not_returned(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {"modes": ["swarm"], "topology": "default_net"},
            },
        }
        roles = compute_external_network_roles(
            application_id="svc-prx-openresty",
            deployment_mode="swarm",
            registry=registry,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertEqual(roles, [])


class TestRenderContainerNetworks(unittest.TestCase):
    def test_output_starts_with_leading_newline(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {"modes": ["swarm"], "topology": "default_net"},
            },
        }
        rendered = render_container_networks(
            application_id="svc-prx-openresty",
            deployment_mode="swarm",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertTrue(rendered.startswith("\n"))
        self.assertIn("networks:", rendered)

    def test_consumer_attach_uses_dict_empty(self):
        registry = {
            "ldap": {
                "role": "svc-db-openldap",
                "entity_name": "openldap",
                "provides": "ldap",
                "overlay": {"modes": ["swarm"], "topology": "shared_net"},
            },
        }
        rendered = render_container_networks(
            application_id="web-app-bookwyrm",
            deployment_mode="swarm",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(
                **{"services.ldap.enabled": True, "services.ldap.shared": True}
            ),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("openldap:\n    {}", rendered)

    def test_provider_default_net_skips_self_attach_emits_alias_block(self):
        registry = {
            "openresty": {
                "role": "svc-prx-openresty",
                "entity_name": "openresty",
                "overlay": {"modes": ["swarm"], "topology": "default_net"},
            },
            "sso": {
                "role": "web-app-keycloak",
                "entity_name": "keycloak",
                "provides": "sso",
                "overlay": {
                    "modes": ["swarm"],
                    "proxy_resolvable": True,
                    "aliases": ["auth.example.com"],
                },
            },
        }
        rendered = render_container_networks(
            application_id="svc-prx-openresty",
            deployment_mode="swarm",
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertNotIn("openresty:\n", rendered)
        self.assertIn("default:", rendered)
        self.assertIn("- auth.example.com", rendered)

    def test_empty_registry_emits_only_default(self):
        rendered = render_container_networks(
            application_id="web-app-plain",
            deployment_mode="swarm",
            registry={},
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertEqual(rendered, "\nnetworks:\n  default:")


class TestRenderEncryption(unittest.TestCase):
    def test_swarm_encrypted_false_emits_lowercase_quoted_literal(self):
        rendered = render_compose_networks(
            application_id="web-app-x",
            deployment_mode="swarm",
            registry={},
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
            swarm_encrypted=False,
        )
        self.assertIn('encrypted: "false"', rendered)
        self.assertNotIn("encrypted: false\n", rendered)
        self.assertNotIn('encrypted: "False"', rendered)


_PROVIDER_REGISTRY = {
    "seaweedfs": {
        "role": "web-app-seaweedfs",
        "entity_name": "seaweedfs",
        "overlay": {"modes": ["compose", "swarm"], "topology": "shared_net"},
    },
}


class TestOwnSharedNetProvider(unittest.TestCase):
    """A multi-service objstore provider (seaweedfs/minio) renders its own
    ``<entity>`` net as external + a bare auto-IPAM project default (so the
    role subnet, owned by the external net, is never re-requested), and its
    sidecar services attach without the provider alias. Locks the
    objstore-net-fix contract so a render edit cannot silently regress it."""

    def test_helper_true_only_for_own_shared_net_provider(self):
        own = [
            {"role": "web-app-seaweedfs", "topology": "shared_net", "is_provider": True}
        ]
        consumer = [
            {
                "role": "web-app-seaweedfs",
                "topology": "shared_net",
                "is_provider": False,
            }
        ]
        self.assertTrue(_own_shared_net_provider(own, "seaweedfs", _entity_name))
        self.assertFalse(_own_shared_net_provider(consumer, "baserow", _entity_name))
        self.assertFalse(_own_shared_net_provider([], "seaweedfs", _entity_name))

    def test_compose_default_drops_name_and_subnet(self):
        rendered = render_compose_networks(
            application_id="web-app-seaweedfs",
            deployment_mode="compose",
            registry=_PROVIDER_REGISTRY,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(
                **{"networks.local.subnet": "192.168.206.0/24"}
            ),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("  seaweedfs:\n    external: true", rendered)
        self.assertIn("  default:\n    driver: bridge", rendered)
        self.assertNotIn("192.168.206", rendered)
        self.assertNotIn("name: seaweedfs", rendered)

    def test_swarm_default_is_nameless_overlay(self):
        rendered = render_compose_networks(
            application_id="web-app-seaweedfs",
            deployment_mode="swarm",
            registry=_PROVIDER_REGISTRY,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )
        self.assertIn("  seaweedfs:\n    external: true", rendered)
        self.assertIn("  default:\n    driver: overlay", rendered)
        self.assertNotIn("name: seaweedfs", rendered)

    def test_main_publishes_alias_sidecar_does_not(self):
        kwargs = {
            "application_id": "web-app-seaweedfs",
            "deployment_mode": "compose",
            "registry": _PROVIDER_REGISTRY,
            "get_entity_name": _entity_name,
            "lookup_config": _const_lookup_config(),
            "lookup_database": _const_lookup_database(),
        }
        main = render_container_networks(**kwargs)
        sidecar = render_container_networks(provider_self_alias=False, **kwargs)
        self.assertIn("seaweedfs:\n    aliases:\n      - seaweedfs", main)
        self.assertIn("seaweedfs:\n    {}", sidecar)
        self.assertNotIn("- seaweedfs", sidecar)


class TestSharedNetworkComposeKey(unittest.TestCase):
    """The key under which the pre-created ``<entity>`` network appears in the
    rendered file. Callers stamp it into ``com.docker.compose.network``; docker
    compose resolves a project's networks against that label, so a key that does
    not match the rendered file makes compose adopt the wrong network and skip
    creating the right one."""

    def _key(self, application_id, registry, deployment_mode="compose"):
        return shared_network_compose_key(
            application_id=application_id,
            deployment_mode=deployment_mode,
            registry=registry,
            get_entity_name=_entity_name,
            lookup_config=_const_lookup_config(),
            lookup_database=_const_lookup_database(),
        )

    def test_own_shared_net_provider_keys_on_its_entity(self):
        for mode in ("compose", "swarm"):
            with self.subTest(mode=mode):
                self.assertEqual(
                    self._key("web-app-seaweedfs", _PROVIDER_REGISTRY, mode),
                    "seaweedfs",
                )

    def test_plain_app_keys_on_default(self):
        for mode in ("compose", "swarm"):
            with self.subTest(mode=mode):
                self.assertEqual(self._key("web-app-baserow", {}, mode), "default")

    def test_consumer_of_a_foreign_provider_keys_on_default(self):
        registry = {
            "ldap": {
                "role": "svc-db-openldap",
                "entity_name": "openldap",
                "provides": "ldap",
                "overlay": {"modes": ["compose"], "topology": "shared_net"},
            },
        }
        self.assertEqual(self._key("web-app-baserow", registry), "default")

    def test_node_local_forces_the_compose_answer(self):
        self.assertEqual(
            shared_network_compose_key(
                application_id="web-app-seaweedfs",
                deployment_mode="swarm",
                registry=_PROVIDER_REGISTRY,
                get_entity_name=_entity_name,
                lookup_config=_const_lookup_config(),
                lookup_database=_const_lookup_database(),
                node_local=True,
            ),
            "seaweedfs",
        )

    def test_key_matches_the_key_the_renderer_emits(self):
        """The contract that makes this a single source: whatever key the
        function reports must be the key the rendered networks block actually
        uses for the entity network."""
        cases = (
            ("web-app-seaweedfs", _PROVIDER_REGISTRY),
            ("web-app-baserow", {}),
        )
        for application_id, registry in cases:
            for mode in ("compose", "swarm"):
                with self.subTest(application_id=application_id, mode=mode):
                    key = self._key(application_id, registry, mode)
                    rendered = render_compose_networks(
                        application_id=application_id,
                        deployment_mode=mode,
                        registry=registry,
                        get_entity_name=_entity_name,
                        lookup_config=_const_lookup_config(
                            **{"networks.local.subnet": "192.168.206.0/24"}
                        ),
                        lookup_database=_const_lookup_database(),
                    )
                    self.assertIn(f"  {key}:", rendered)
                    entity = _entity_name(application_id)
                    if key == "default":
                        self.assertIn(f"    name: {entity}", rendered)
                    else:
                        self.assertEqual(key, entity)
                        self.assertNotIn(f"    name: {entity}", rendered)


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