mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 11:46:35 +00:00
databases.csv rows come only from consumers, through sys-svc-rdbms. A database provider deployed without one, like the lean svc-db-postgres#1 variant (mirror, tor, backup and postgres, nothing else), therefore has no row. baudolo copies its volume live and records it undumped, and the drill failed with "postgres_data (postgres) ... no sql dump" in fork run 35731501845, on a cluster that holds only what the engine creates itself. The new database_consumerless_volumes lookup names the data volume of every deployed provider that no deployed role consumes as a shared database; a dedicated database brings its own volume and does not count. test.env hands the list to verify/backup.sh, which notes those volumes instead of failing. Every other undumped database volume still fails, above all one whose consumer forgot to seed its row, since that consumer is deployed. Only the drill's verdict changes; what gets backed up does not, and the CLI tests run on dev, act and github runtimes only. The one case it now lets pass is a dev host still carrying the database of an app its inventory no longer lists: that data sits undumped in the provider volume and reads as NOTE instead of FAIL. A CI job always starts on a fresh host. Counting the databases in the copied data directory was rejected: the bootstrap set depends on the image, and postgis/postgis adds template_postgis to it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""Unit tests for ``plugins/lookup/database_consumerless_volumes.py``.
|
|
|
|
Pins the consumer predicate: a deployed role consumes a provider when it
|
|
enables a shared database of that provider's engine. A dedicated database, a
|
|
provider itself, or a consumer of the other engine never counts.
|
|
"""
|
|
|
|
import importlib.util
|
|
import unittest
|
|
import unittest.mock as mock
|
|
|
|
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
|
|
|
|
|
|
def _database(engine: str, *, shared: bool = True) -> dict:
|
|
return {"services": {engine: {"enabled": True, "shared": shared}}}
|
|
|
|
|
|
class DatabaseConsumerlessVolumesTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.mod = _load_module(
|
|
"plugins/lookup/database_consumerless_volumes.py",
|
|
"lookup_database_consumerless_volumes",
|
|
)
|
|
|
|
def setUp(self):
|
|
self._original_loader = self.mod.lookup_loader
|
|
self._applications: dict = {}
|
|
|
|
def _get(name, **_kwargs):
|
|
if name == "applications":
|
|
return mock.MagicMock(run=lambda *_a, **_k: [self._applications])
|
|
return mock.MagicMock(run=self._volume_run)
|
|
|
|
loader_mock = mock.MagicMock()
|
|
loader_mock.get.side_effect = _get
|
|
self.mod.lookup_loader = loader_mock
|
|
|
|
def tearDown(self):
|
|
self.mod.lookup_loader = self._original_loader
|
|
|
|
@staticmethod
|
|
def _volume_run(terms, **_kwargs):
|
|
provider, semantic = terms
|
|
return [{"name": f"{provider.removeprefix('svc-db-')}_{semantic}"}]
|
|
|
|
def _run(self, group_names, applications, terms=()):
|
|
self._applications = applications
|
|
lk = self.mod.LookupModule()
|
|
lk._loader = mock.MagicMock()
|
|
lk._templar = mock.MagicMock()
|
|
return lk.run(list(terms), variables={"group_names": list(group_names)})
|
|
|
|
def test_terms_are_refused(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self._run([], {}, terms=("postgres",))
|
|
|
|
def test_a_provider_without_a_consumer_is_named(self):
|
|
result = self._run(
|
|
["svc-db-postgres", "web-svc-mirror"],
|
|
{"svc-db-postgres": _database("postgres")},
|
|
)
|
|
self.assertEqual(result, [["postgres_data"]])
|
|
|
|
def test_a_shared_consumer_claims_its_provider(self):
|
|
result = self._run(
|
|
["svc-db-postgres", "web-app-matomo"],
|
|
{"web-app-matomo": _database("postgres")},
|
|
)
|
|
self.assertEqual(result, [[]])
|
|
|
|
def test_a_dedicated_database_claims_no_provider(self):
|
|
result = self._run(
|
|
["svc-db-postgres", "web-app-a"],
|
|
{"web-app-a": _database("postgres", shared=False)},
|
|
)
|
|
self.assertEqual(result, [["postgres_data"]])
|
|
|
|
def test_a_consumer_claims_only_its_own_engine(self):
|
|
result = self._run(
|
|
["svc-db-mariadb", "svc-db-postgres", "web-app-matomo"],
|
|
{"web-app-matomo": _database("mariadb")},
|
|
)
|
|
self.assertEqual(result, [["postgres_data"]])
|
|
|
|
def test_a_provider_that_is_not_deployed_is_not_named(self):
|
|
self.assertEqual(self._run(["web-app-a"], {}), [[]])
|
|
|
|
def test_a_role_with_two_database_services_fails_loudly(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self._run(
|
|
["svc-db-postgres", "web-app-a"],
|
|
{
|
|
"web-app-a": {
|
|
"services": {
|
|
"mariadb": {"enabled": True, "shared": True},
|
|
"postgres": {"enabled": True, "shared": True},
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|