"""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_short_read_only_time_mount_is_removed_from_stack(self): text = COMPOSE.replace(" ports:", " volumes: [/etc/localtime:/etc/localtime:ro]\n ports:") template = custom.template_from_compose(text) app = next(service for service in ordered_services(template) if service['name'] == 'app') self.assertEqual(app['compose']['volumes'], []) self.assertFalse(any('System path' in error for error in template['compatibility']['untranslated_blockers'])) self.assertTrue(template['compatibility']['automatic_install_candidate']) 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_each_registry_line_names_its_image(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, 'in registry'), (False, 'app missing')]), \ patch.object(custom, '_registry_refused', return_value=False), \ patch.object(custom, 'install_template'): custom.explore(ui) self.assertIn('postgres:16: in registry', ui.text) self.assertNotIn('adminer:4: app missing', ui.text) @patch('proxmenux_oci.i18n.language', return_value='en') def test_stack_summary_does_not_announce_questions_about_its_variables(self, _): template = custom.template_from_compose(COMPOSE.replace( " ports:", " environment:\n ADMINER_DEFAULT_SERVER: db\n ports:")) summary = custom.describe(template) self.assertNotIn('Variables the installation asks for', summary) self.assertIn('Variables (app):\n ADMINER_DEFAULT_SERVER', summary) single = custom.template_from_compose('services: {app: {image: nginx:1.27, environment: {MODE: x}}}') self.assertIn('Variables the installation asks for', custom.describe(single)) def test_the_name_given_names_the_stack(self): template = custom.template_from_compose('name: adminer-postgres\n' + COMPOSE) class UI: def ask(self, text, default=None, required=True): self.default = default return 'tools' ui = UI() self.assertEqual(custom._name(ui, template), 'tools') self.assertEqual(ui.default, 'adminer-postgres') self.assertEqual(template['compose_stack']['project_name'], 'tools') 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()