mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 19:56:53 +00:00
kevinveenbirkenbachandClaude Opus 5.5
e611e5ea05
sys-svc-objstore includes the provider role with application_id set to "{{ _provider_role }}", which is itself lookup('objstore', ...). For a consumer served only on onion the objstore lookup sets proxy_required and calls lookup('tls', consumer_id); without a consumer, align_domain_to_consumer renders variables['application_id'], which runs the objstore lookup again, and the chain repeats until Ansible stops with "Recursive loop detected" / "Maximum depth exceeded" (fork run 40042690056, jobs Moodle#2 and Nextcloud#12, log line 286323). Only web-app-matrix sets single_mode, so only its multi-mode onion deploys hit it.
The tls lookup takes a consumer= kwarg and hands it to align_domain_to_consumer, the way canonical_url already does, and the objstore lookup names its consumer. The default stays empty, so no other caller changes. Regression tests cover both lookups; an onion consumer now resolves to http://matrix.<onion>/objstore/matrix.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
414 lines
16 KiB
Python
414 lines
16 KiB
Python
"""Unit tests for ``plugins/lookup/objstore.py``.
|
|
|
|
The lookup branches on which object-store engine (seaweedfs|minio) the
|
|
consumer has enabled and yields a uniform S3 connection payload.
|
|
``get_merged_applications`` is mocked so the tests stay hermetic.
|
|
"""
|
|
|
|
import importlib.util
|
|
import unittest
|
|
import unittest.mock as mock
|
|
from typing import ClassVar
|
|
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 ObjstoreLookupTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.mod = _load_module("plugins/lookup/objstore.py", "lookup_objstore")
|
|
|
|
def _make_lookup(self, available_vars: dict):
|
|
lm = self.mod.LookupModule()
|
|
lm._templar = _DummyTemplar(available_vars)
|
|
lm._loader = mock.MagicMock()
|
|
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 _run(self, terms, applications, vars_=None):
|
|
if vars_ is None:
|
|
vars_ = {"DIR_COMPOSITIONS": "/opt/compose/"}
|
|
lookup = self._make_lookup(vars_)
|
|
with (
|
|
mock.patch.object(self.mod, "lookup_loader") as loader_mock,
|
|
patch.object(self.mod, "entity_name", side_effect=self._fake_entity_name),
|
|
):
|
|
loader_mock.get.return_value = mock.MagicMock(
|
|
run=lambda *_a, **_k: [applications]
|
|
)
|
|
return lookup.run(terms, variables=vars_)
|
|
|
|
def test_invalid_terms_raises(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self._run([], {})
|
|
|
|
with self.assertRaises(AnsibleError):
|
|
self._run(["a", "b", "c"], {})
|
|
|
|
def test_empty_consumer_id_raises(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self._run([" "], {})
|
|
|
|
def test_missing_dir_compositions_raises(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self._run(["web-app-foo"], {"web-app-foo": {"services": {}}}, vars_={})
|
|
|
|
def test_no_engine_enabled_returns_disabled_payload(self):
|
|
applications = {"web-app-foo": {"services": {}}}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["id"], "")
|
|
self.assertEqual(out["engine"], "")
|
|
self.assertFalse(out["enabled"])
|
|
self.assertFalse(out["shared"])
|
|
self.assertFalse(out["local"])
|
|
self.assertEqual(out["host"], "")
|
|
self.assertEqual(out["port"], "")
|
|
self.assertEqual(out["network"], "")
|
|
self.assertEqual(out["container"], "")
|
|
self.assertEqual(out["bucket"], "foo")
|
|
self.assertEqual(out["access_key"], "foo")
|
|
self.assertEqual(out["secret_key"], "")
|
|
self.assertEqual(out["region"], "us-east-1")
|
|
self.assertEqual(out["env"], "")
|
|
self.assertEqual(out["volume"], "")
|
|
self.assertEqual(out["endpoint"], "")
|
|
self.assertEqual(out["url"], "")
|
|
self.assertEqual(out["public_domain"], "")
|
|
self.assertEqual(out["public_bucket"], "")
|
|
self.assertEqual(out["public_endpoint"], "")
|
|
self.assertEqual(out["public_url"], "")
|
|
self.assertEqual(out["reach_host"], "127.0.0.1")
|
|
|
|
self.assertEqual(self._run(["web-app-foo", "url"], applications)[0], "")
|
|
|
|
def test_seaweedfs_shared_uses_central_provider_values(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {"seaweedfs": {"enabled": True, "shared": True}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "sk"}},
|
|
},
|
|
"web-svc-seaweedfs": {
|
|
"services": {
|
|
"seaweedfs": {"name": "seaweedfs-central", "api_port": 8334}
|
|
},
|
|
"domains": {"canonical": {"api": "api.seaweedfs.s3.example.com"}},
|
|
},
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["id"], "web-svc-seaweedfs")
|
|
self.assertEqual(out["engine"], "seaweedfs")
|
|
self.assertTrue(out["enabled"])
|
|
self.assertTrue(out["shared"])
|
|
self.assertFalse(out["local"])
|
|
self.assertEqual(out["host"], "seaweedfs-central")
|
|
self.assertEqual(out["port"], 8334)
|
|
self.assertEqual(out["network"], "seaweedfs")
|
|
self.assertEqual(out["container"], "seaweedfs-central")
|
|
self.assertEqual(out["volume"], "")
|
|
self.assertEqual(out["endpoint"], "seaweedfs-central:8334")
|
|
self.assertEqual(out["url"], "http://seaweedfs-central:8334")
|
|
self.assertEqual(out["public_domain"], "api.seaweedfs.s3.example.com")
|
|
self.assertEqual(out["public_bucket"], "api.seaweedfs.s3.example.com/foo")
|
|
self.assertEqual(out["public_endpoint"], "https://api.seaweedfs.s3.example.com")
|
|
self.assertEqual(out["public_url"], "https://api.seaweedfs.s3.example.com/foo")
|
|
|
|
self.assertEqual(
|
|
self._run(["web-app-foo", "url"], applications)[0],
|
|
"http://seaweedfs-central:8334",
|
|
)
|
|
|
|
def test_named_canonical_siblings_do_not_win_over_the_api(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {"minio": {"enabled": True, "shared": True}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "sk"}},
|
|
},
|
|
"web-app-minio": {
|
|
"services": {"minio": {"name": "minio-central"}},
|
|
"domains": {
|
|
"canonical": {
|
|
"console": "console.minio.s3.example.com",
|
|
"api": "api.minio.s3.example.com",
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["public_domain"], "api.minio.s3.example.com")
|
|
self.assertEqual(out["public_endpoint"], "https://api.minio.s3.example.com")
|
|
|
|
def test_credentialed_url_percent_encodes_both_halves(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {"seaweedfs": {"enabled": True, "shared": True}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "a/b+c=d"}},
|
|
},
|
|
"web-svc-seaweedfs": {
|
|
"services": {
|
|
"seaweedfs": {"name": "seaweedfs-central", "api_port": 8334}
|
|
},
|
|
},
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(
|
|
out["credentialed_url"],
|
|
"http://foo:a%2Fb%2Bc%3Dd@seaweedfs-central:8334",
|
|
)
|
|
self.assertEqual(out["url"], "http://seaweedfs-central:8334")
|
|
|
|
def test_credentialed_url_is_empty_without_an_engine(self):
|
|
applications = {"web-app-foo": {"services": {}}}
|
|
self.assertEqual(
|
|
self._run(["web-app-foo", "credentialed_url"], applications)[0], ""
|
|
)
|
|
|
|
def test_public_url_scheme_follows_provider_tls(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {"minio": {"enabled": True, "shared": True}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "sk"}},
|
|
},
|
|
"web-app-minio": {
|
|
"services": {"minio": {"name": "minio-central"}},
|
|
"server": {"tls": {"enabled": False}},
|
|
"domains": {"canonical": {"api": "api.minio.s3.example.com"}},
|
|
},
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["public_domain"], "api.minio.s3.example.com")
|
|
self.assertEqual(out["public_url"], "http://api.minio.s3.example.com/foo")
|
|
|
|
def test_seaweedfs_dedicated_matches_embedded_instance(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {"seaweedfs": {"enabled": True, "shared": False}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "sk"}},
|
|
}
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["id"], "web-svc-seaweedfs")
|
|
self.assertEqual(out["engine"], "seaweedfs")
|
|
self.assertTrue(out["enabled"])
|
|
self.assertFalse(out["shared"])
|
|
self.assertTrue(out["local"])
|
|
self.assertEqual(out["host"], "seaweedfs")
|
|
self.assertEqual(out["port"], 8333)
|
|
self.assertEqual(out["network"], "foo")
|
|
self.assertEqual(out["container"], "foo-seaweedfs")
|
|
self.assertEqual(out["volume"], "foo_seaweedfs")
|
|
self.assertEqual(out["url"], "http://seaweedfs:8333")
|
|
self.assertEqual(out["env"], "/opt/compose/foo/.env/objstore.env")
|
|
self.assertEqual(out["image"], "chrislusf/seaweedfs")
|
|
self.assertEqual(out["version"], "latest")
|
|
|
|
def test_minio_shared_defaults_port_9000(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {"minio": {"enabled": True, "shared": True}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "sk"}},
|
|
},
|
|
"web-app-minio": {"services": {"minio": {"name": "minio-central"}}},
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["id"], "web-app-minio")
|
|
self.assertEqual(out["engine"], "minio")
|
|
self.assertTrue(out["enabled"])
|
|
self.assertTrue(out["shared"])
|
|
self.assertFalse(out["local"])
|
|
self.assertEqual(out["host"], "minio-central")
|
|
self.assertEqual(out["port"], 9000)
|
|
self.assertEqual(out["network"], "minio")
|
|
self.assertEqual(out["url"], "http://minio-central:9000")
|
|
|
|
def test_multiple_engines_enabled_raises(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {
|
|
"seaweedfs": {"enabled": True, "shared": False},
|
|
"minio": {"enabled": True, "shared": False},
|
|
}
|
|
}
|
|
}
|
|
|
|
with self.assertRaises(AnsibleError):
|
|
self._run(["web-app-foo"], applications)
|
|
|
|
def test_secret_key_and_public_passthrough(self):
|
|
applications = {
|
|
"web-app-foo": {
|
|
"services": {
|
|
"seaweedfs": {"enabled": True, "shared": False, "public": True}
|
|
},
|
|
"secrets": {"credentials": {"objstore_secret_key": "s3cr3t"}},
|
|
}
|
|
}
|
|
|
|
out = self._run(["web-app-foo"], applications)[0]
|
|
|
|
self.assertEqual(out["secret_key"], "s3cr3t")
|
|
self.assertTrue(out["public"])
|
|
self.assertEqual(out["access_key"], "foo")
|
|
self.assertEqual(out["bucket"], "foo")
|
|
|
|
self.assertEqual(
|
|
self._run(["web-app-foo", "secret_key"], applications)[0], "s3cr3t"
|
|
)
|
|
|
|
|
|
class ObjstoreSameOriginProxyTests(unittest.TestCase):
|
|
"""Pins the rebase of the browser-facing values onto the consumer's own
|
|
origin. web-svc-seaweedfs publishes no onion vhost (the tor-contract in its
|
|
meta/services.yml), so a page served only on an onion reaches the provider's
|
|
clearnet host through no route at all.
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.mod = _load_module("plugins/lookup/objstore.py", "lookup_objstore")
|
|
|
|
APPLICATIONS: ClassVar[dict] = {
|
|
"web-app-foo": {
|
|
"services": {"seaweedfs": {"enabled": True, "shared": True}},
|
|
"secrets": {"credentials": {"objstore_secret_key": "sk"}},
|
|
},
|
|
"web-svc-seaweedfs": {
|
|
"services": {"seaweedfs": {"name": "seaweedfs-central"}},
|
|
"domains": {"canonical": {"api": "api.seaweedfs.s3.example.com"}},
|
|
},
|
|
}
|
|
|
|
def _resolve(self, *, consumer_domains, consumer_web="http"):
|
|
vars_ = {"DIR_COMPOSITIONS": "/opt/compose/", "TLS_ENABLED": True}
|
|
lm = self.mod.LookupModule()
|
|
lm._templar = _DummyTemplar(vars_)
|
|
lm._loader = mock.MagicMock()
|
|
|
|
payloads = {
|
|
"applications": self.APPLICATIONS,
|
|
"domains": {"web-app-foo": consumer_domains},
|
|
"tls": {
|
|
"domain": consumer_domains[0],
|
|
"protocols": {"web": consumer_web},
|
|
},
|
|
}
|
|
|
|
self.run_kwargs = {}
|
|
|
|
def fake_get(name, **_kwargs):
|
|
def run(_terms, **kwargs):
|
|
self.run_kwargs[name] = kwargs
|
|
return [payloads[name]]
|
|
|
|
return mock.MagicMock(run=run)
|
|
|
|
with (
|
|
mock.patch.object(self.mod, "lookup_loader") as loader_mock,
|
|
patch.object(
|
|
self.mod,
|
|
"entity_name",
|
|
side_effect=lambda r: r.split("-", 2)[-1],
|
|
),
|
|
):
|
|
loader_mock.get.side_effect = fake_get
|
|
return lm.run(["web-app-foo"], variables=vars_)[0]
|
|
|
|
def test_an_onion_only_consumer_gets_the_bucket_on_its_own_origin(self):
|
|
out = self._resolve(consumer_domains=["foo.abcdefghij.onion"])
|
|
|
|
self.assertTrue(out["proxy_required"])
|
|
self.assertEqual(out["proxy_prefix"], "/objstore")
|
|
self.assertEqual(
|
|
out["public_url"],
|
|
"http://foo.abcdefghij.onion/objstore/foo",
|
|
"the browser must dial the page's own origin, never the provider's "
|
|
"clearnet host it cannot resolve through Tor",
|
|
)
|
|
self.assertEqual(
|
|
out["public_endpoint"],
|
|
"https://api.seaweedfs.s3.example.com",
|
|
"a server-side S3 client runs in the container, which has no Tor "
|
|
"resolver, so its endpoint must stay on the provider's own host",
|
|
)
|
|
self.assertEqual(
|
|
(out["upstream_domain"], out["upstream_scheme"]),
|
|
("api.seaweedfs.s3.example.com", "https"),
|
|
"the nginx fragment still has to reach the provider's own vhost, so "
|
|
"the rebase must not overwrite the upstream it proxies to",
|
|
)
|
|
|
|
def test_the_consumer_origin_is_resolved_for_the_named_consumer(self):
|
|
self._resolve(consumer_domains=["foo.abcdefghij.onion"])
|
|
|
|
self.assertEqual(
|
|
self.run_kwargs["tls"].get("consumer"),
|
|
"web-app-foo",
|
|
"sys-svc-objstore binds application_id to a template over this very "
|
|
"lookup, so tls falling back to application_id recurses until Jinja "
|
|
"gives up",
|
|
)
|
|
|
|
def test_a_clearnet_consumer_keeps_the_provider_host(self):
|
|
out = self._resolve(consumer_domains=["foo.example.com"], consumer_web="https")
|
|
|
|
self.assertFalse(
|
|
out["proxy_required"],
|
|
"a clearnet page reaches the provider directly, so no deployment "
|
|
"pays for a proxy hop it does not need",
|
|
)
|
|
self.assertEqual(out["public_domain"], "api.seaweedfs.s3.example.com")
|
|
self.assertEqual(out["public_url"], "https://api.seaweedfs.s3.example.com/foo")
|
|
|
|
def test_a_consumer_on_both_families_keeps_the_provider_host(self):
|
|
out = self._resolve(
|
|
consumer_domains=["foo.example.com", "foo.abcdefghij.onion"],
|
|
)
|
|
|
|
self.assertFalse(out["proxy_required"])
|
|
self.assertEqual(out["public_domain"], "api.seaweedfs.s3.example.com")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|