import json import sys import tempfile from pathlib import Path import unittest from unittest.mock import patch SCRIPTS = Path(__file__).resolve().parents[1] sys.path.insert(0, str(SCRIPTS)) import lxc_apps ICON = 'https://cdn.jsdelivr.net/gh/selfhst/icons@main/webp/go2rtc.webp' VERSION_FILE = {'path': '/opt/frigate/frigate/version.py', 'regex': 'VERSION = "([0-9]+(?:\\.[0-9]+)+)'} ENDPOINTS = [ {'label': 'Frigate WebUI', 'scheme': 'http', 'port': 5000, 'path': '/', 'source': 'x'}, {'label': 'go2rtc WebUI', 'scheme': 'http', 'port': 1984, 'path': '/', 'source': 'x'}, ] class CatalogFixture(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.root = Path(tmp.name) (self.root / 'catalog/apps').mkdir(parents=True) (self.root / 'catalog/index.json').write_text(json.dumps({'applications': [ {'id': 'frigate', 'template_id': 'image-frigate', 'template': 'apps/frigate.json', 'icon': 'https://cdn.jsdelivr.net/gh/selfhst/icons@main/webp/frigate.webp'}]})) catalog_endpoints = [dict(ENDPOINTS[0]), dict(ENDPOINTS[1], icon=ICON)] (self.root / 'catalog/apps/frigate.json').write_text(json.dumps( {'id': 'image-frigate', 'first_run': {'endpoints': catalog_endpoints, 'version_file': VERSION_FILE}})) patches = [patch.object(lxc_apps, '_OCI_INSTANCE_ROOT', str(self.root / 'instances')), patch.object(lxc_apps, '_OCI_CATALOG_INDEX', str(self.root / 'catalog/index.json')), patch.object(lxc_apps, '_oci_catalog_cache', None)] for item in patches: item.start() self.addCleanup(item.stop) def install(self, endpoints, template_id='image-frigate'): (self.root / 'instances/190').mkdir(parents=True) (self.root / 'instances/190/oci-compose.json').write_text(json.dumps({ 'vmid': 190, 'status': 'installed', 'installation_id': 'install-190', 'observed': {'image': {'manifest_digest': 'sha256:' + 'b1' * 32, 'architecture': 'amd64'}}, 'template': {'id': template_id, 'catalog_ui': {'title': 'Frigate'}, 'first_run': {'endpoints': endpoints}, 'container_contract': {'image': {'reference': 'ghcr.io/blakeblackshear/frigate:stable'}, 'ports': [{'container_port': 8971}]}}})) return lxc_apps._oci_instance_meta(190) class OciEndpointTests(CatalogFixture): def test_second_endpoint_is_its_own_service_with_the_current_catalog_icon(self): # The record was written before the catalog gave go2rtc an icon. meta = self.install(ENDPOINTS) self.assertEqual(meta['endpoints'], [ {'port': 5000, 'scheme': 'http', 'path': '/', 'description': '', 'logo_url': ''}, {'port': 1984, 'scheme': 'http', 'path': '/', 'description': 'go2rtc', 'logo_url': ICON}, ]) self.assertEqual(meta['endpoint_port'], 5000) def test_single_endpoint_and_unknown_template(self): meta = self.install([ENDPOINTS[0]], template_id='image-other') self.assertEqual(meta['endpoints'], [ {'port': 5000, 'scheme': 'http', 'path': '/', 'description': '', 'logo_url': ''}]) def test_generic_second_endpoint_is_not_a_service(self): # LinuxServer lists the same interface over http and https as "Web UI 1" and "Web UI 2". meta = self.install([dict(ENDPOINTS[0], label='Web UI 1', port=3000), dict(ENDPOINTS[0], label='Web UI 2', port=3001, scheme='https')], template_id='image-krita') self.assertEqual([e['port'] for e in meta['endpoints']], [3000]) def test_service_name_from_label(self): for label, name in [('go2rtc WebUI', 'go2rtc'), ('Admin Web UI', 'Admin'), ('Setup UI', 'Setup'), ('Dashboard', 'Dashboard'), ('WebUI', 'WebUI'), ('', '')]: with self.subTest(label=label): self.assertEqual(lxc_apps._oci_service_name(label), name) def test_suggestion_offers_every_endpoint(self): self.install(ENDPOINTS) with patch.object(lxc_apps, '_probe_listening_ports', return_value=[]), \ patch.object(lxc_apps, '_helper_slug_meta', return_value=None), \ patch.object(lxc_apps, '_fetch_tracking_hints', return_value={}), \ patch.object(lxc_apps, '_oci_image_versions', return_value={'error': 'offline'}), \ patch.object(lxc_apps, '_read_sidecar', return_value=None): suggestion = lxc_apps.get_suggestions(190) self.assertEqual(suggestion['default_ports'], [5000, 1984]) self.assertEqual([d['description'] for d in suggestion['port_details']], ['', 'go2rtc']) self.assertEqual(suggestion['port_details'][1]['logo_url'], ICON) def test_automatic_registration_registers_every_endpoint(self): self.install(ENDPOINTS) with patch.object(lxc_apps, '_read_sidecar', return_value=None), \ patch.object(lxc_apps, '_oci_dismissed', return_value={}), \ patch.object(lxc_apps, 'add_app', return_value=(True, {})) as add: self.assertTrue(lxc_apps.ensure_oci_registration(190)) ports = add.call_args.args[1]['ports'] self.assertEqual([(p['port'], p['description']) for p in ports], [(5000, ''), (1984, 'go2rtc')]) self.assertNotIn('logo_url', ports[0]) self.assertEqual(ports[1]['logo_url'], ICON) class OciGuestVersionTests(CatalogFixture): def test_version_read_from_the_file_the_catalog_names(self): with patch.object(lxc_apps, '_pct_exec', return_value=(0, 'VERSION = "0.18.0-77a66e7"\n', '')) as run: self.assertEqual(lxc_apps._oci_guest_version(190, 'image-frigate'), '0.18.0') run.assert_called_once_with(190, ['cat', '/opt/frigate/frigate/version.py']) def test_stopped_container_and_template_without_version_file(self): with patch.object(lxc_apps, '_pct_exec', return_value=(1, '', 'CT 190 not running')): self.assertIsNone(lxc_apps._oci_guest_version(190, 'image-frigate')) with patch.object(lxc_apps, '_pct_exec') as run: self.assertIsNone(lxc_apps._oci_guest_version(190, 'image-other')) run.assert_not_called() def test_image_without_version_label_uses_the_guest(self): self.install(ENDPOINTS) # A cached answer without a version must not stop the guest read. known = {'installed_digest': 'sha256:' + 'b1' * 32, 'installed_version': None, 'image_created': '2026-09-06'} with patch.object(lxc_apps, '_oci_operation_running', return_value=False), \ patch.object(lxc_apps, '_oci_state_module', return_value=object()), \ patch.object(lxc_apps, '_pct_exec', return_value=(0, 'VERSION = "0.18.0-77a66e7"', '')): result = lxc_apps._oci_image_versions(190, known=known, with_latest=False) self.assertEqual(result['installed_version'], '0.18.0') self.assertEqual(result['image_created'], '2026-09-06') if __name__ == '__main__': unittest.main()