Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-02 10:52:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(oci): support custom multi-service Compose applications

Browse Source
This commit is contained in:
Jonatan
2026-10-28 06:28:18 +01:00
parent cdb1bd0cbf
commit 692c7f7256
14 changed files with 424 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
oci/tests/test_custom_compose_stacks.py
+111
View File
@@ -0,0 +1,111 @@
"""Pasted Compose files enter the same checked stack flow as catalog recipes."""
import sys
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'src'))
from proxmenux_oci import custom
from proxmenux_oci.converter import ConversionError
from proxmenux_oci.stack import ordered_services, resolve_environments, service_template
COMPOSE = '''services:
db:
image: postgres:16
environment:
POSTGRES_PASSWORD: supplied-password
volumes: [dbdata:/var/lib/postgresql/data]
app:
image: adminer:4
ports: ['8080:8080']
depends_on: [db]
volumes:
dbdata:
'''
class CustomComposeStacks(unittest.TestCase):
def test_stack_uses_existing_installer_and_preserves_versions_and_environment(self):
template = custom.template_from_compose(COMPOSE)
self.assertTrue(template['compatibility']['automatic_install_candidate'])
self.assertEqual(template['proxmox']['installer_profile']['stack_driver'], 'generic-multi-lxc-stack')
self.assertEqual(template['compose_stack']['main_service'], 'app')
services = ordered_services(template)
self.assertEqual([s['name'] for s in services], ['db', 'app'])
self.assertEqual([s['image'] for s in services], ['postgres:16', 'adminer:4'])
self.assertEqual(resolve_environments(services, 'UTC')['db']['POSTGRES_PASSWORD'], 'supplied-password')
self.assertEqual(service_template(template, services[0])['container_contract']['image']['reference'], 'postgres:16')
self.assertNotIn('supplied-password', custom.describe(template))
def test_stack_summary_hides_credentials_embedded_in_a_url(self):
document = COMPOSE.replace(
" ports:",
" environment: {DATABASE_URL: 'postgres://user:embedded-secret@db/app'}\n ports:")
template = custom.template_from_compose(document)
summary = custom.describe(template)
self.assertIn('DATABASE_URL', summary)
self.assertNotIn('embedded-secret', summary)
self.assertNotIn('postgres://user:', summary)
app = next(service for service in ordered_services(template) if service['name'] == 'app')
self.assertIn('embedded-secret', app['compose']['environment']['DATABASE_URL'])
def test_invalid_stack_metadata_has_a_conversion_error(self):
for metadata in ('x-casaos: not-a-mapping', 'x-casaos: [app]',
'x-casaos: {main: missing-service}'):
with self.subTest(metadata=metadata), self.assertRaises(ConversionError):
custom.template_from_compose(COMPOSE + metadata + '\n')
def test_written_digest_and_tag_survive_in_full_image_contracts(self):
digest = 'sha256:' + 'a' * 64
reference = 'ghcr.io/example/adminer:4@' + digest
template = custom.template_from_compose(COMPOSE.replace('adminer:4', reference))
main_image = template['container_contract']['image']
for image in (main_image, service_template(template, next(
service for service in ordered_services(template) if service['name'] == 'app'),
main=True)['container_contract']['image']):
self.assertEqual(image['reference'], reference)
self.assertEqual(image['registry'], 'ghcr.io')
self.assertEqual(image['repository'], 'ghcr.io/example/adminer')
self.assertEqual(image['tag'], '4')
self.assertEqual(image['digest'], digest)
self.assertEqual(image['pull_policy'],
'resolve-written-reference-to-architecture-digest-at-install')
dependency = service_template(template, ordered_services(template)[0])['container_contract']['image']
self.assertEqual(dependency['reference'], 'postgres:16')
self.assertEqual(dependency['repository'], 'postgres')
self.assertEqual(dependency['tag'], '16')
self.assertIsNone(dependency['digest'])
def test_reject_unsafe_dependency_mount_or_build(self):
for value in (' volumes: [/var/run/docker.sock:/var/run/docker.sock]',
' build: .'):
with self.subTest(value=value), self.assertRaises(ConversionError):
custom.template_from_compose(COMPOSE.replace(' volumes: [dbdata:/var/lib/postgresql/data]', value))
def test_shared_storage_is_supported_by_custom_stack(self):
template = custom.template_from_compose(COMPOSE.replace(" ports:", " volumes: [dbdata:/shared]\n ports:"))
self.assertTrue(template['compatibility']['automatic_install_candidate'])
self.assertEqual(template['compatibility']['untranslated_blockers'], [])
def test_every_image_checked_before_install(self):
class UI:
def message(self, text, title):
self.text = text
ui = UI()
with patch.object(custom, 'read_definition', return_value=(COMPOSE, 'Compose')), \
patch.object(custom, 'registry_report', side_effect=[(True, 'db ok'), (False, 'app missing')]) as report, \
patch.object(custom, '_registry_refused', return_value=False), \
patch.object(custom, 'install_template') as install:
custom.explore(ui)
self.assertEqual([c.args[0] for c in report.call_args_list], ['postgres:16', 'adminer:4'])
install.assert_not_called()
self.assertIn('app missing', ui.text)
def test_single_image_still_uses_single_container_flow(self):
template = custom.template_from_compose('services: {app: {image: nginx:1.27}}')
self.assertNotEqual(template['proxmox'].get('installer_profile', {}).get('stack_driver'), 'generic-multi-lxc-stack')
self.assertEqual(template['container_contract']['image']['reference'], 'nginx:1.27')
if __name__ == '__main__':
unittest.main()
oci/tests/test_custom_multiapp_stack.py
+121
View File
@@ -0,0 +1,121 @@
"""Custom stacks keep each app's endpoint and actual shared storage identity."""
from pathlib import Path
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'src'))
from proxmenux_oci.casaos import convert_casaos_compose
from proxmenux_oci.stack import apply_stack_support, assess, build_stack, DefaultsUI, service_template
from proxmenux_oci.installer import DEFAULT_MODE, ADVANCED_MODE
COMPOSE = '''name: arr-test
services:
sonarr:
image: lscr.io/linuxserver/sonarr:latest
ports: ['8989:8989']
environment: {PUID: '1000', PGID: '1000'}
volumes: [sonarr-config:/config, shared-data:/data]
radarr:
image: lscr.io/linuxserver/radarr:latest
ports: ['7878:7878']
environment: {PUID: '1000', PGID: '1000'}
volumes: [radarr-config:/config, shared-data:/data]
volumes:
sonarr-config:
radarr-config:
shared-data:
x-casaos:
main: sonarr
port_map: '8989'
scheme: http
index: /
title: {en_US: Arr Test}
'''
def template(custom=True, text=COMPOSE):
result = convert_casaos_compose(text, 'local', '', 'local/arr-test/docker-compose.yml', '')
result['proxmox']['custom_compose_stack'] = custom
apply_stack_support(result)
return result
class UI(DefaultsUI):
def __init__(self):
self.asks = []
def ask(self, text, default=None, **kwargs):
self.asks.append(text)
return default or ''
class CustomMultiAppStackTests(unittest.TestCase):
def test_custom_opt_in_supports_two_web_apps_but_catalog_remains_guarded(self):
self.assertEqual(assess(template()), [])
self.assertTrue(assess(template(custom=False)))
def test_app_without_own_port_is_rejected(self):
result = template(text=COMPOSE.replace(" ports: ['7878:7878']\n", ''))
self.assertTrue(any('health and persistence' in e for e in assess(result)))
def test_service_image_contract_preserves_tag_and_digest(self):
parent = template()
service = parent['compose_stack']['services'][0]
for reference, repository, tag, digest in (
('ghcr.io/example/app:2.4', 'ghcr.io/example/app', '2.4', None),
('ghcr.io/example/app:2.4@sha256:'+'a'*64, 'ghcr.io/example/app', '2.4', 'sha256:'+'a'*64),
):
with self.subTest(reference=reference):
service['image'] = reference
service['compose']['image'] = reference
image = service_template(parent, service)['container_contract']['image']
self.assertEqual((image['reference'], image['repository'], image['tag'], image['digest']),
(reference, repository, tag, digest))
self.assertEqual(image['registry'], 'ghcr.io')
def test_compose_extension_metadata_is_accepted_only_for_custom(self):
text = 'x-common-environment: &common {PUID: \'1000\'}\n' + COMPOSE
self.assertEqual(assess(template(text=text)), [])
self.assertTrue(assess(template(custom=False, text=text)))
def test_shared_owners_must_match(self):
result = template(text=COMPOSE.replace("environment: {PUID: '1000', PGID: '1000'}", "environment: {PUID: '1001', PGID: '1000'}", 1))
self.assertTrue(any('shared' in e for e in assess(result)))
def test_advanced_addresses_are_kept_per_service(self):
ui = UI()
supplied = {'sonarr': '192.0.2.21/24', 'radarr': '192.0.2.22/24'}
with patch('proxmenux_oci.installer.ask_storage', return_value='local-zfs'), \
patch('proxmenux_oci.installer.ask_bridge', return_value='vmbr0'), \
patch('proxmenux_oci.host.timezone', return_value='UTC'), \
patch('proxmenux_oci.custom_mounts.ask_stack_custom_mounts'), \
patch('proxmenux_oci.extra_devices.ask_stack_extra_devices'), \
patch('proxmenux_oci.network.ask_addresses', return_value=(supplied, '192.0.2.1')):
plan = build_stack(template(), ui, ADVANCED_MODE)
self.assertEqual({s['name']:s['frontend_ipv4'] for s in plan['services']}, supplied)
self.assertEqual(plan['network']['frontend_gateway'], '192.0.2.1')
def test_build_assigns_distinct_endpoints_and_shared_bind_source(self):
ui = UI()
with patch('proxmenux_oci.installer.ask_storage', return_value='local-zfs'), \
patch('proxmenux_oci.installer.ask_bridge', return_value='vmbr0'), \
patch('proxmenux_oci.host.timezone', return_value='UTC'), \
patch('proxmenux_oci.network.ask_addresses', side_effect=lambda ui, bridge, labels: ({name:'dhcp' for name in labels}, None)):
plan = build_stack(template(), ui, DEFAULT_MODE)
members = {s['name']:s for s in plan['services']}
self.assertEqual(members['sonarr']['healthcheck']['endpoint']['port'], 8989)
self.assertEqual(members['radarr']['healthcheck']['endpoint']['port'], 7878)
for member in members.values():
self.assertTrue(member['frontend'])
self.assertEqual(member['frontend_ipv4'], 'dhcp')
binds = [next(m for m in s['deployment']['mounts'] if m['container_path']=='/data') for s in members.values()]
self.assertEqual(binds[0]['source'], binds[1]['source'])
self.assertEqual(binds[0]['host_owner'], [101000, 101000])
self.assertTrue(all(m['type']=='host-bind' and m['backup'] is False for m in binds))
self.assertEqual(len([text for text in ui.asks if text.endswith('shared-data')]), 1)
configs = [next(m for m in s['deployment']['mounts'] if m['container_path']=='/config') for s in members.values()]
self.assertTrue(all(m['type']=='managed-volume' for m in configs))
if __name__ == '__main__':
unittest.main()