kevinveenbirkenbach/computer-playbook
1
0 Fork 0
mirror of https://github.com/kevinveenbirkenbach/computer-playbook.git synced 2026-11-08 19:56:53 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
e16d807e6286df2d34fdd323cc7a2c836cef12f2
computer-playbook/tests/unit/python/plugins/lookup/test_database.py
T
kevinveenbirkenbachandClaude Opus 5 e2948332fd fix(merge): repair what the main merge left broken 
The merge brought in main's new roles and carried this branch's cross-cutting
renames and lints past them, so files that never conflicted were left
inconsistent. make quality-high is green again: 4 test targets, 15 lint targets.

a05071fe78 renamed get_entity_name to entity_name. Three files main added still
used the old name and merged cleanly because this branch never touched them:
plugins/lookup/email.py raised ImportError on import, which also took down
sys-ctl-hlth-msmtp's mail_recipient lookup; web-app-stalwart/tasks/00_core.yml
applied `| get_entity_name`, a filter that does not exist (the registration in
plugins/filter/get/entity_name.py is `entity_name`), so a compose deploy of
stalwart would have failed while rendering sys-stk-front-proxy's vars; and
test_database.py patched a name the module no longer binds.

web-app-stalwart and web-app-mailu spell DEPLOYMENT_MODE out in 15 places. The
named booleans in group_vars/all/18_swarm.yml are the SPOT and the bare form is
the idiom: 177 existing uses, and web-app-listmonk/tasks/utils/mcp.yml:42,49
drives both as mutually exclusive `when:` conditions, which a truthy string
would break.

svc-net-tor-smtp pinned aiosmtpd and PySocks inside its Dockerfile, where
Dependabot cannot see them. The pins move to files/requirements.txt; since
sys-svc-compose stages only the Dockerfile, 00_core.yml now stages the
requirements file into the build context too.

stalwart's webui_version had no upstream declared. Its three inline
`environment:` blocks stay, with a nocheck each: the role sets no
docker_compose_env, so one templates/env.j2 would also be the env_file of the
sibling containers, and clamav would receive the recovery admin credential and
the OIDC client secret.

The 13 bindings in stalwart's playwright env.js defaulted to "" and were
consumed unconditionally. Every key is emitted unconditionally by
playwright.env.j2 from a lookup that fails rather than yields empty, and
OIDC_ISSUER_URL always concatenates the realm path, so requireDotenvValue is
the correct binding. gitlab's adminUsername gets the toBeTruthy guard its two
siblings already had.

web-svc-api and svc-cache-package collided with stalwart on local.http 8094 and
8095 and move to 8098 and 8099, which exhausts the 8001-8099 band.
svc-ai-agent-broker collided with stalwart on 192.168.223.0/24 and moves to
192.168.81.0/24; the suggester only subdivides blocks already in use at the
requested prefix, so the move was validated with --block.

web-app-weblate calls lookup('email') without emitting
lookup('container_extra_hosts'), so on an onion deploy its SMTP client drops
every message without an error.

test_main.py read the primary domain from a literal while main moved
INFINITO_DOMAIN to main.infinito.test. It now takes the value from
default_domain_primary(), in the shape e2beab9f1c uses on
feature/web-app-wazuh/add-role, so that branch merges here without a conflict.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-10-04 03:21:09 +02:00

562 lines
19 KiB
Python
Raw Blame History

import importlib.util
import unittest
import unittest.mock as mock
from unittest.mock import patch
from ansible.errors import AnsibleError
from . import PROJECT_ROOT
def _load_module(rel_path: str, name: str):
path = PROJECT_ROOT / rel_path
spec = importlib.util.spec_from_file_location(name, str(path))
mod = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(mod)
return mod
class _DummyTemplar:
def __init__(self, available_variables):
self.available_variables = available_variables
class DatabaseLookupTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.db_lookup_mod = _load_module(
"plugins/lookup/database.py", "lookup_database"
)
def _make_lookup(self, available_vars: dict):
lm = self.db_lookup_mod.LookupModule()
lm._templar = _DummyTemplar(available_vars)
lm._loader = mock.MagicMock()
patcher = mock.patch.object(self.db_lookup_mod, "lookup_loader")
loader_mock = patcher.start()
self.addCleanup(patcher.stop)
loader_mock.get.return_value = mock.MagicMock(
run=lambda _terms, variables=None, **_k: [
(variables or {}).get("applications", {})
]
)
return lm
@staticmethod
def _fake_entity_name(role_name: str) -> str:
"""
Make entity resolution deterministic for unit tests (no filesystem access).
Mirrors the typical behavior for your role naming.
"""
role_name = role_name.strip()
for prefix in ("web-app-", "web-svc-", "svc-db-", "svc-", "persona-"):
if role_name.startswith(prefix):
return role_name[len(prefix) :]
return role_name
def test_invalid_terms_raises(self):
vars_ = {"applications": {}, "ports": {}, "DIR_COMPOSITIONS": "/opt/compose/"}
lookup = self._make_lookup(vars_)
with self.assertRaises(AnsibleError):
lookup.run([], variables=vars_)
with self.assertRaises(AnsibleError):
lookup.run(["a", "b", "c"], variables=vars_)
def test_kwarg_want_is_not_supported_raises(self):
vars_ = {"applications": {}, "ports": {}, "DIR_COMPOSITIONS": "/opt/compose/"}
lookup = self._make_lookup(vars_)
with self.assertRaises(AnsibleError):
lookup.run(["web-app-foo"], variables=vars_, want="url_full")
def test_no_dbtype_configured_returns_empty_like_vars_logic(self):
applications = {
"web-app-foo": {
"services": {},
"secrets": {"credentials": {"database_password": "pw"}},
}
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertFalse(out["enabled"])
self.assertFalse(out["shared"])
self.assertFalse(out["local"])
self.assertEqual(out.get("id", ""), "")
self.assertEqual(out["type"], "")
self.assertEqual(out["name"], "foo")
self.assertEqual(out["username"], "foo")
self.assertEqual(out["host"], "")
self.assertEqual(out["network"], "")
self.assertEqual(out["port"], "")
self.assertEqual(out["env"], "")
self.assertEqual(out["url_jdbc"], "")
self.assertEqual(out["url_full"], "")
self.assertEqual(out["volume"], "")
self.assertEqual(out["image"], "")
self.assertEqual(out["version"], "")
self.assertEqual(out["reach_host"], "127.0.0.1")
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
self.assertEqual(
lookup.run(["web-app-foo", "url_full"], variables=vars_)[0], ""
)
def test_postgres_dedicated_matches_helper_variables_definition(self):
applications = {
"web-app-foo": {
"services": {"postgres": {"enabled": True, "shared": False}},
"secrets": {"credentials": {"database_password": "pw"}},
},
"svc-db-postgres": {
"services": {
"postgres": {
"name": "postgres-central",
"image": "postgis/postgis",
"version": "16",
"ports": {"local": {"postgres": "5432"}},
}
}
},
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertTrue(out["enabled"])
self.assertFalse(out["shared"])
self.assertTrue(out["local"])
self.assertTrue(out["managed"])
self.assertEqual(out["id"], "svc-db-postgres")
self.assertEqual(out["type"], "postgres")
self.assertEqual(out["name"], "foo")
self.assertEqual(out["username"], "foo")
self.assertEqual(out["host"], "database")
self.assertEqual(out["container"], "foo-database")
self.assertEqual(out["network"], "foo")
self.assertEqual(out["password"], "pw")
self.assertEqual(out["port"], "5432")
self.assertEqual(out["env"], "/opt/compose/foo/.env/postgres.env")
self.assertEqual(out["initdb_dir"], "/opt/compose/foo/.initdb.d/")
self.assertEqual(out["build_dir"], "/opt/compose/foo/.postgres-build/")
self.assertEqual(out["url_jdbc"], "jdbc:postgresql://database:5432/foo")
self.assertEqual(out["url_full"], "postgres://foo:pw@database:5432/foo")
self.assertEqual(out["volume"], "foo_database")
self.assertEqual(out["image"], "postgis/postgis")
self.assertEqual(out["version"], "16")
self.assertEqual(out["reach_host"], "127.0.0.1")
self.assertEqual(out["instance"], "foo")
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
self.assertEqual(
lookup.run(["web-app-foo", "url_full"], variables=vars_)[0],
"postgres://foo:pw@database:5432/foo",
)
self.assertEqual(
lookup.run(["web-app-foo", "port"], variables=vars_)[0],
"5432",
)
def test_dedicated_with_name_override_is_not_managed(self):
applications = {
"web-app-foo": {
"services": {
"postgres": {
"enabled": True,
"shared": False,
"name": "foo-postgres-1",
}
},
"credentials": {"database_password": "pw"},
},
"svc-db-postgres": {
"services": {
"postgres": {
"name": "postgres-central",
"image": "postgis/postgis",
"version": "16",
"ports": {"local": {"postgres": "5432"}},
}
}
},
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertTrue(out["enabled"])
self.assertFalse(out["shared"])
self.assertTrue(out["local"])
self.assertFalse(out["managed"])
def test_postgres_shared_uses_central_name_for_host_instance_container_volume(self):
applications = {
"web-app-foo": {
"services": {"postgres": {"enabled": True, "shared": True}},
"secrets": {"credentials": {"database_password": "pw"}},
},
"svc-db-postgres": {
"services": {
"postgres": {
"name": "postgres-central",
"version": "16",
"ports": {"local": {"postgres": "5432"}},
}
}
},
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertTrue(out["enabled"])
self.assertTrue(out["shared"])
self.assertFalse(out["local"])
self.assertFalse(out["managed"])
self.assertEqual(out["id"], "svc-db-postgres")
self.assertEqual(out["host"], "postgres-central")
self.assertEqual(out["instance"], "postgres-central")
self.assertEqual(out["container"], "postgres")
self.assertEqual(out["network"], "postgres")
self.assertEqual(out["volume"], "postgres-central")
self.assertEqual(out["url_jdbc"], "jdbc:postgresql://postgres-central:5432/foo")
self.assertEqual(out["url_full"], "postgres://foo:pw@postgres-central:5432/foo")
def test_mariadb_jdbc_scheme_stays_mariadb(self):
applications = {
"web-app-foo": {
"services": {"mariadb": {"enabled": True, "shared": False}},
"secrets": {"credentials": {"database_password": "pw"}},
},
"svc-db-mariadb": {
"services": {
"mariadb": {
"name": "mariadb-central",
"version": "11.4",
"ports": {"local": {"mariadb": "3306"}},
}
}
},
}
ports = {"localhost": {"database": {"svc-db-mariadb": "3306"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertTrue(out["enabled"])
self.assertFalse(out["shared"])
self.assertTrue(out["local"])
self.assertEqual(out["id"], "svc-db-mariadb")
self.assertEqual(out["type"], "mariadb")
self.assertEqual(out["host"], "database")
self.assertEqual(out["port"], "3306")
self.assertEqual(out["env"], "/opt/compose/foo/.env/mariadb.env")
self.assertEqual(out["url_jdbc"], "jdbc:mariadb://database:3306/foo")
self.assertEqual(out["url_full"], "mariadb://foo:pw@database:3306/foo")
def test_version_override_on_consumer_wins_over_default(self):
applications = {
"web-app-foo": {
"services": {
"postgres": {
"enabled": True,
"shared": False,
"version": "15",
}
},
"secrets": {"credentials": {"database_password": "pw"}},
},
"svc-db-postgres": {
"services": {"postgres": {"name": "postgres-central", "version": "16"}}
},
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertTrue(out["enabled"])
self.assertFalse(out["shared"])
self.assertEqual(out["id"], "svc-db-postgres")
self.assertEqual(out["version"], "15")
def test_image_override_on_consumer_wins_over_default(self):
applications = {
"web-app-foo": {
"services": {
"postgres": {
"enabled": True,
"shared": False,
"image": "postgres",
"version": "17",
}
},
"credentials": {"database_password": "pw"},
},
"svc-db-postgres": {
"services": {
"postgres": {
"name": "postgres-central",
"image": "postgis/postgis",
"version": "16",
}
}
},
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertEqual(out["image"], "postgres")
self.assertEqual(out["version"], "17")
def test_local_flag_for_variant_disabling_dedicated_db(self):
applications = {
"web-app-foo": {
"services": {"postgres": {"enabled": False, "shared": False}},
"secrets": {"credentials": {"database_password": "pw"}},
}
}
ports = {"localhost": {"database": {"svc-db-postgres": "5432"}}}
vars_ = {
"applications": applications,
"ports": ports,
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
out = lookup.run(["web-app-foo"], variables=vars_)[0]
self.assertFalse(out["enabled"])
self.assertFalse(out["shared"])
self.assertFalse(out["local"])
self.assertEqual(out["host"], "")
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
self.assertFalse(lookup.run(["web-app-foo", "local"], variables=vars_)[0])
def test_multiple_database_services_raise(self):
applications = {
"web-app-foo": {
"services": {
"mariadb": {"enabled": True, "shared": False},
"postgres": {"enabled": True, "shared": False},
}
}
}
vars_ = {
"applications": applications,
"ports": {"localhost": {"database": {}}},
"DIR_COMPOSITIONS": "/opt/compose/",
}
lookup = self._make_lookup(vars_)
with self.assertRaises(AnsibleError):
lookup.run(["web-app-foo"], variables=vars_)
def test_a_dedicated_engine_keeps_the_name_its_role_declares(self):
applications = {
"web-app-foo": {
"services": {
"postgres": {
"enabled": True,
"shared": False,
"name": "foo-postgres-1",
}
},
"secrets": {"credentials": {"database_password": "pw"}},
},
"svc-db-postgres": {
"services": {
"postgres": {"version": "16", "ports": {"local": {"postgres": ""}}}
}
},
}
vars_ = {
"applications": applications,
"ports": {"localhost": {"database": {"svc-db-postgres": "5432"}}},
"DIR_COMPOSITIONS": "/opt/compose/",
}
self.assertEqual(self._resolve(vars_)["container"], "foo-postgres-1")
def _mode_vars(self, shared, **extra):
applications = {
"web-app-foo": {
"services": {"postgres": {"enabled": True, "shared": shared}},
"secrets": {"credentials": {"database_password": "pw"}},
},
"svc-db-postgres": {
"services": {
"postgres": {
"name": "postgres",
"version": "16",
"ports": {"local": {"postgres": ""}},
}
}
},
}
return {
"applications": applications,
"ports": {"localhost": {"database": {"svc-db-postgres": "5432"}}},
"DIR_COMPOSITIONS": "/opt/compose/",
**extra,
}
def _dedicated_vars(self, **extra):
return self._mode_vars(False, **extra)
def _resolve(self, vars_):
lookup = self._make_lookup(vars_)
with patch.object(
self.db_lookup_mod,
"entity_name",
side_effect=self._fake_entity_name,
):
return lookup.run(["web-app-foo"], variables=vars_)[0]
def test_swarm_names_the_service_after_the_stack(self):
out = self._resolve(self._dedicated_vars(DEPLOYMENT_MODE="swarm"))
self.assertEqual(out["service_name"], "foo_database")
self.assertIn("resolve-container-id", out["address"])
def test_compose_mode_force_beats_the_cluster_mode(self):
out = self._resolve(
self._dedicated_vars(DEPLOYMENT_MODE="swarm", compose_mode_force="compose")
)
self.assertEqual(out["service_name"], "database")
self.assertEqual(out["address"], out["container"])
def test_an_empty_force_falls_back_to_the_cluster_mode(self):
out = self._resolve(
self._dedicated_vars(DEPLOYMENT_MODE="swarm", compose_mode_force="")
)
self.assertEqual(out["service_name"], "foo_database")
def test_a_consumers_force_does_not_reach_the_central_engine(self):
out = self._resolve(
self._mode_vars(True, DEPLOYMENT_MODE="swarm", compose_mode_force="compose")
)
self.assertEqual(out["service_name"], "postgres_postgres")
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink