from pathlib import Path import pytest from fastapi.testclient import TestClient from sqlmodel import SQLModel, create_engine @pytest.fixture def api_client(tmp_path, monkeypatch): db_path = tmp_path / "inventory_test.db" monkeypatch.setenv("APPOPS_DB_PATH", str(db_path)) monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token") from backend.app.main import app from backend.app.db import session from backend.app.db.init_db import init_db engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False}) session.engine = engine SQLModel.metadata.drop_all(engine) SQLModel.metadata.create_all(engine) init_db() with TestClient(app) as client: yield client def test_classifier_known_patterns(): from backend.app.services.app_classifier import classify_container cases = [ ({'name': 'binhex-sonarr', 'image': 'binhex/arch-sonarr', 'ports': ['8989:8989']}, 'sonarr', 'media', 'sonarr', 'medium', False), ({'name': 'binhex-radarr', 'image': 'binhex/arch-radarr', 'ports': ['7878:7878']}, 'radarr', 'media', 'radarr', 'medium', False), ({'name': 'authentik-server', 'image': 'ghcr.io/goauthentik/server'}, 'authentik', 'auth', 'authentik', 'critical', True), ({'name': 'postgres', 'image': 'postgres:16'}, 'postgres', 'database', 'none', 'critical', True), ({'name': 'nginxproxymanager', 'image': 'jc21/nginx-proxy-manager'}, 'nginx_proxy_manager', 'network', 'nginx_proxy_manager', 'critical', True), ({'name': 'widefrog', 'image': 'widefrog/app', 'ports': ['1207:1207']}, 'widefrog', 'media', 'widefrog', 'medium', False), ({'name': 'adguard-home', 'image': 'adguard/adguardhome'}, 'adguard_home', 'network', 'none', 'high', True), ({'name': 'audiobookshelf', 'image': 'ghcr.io/advplyr/audiobookshelf'}, 'audiobookshelf', 'media', 'none', 'medium', False), ({'name': 'vaultwarden', 'image': 'vaultwarden/server'}, 'vaultwarden', 'auth', 'none', 'critical', True), ({'name': 'netdata', 'image': 'netdata/netdata'}, 'netdata', 'monitoring', 'none', 'medium', False), ({'name': 'Gpu Hot', 'image': ''}, 'gpu_hot', 'monitoring', 'none', 'medium', False), ({'name': 'System Monitor', 'image': ''}, 'system_monitor', 'monitoring', 'none', 'medium', False), ({'name': 'Unraid Appops Gateway 1216', 'image': ''}, 'unraid_appops_gateway', 'infrastructure', 'none', 'critical', True), ({'name': 'netronome', 'image': 'ghcr.io/autobrr/netronome'}, 'netronome', 'monitoring', 'none', 'medium', False), ({'name': 'grimmory', 'image': 'ghcr.io/grimmory-tools/grimmory'}, 'grimmory', 'media', 'none', 'medium', False), ({'name': 'leafwiki', 'image': 'ghcr.io/perber/leafwiki'}, 'leafwiki', 'productivity', 'none', 'medium', False), ({'name': 'hivekeep', 'image': 'ghcr.io/marlburrow/hivekeep'}, 'hivekeep', 'infrastructure', 'none', 'high', True), ({'name': 'stirlingpdf', 'image': 'stirlingtools/stirling-pdf'}, 'stirlingpdf', 'productivity', 'none', 'medium', False), ({'name': 'geointel', 'image': 'geointel-all-in-one:latest'}, 'geointel', 'ai', 'none', 'medium', True), ({'name': 'opscenter-it-tools', 'image': 'opscenter-it-tools:latest'}, 'opscenter_it_tools', 'dev', 'none', 'high', True), ({'name': 'projectbrain', 'image': 'projectbrain-unraid:latest'}, 'projectbrain', 'productivity', 'none', 'high', True), ({'name': 'psforge', 'image': 'psforge:local'}, 'psforge', 'productivity', 'none', 'medium', True), ({'name': 'QuizMonitor', 'image': 'quiz-monitor:latest'}, 'quizmonitor', 'ai', 'none', 'high', True), ({'name': 'spoortracker', 'image': 'spoortracker:unraid'}, 'spoortracker', 'monitoring', 'none', 'low', False), ({'name': 'certscope', 'image': 'certscope-app'}, 'certscope', 'monitoring', 'none', 'low', False), ({'name': 'ps4-http', 'image': 'python:3-alpine'}, 'ps4_http', 'infrastructure', 'none', 'low', False), ({'name': 'launchpad', 'image': 'launchpad:local'}, 'launchpad', 'infrastructure', 'none', 'medium', False), ({'name': 'media-codec-dashboard', 'image': 'media-codec-dashboard-media-codec-dashboard'}, 'media_codec_dashboard', 'media', 'none', 'medium', False), ({'name': 'ollama', 'image': 'ollama/ollama'}, 'ollama', 'infrastructure', 'none', 'medium', False), ({'name': 'something-random', 'image': 'local/random'}, 'unknown', 'unknown', 'none', 'low', False), ] for raw, app_type, category, connector, criticality, protected in cases: result = classify_container(raw) assert result['app_type'] == app_type assert result['category'] == category assert result['connector'] == connector assert result['criticality'] == criticality assert result['protected'] is protected unknown = classify_container({'name': 'something-random', 'image': 'local/random'}) assert unknown['import_status'] == 'unmapped' assert unknown['confidence'] < 0.5 def test_unknown_canonical_apps_receive_conservative_classification_upgrade(api_client): from sqlmodel import Session from backend.app.db.init_db import _upgrade_unknown_app_classifications from backend.app.db.models import App, AppContainerMapping, DiscoveredContainer from backend.app.db.session import engine with Session(engine) as session: session.add(App(id='vaultwarden-local', name='Vaultwarden', category='Unknown')) session.add(App(id='custom-local', name='example-host Custom Tool', category='Unknown')) session.add(App(id='relay-local', name='Mail helper', category='Unknown')) session.add(AppContainerMapping(id='map-relay', app_id='relay-local', container_id='relay-container', container_name='mail-helper')) session.add(DiscoveredContainer( id='disc-relay', discovery_run_id='run-relay', container_id='relay-container', name='mail-helper', image='mwader/postfix-relay:latest', )) session.commit() assert _upgrade_unknown_app_classifications(session) is True session.commit() vaultwarden = session.get(App, 'vaultwarden-local') custom = session.get(App, 'custom-local') relay = session.get(App, 'relay-local') assert vaultwarden is not None assert vaultwarden.category == 'Auth' assert vaultwarden.criticality == 'critical' assert vaultwarden.protected is True assert custom is not None assert custom.category == 'Unknown' assert relay is not None assert relay.category == 'Network' assert relay.protected is True def test_active_app_views_exclude_deployment_history(api_client): from sqlmodel import Session from backend.app.db.models import App from backend.app.db.session import engine with Session(engine) as session: session.add(App(id='unraid_appops_gateway', name='Unraid AppOps Gateway', category='Infrastructure')) session.add(App(id='unraid_appops_gateway_1216', name='Unraid AppOps Gateway 1216', category='Infrastructure')) session.add(App( id='unraid_appops_gateway_1216_pre_20260712t203550z', name='Unraid AppOps Gateway 1216 pre deployment', category='Infrastructure', )) session.commit() app_ids = {item['id'] for item in api_client.get('/api/apps').json()} assert 'unraid_appops_gateway_1216' in app_ids assert 'unraid_appops_gateway' not in app_ids assert 'unraid_appops_gateway_1216_pre_20260712t203550z' not in app_ids inventory_ids = { item['id'] for item in api_client.get('/api/inventory/apps?include_unmanaged=false').json() } assert 'unraid_appops_gateway_1216' in inventory_ids assert 'unraid_appops_gateway' not in inventory_ids assert 'unraid_appops_gateway_1216_pre_20260712t203550z' not in inventory_ids def test_canonical_app_metadata_update_is_audited_and_metadata_only(api_client): before = api_client.get('/api/apps/sonarr').json() response = api_client.patch('/api/apps/sonarr/metadata', json={ 'name': 'Sonarr Media', 'category': 'Media Automation', 'criticality': 'high', }) assert response.status_code == 200 payload = response.json() assert payload['success'] is True assert payload['app']['name'] == 'Sonarr Media' assert payload['app']['category'] == 'Media Automation' assert payload['app']['criticality'] == 'high' app = api_client.get('/api/apps/sonarr').json() assert app['name'] == 'Sonarr Media' assert app['category'] == 'Media Automation' assert app['connectorStatus'] == before['connectorStatus'] assert app['voiceEnabled'] == before['voiceEnabled'] assert app['protected'] == before['protected'] audit = api_client.get('/api/audit').json() event = next(item for item in audit if item['action_id'] == 'app.metadata.update') assert event['target'] == 'sonarr' assert event['result'] == 'success' def test_canonical_app_metadata_update_rejects_invalid_values(api_client): response = api_client.patch('/api/apps/sonarr/metadata', json={ 'name': '', 'category': '', 'criticality': 'extreme', }) assert response.status_code == 200 assert response.json()['success'] is False assert response.json()['status'] == 'validation_error' def test_canonical_app_uses_live_published_port_for_external_url(api_client): from datetime import datetime, timezone from sqlmodel import Session from backend.app.core.config import settings from backend.app.db.models import App, AppContainerMapping, DiscoveredContainer, DiscoveryRun from backend.app.db.session import engine from backend.app.services.inventory_discovery import invalidate_discovery_cache previous_url = settings.apis.get('UNRAID_URL') settings.apis['UNRAID_URL'] = 'http://198.18.10.150' try: with Session(engine) as session: now = datetime.now(timezone.utc) run = DiscoveryRun(id='runtime-link-run', status='complete', started_at=now, finished_at=now) session.add(run) session.add(App(id='runtime_link_app', name='Runtime Link App', category='Dev')) session.add(DiscoveredContainer( id='runtime-link-discovered', discovery_run_id=run.id, container_id='runtime-link-container', name='runtime-link-app', image='runtime-link:test', state='running', status='healthy', ports_json=['4321:8080'], import_status='managed', matched_app_id='runtime_link_app', )) session.add(AppContainerMapping( id='runtime-link-mapping', app_id='runtime_link_app', container_id='runtime-link-container', container_name='runtime-link-app', )) session.commit() invalidate_discovery_cache() payload = api_client.get('/api/apps/runtime_link_app').json() assert payload['url'] == 'http://198.18.10.150' assert payload['port'] == 4321 assert 'api_key' not in str(payload).lower() assert 'token' not in str(payload).lower() finally: settings.apis['UNRAID_URL'] = previous_url def test_app_runtime_urls_strip_credentials_query_and_fragment(): from backend.app.api.routes_apps import _credential_free_url assert _credential_free_url('https://operator:secret@example.test:8443/app/?token=hidden#panel') == 'https://example.test:8443/app' assert _credential_free_url('javascript:alert(1)') is None assert _credential_free_url('http://example.test:99999') is None def test_category_labels_preserve_operational_acronyms(): from backend.app.services.app_registry import category_label assert category_label('ai') == 'AI' assert category_label('it_tools') == 'IT Tools' assert category_label('api_services') == 'API Services' def test_discovery_missing_config_returns_failure(api_client): response = api_client.post('/api/inventory/discover') assert response.status_code == 200 payload = response.json() assert payload['success'] is False assert payload['status'] == 'missing_config' assert payload['containersFound'] == 0 assert payload['runId'].startswith('disc-') def test_discovery_redacts_secret_metadata_and_never_stores_env(api_client, monkeypatch): secret = 'SUPER_SECRET_TOKEN' def fake_test_connection(self): return {'status': 'connected'} def fake_list_containers(self): return { 'status': 'ok', 'containers': [ { 'id': 'abc123', 'name': 'binhex-sonarr', 'image': 'binhex/arch-sonarr:latest', 'state': 'running', 'status': 'running', 'ports': ['8989:8989'], 'mounts': [{'Source': '/srv/example/appdata/sonarr', 'Destination': '/config'}], 'networks': ['bridge'], 'labels': {'com.example.token': secret, 'safe': 'value'}, 'env': {'API_KEY': secret}, 'raw': {'Env': [f'API_KEY={secret}'], 'Labels': {'secret': secret}}, } ], } monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', fake_test_connection, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', fake_list_containers, raising=True) response = api_client.post('/api/inventory/discover') payload = response.json() assert response.status_code == 200 assert payload['success'] is True assert payload['containersFound'] == 1 assert secret not in str(payload) discovered = api_client.get('/api/inventory/discovered-containers').json() assert len(discovered) == 1 assert secret not in str(discovered) assert 'Env' not in str(discovered) assert discovered[0]['suggestedAppType'] == 'sonarr' assert discovered[0]['importStatus'] == 'discovered' def test_import_discovered_container_creates_app_mapping_and_safe_defaults(api_client, monkeypatch): def fake_test_connection(self): return {'status': 'connected'} def fake_list_containers(self): return { 'status': 'ok', 'containers': [ { 'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'binhex/arch-sonarr:latest', 'state': 'running', 'status': 'running', 'ports': ['8989:8989'], }, { 'id': 'auth-container', 'name': 'authentik-server', 'image': 'ghcr.io/goauthentik/server:latest', 'state': 'running', 'status': 'running', 'ports': ['9000:9000'], }, ], } monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', fake_test_connection, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', fake_list_containers, raising=True) api_client.post('/api/inventory/discover') suggestions = api_client.get('/api/inventory/suggestions').json() sonarr = next(item for item in suggestions if item['suggestedAppType'] == 'sonarr') authentik = next(item for item in suggestions if item['suggestedAppType'] == 'authentik') sonarr_import = api_client.post('/api/inventory/import', json={'discoveredContainerId': sonarr['id']}).json() auth_import = api_client.post('/api/inventory/import', json={'discoveredContainerId': authentik['id']}).json() assert sonarr_import['success'] is True assert sonarr_import['app']['id'] == 'sonarr' assert sonarr_import['app']['voiceEnabled'] is False assert auth_import['app']['protected'] is True assert auth_import['app']['voiceEnabled'] is False mappings = api_client.get('/api/inventory/mappings').json() assert any(mapping['appId'] == 'sonarr' and mapping['containerId'] == 'sonarr-container' for mapping in mappings) assert any(mapping['appId'] == 'authentik' and mapping['containerId'] == 'auth-container' for mapping in mappings) policy = api_client.post( '/api/policies/evaluate', json={'action': 'container.stop', 'target': 'authentik', 'source': 'voice'}, ).json() assert policy['decision'] in {'blocked', 'approval_required'} def test_complete_inventory_review_marks_managed_apps_reviewed(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'binhex/arch-sonarr', 'state': 'running', 'status': 'running'}]}, raising=True, ) api_client.post('/api/inventory/discover') suggestion = api_client.get('/api/inventory/suggestions').json()[0] api_client.post('/api/inventory/import', json={'discoveredContainerId': suggestion['id']}) before = api_client.get('/api/setup/status').json() assert before['summary']['appsNeedReview'] == 1 payload = api_client.post('/api/inventory/review/complete', json={}).json() assert payload['success'] is True assert payload['reviewed'] == 1 after = api_client.get('/api/setup/status').json() assert after['summary']['appsNeedReview'] == 0 assert {step['id']: step['status'] for step in after['steps']}['review'] == 'complete' def test_inventory_ignore_survives_recreated_container_id(api_client, monkeypatch): current_id = {'value': 'custom-container-1'} monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': current_id['value'], 'name': 'release-candidate-stopped', 'image': 'local/candidate:latest', 'state': 'exited', 'status': 'exited'}]}, raising=True, ) api_client.post('/api/inventory/discover') first = api_client.get('/api/inventory/discovered-containers').json()[0] api_client.post('/api/inventory/ignore', json={'discoveredContainerId': first['id']}) current_id['value'] = 'custom-container-2' api_client.post('/api/inventory/discover') latest = next(item for item in api_client.get('/api/inventory/discovered-containers').json() if item['containerId'] == 'custom-container-2') assert latest['importStatus'] == 'ignored' assert api_client.get('/api/setup/status').json()['summary']['appsNeedReview'] == 0 def test_inventory_ignore_survives_rediscovery_for_mapped_container(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'mapped-migrate-1', 'name': 'mapped-migrate', 'image': 'local/migrate:latest', 'state': 'exited', 'status': 'unhealthy'}]}, raising=True, ) api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/discovered-containers').json()[0] imported = api_client.post('/api/inventory/import', json={'discoveredContainerId': discovered['id']}).json() assert imported['success'] is True latest = api_client.get('/api/inventory/discovered-containers').json()[0] api_client.post('/api/inventory/ignore', json={'discoveredContainerId': latest['id']}) api_client.post('/api/inventory/discover') refreshed = api_client.get('/api/inventory/discovered-containers').json()[0] assert refreshed['importStatus'] == 'ignored' app = next(item for item in api_client.get('/api/apps').json() if item['id'] == imported['app']['id']) assert app['status'] == 'healthy' assert app['containerStatus'] == 'ignored' def test_inventory_auto_ignores_ephemeral_ci_and_buildkit_containers(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: { 'status': 'ok', 'containers': [ {'id': 'ci-1', 'name': 'GITEA-ACTIONS-TASK-42-WORKFLOW-validation-JOB-browser-abc', 'image': 'docker.gitea.com/runner-images:ubuntu-latest', 'state': 'running', 'status': 'running'}, {'id': 'buildkit-1', 'name': 'buildx_buildkit_builder-example', 'image': 'moby/buildkit:buildx-stable-1', 'state': 'running', 'status': 'running'}, {'id': 'phase-1', 'name': 'beautiful_colden', 'image': 'devrunbook-phase12b-migrate', 'state': 'running', 'status': 'running'}, {'id': 'candidate-1', 'name': 'ludarium-candidate-039-postgres-1', 'image': 'postgres:16-alpine', 'state': 'running', 'status': 'running'}, {'id': 'dotnet-sdk-1', 'name': 'friendly_matsumoto', 'image': 'mcr.microsoft.com/dotnet/sdk:10.0.300', 'state': 'running', 'status': 'running'}, ], }, raising=True, ) api_client.post('/api/inventory/discover') latest = api_client.get('/api/inventory/discovered-containers').json() assert {item['importStatus'] for item in latest} == {'ignored'} assert api_client.get('/api/setup/status').json()['summary']['appsNeedReview'] == 0 def test_setup_managed_app_count_matches_canonical_apps_not_mappings(api_client): apps = api_client.get('/api/apps').json() assert len(apps) > 0 mapping_response = api_client.patch( f"/api/apps/{apps[0]['id']}/container-mapping", json={'containerId': 'secondary-container-for-same-app', 'containerName': 'secondary-container-for-same-app'}, ).json() assert mapping_response['success'] is True status = api_client.get('/api/setup/status').json() assert status['summary']['appsManaged'] == len(api_client.get('/api/apps').json()) assert status['managedApps'] == status['summary']['appsManaged'] def test_update_discovered_review_metadata_updates_only_appops_classification(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'custom-container', 'name': 'custom-app', 'image': 'local/custom:latest', 'state': 'running', 'status': 'running'}]}, raising=True, ) api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/discovered-containers').json()[0] payload = api_client.post( '/api/inventory/review/update', json={ 'discoveredContainerId': discovered['id'], 'displayName': 'Custom Ops App', 'category': 'ops tooling', 'criticality': 'high', 'connectorType': 'none', 'policyProfile': 'review_only', }, ).json() assert payload['success'] is True updated = api_client.get('/api/inventory/discovered-containers').json()[0] assert updated['displayName'] == 'Custom Ops App' assert updated['suggestedCategory'] == 'ops_tooling' assert updated['suggestedCriticality'] == 'high' assert updated['suggestedConnector'] == 'none' assert updated['suggestedPolicyProfile'] == 'review_only' assert updated['importStatus'] == 'unmapped' assert api_client.get('/api/inventory/mappings').json() == [] def test_map_discovered_container_to_existing_app_creates_manual_mapping_without_new_app(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'custom-container', 'name': 'custom-app', 'image': 'local/custom:latest', 'state': 'running', 'status': 'running'}]}, raising=True, ) api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/discovered-containers').json()[0] payload = api_client.post('/api/inventory/map-existing', json={'discoveredContainerId': discovered['id'], 'appId': 'widefrog'}).json() assert payload['success'] is True assert payload['app']['id'] == 'widefrog' assert payload['app']['voiceEnabled'] is False assert payload['mapping']['appId'] == 'widefrog' assert payload['mapping']['containerId'] == 'custom-container' assert payload['mapping']['source'] == 'manual_review' mappings = api_client.get('/api/inventory/mappings').json() assert any(mapping['appId'] == 'widefrog' and mapping['containerId'] == 'custom-container' for mapping in mappings) assert all(mapping['appId'] != 'unknown' for mapping in mappings) updated = api_client.get('/api/inventory/discovered-containers').json()[0] assert updated['importStatus'] == 'managed' assert updated['matchedAppId'] == 'widefrog' def test_discovery_lists_only_latest_container_records(api_client, monkeypatch): first = {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}]} second = {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}]} calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) api_client.post('/api/inventory/discover') api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/discovered-containers').json() setup = api_client.get('/api/setup/status').json() assert len(discovered) == 1 assert discovered[0]['containerId'] == 'sonarr-container' assert setup['summary']['containersDiscovered'] == 1 def test_latest_inventory_lookup_uses_a_single_window_query(api_client): from datetime import datetime from sqlalchemy import event from sqlmodel import Session from backend.app.db.models import DiscoveredContainer, DiscoveryRun from backend.app.db.session import engine from backend.app.services.inventory_discovery import _latest_by_container, invalidate_discovery_cache with Session(engine) as session: for index in range(3): run = DiscoveryRun(id=f'window-run-{index}', status='success', started_at=datetime(2026, 7, 13, 8, index)) session.add(run) session.add(DiscoveredContainer( id=f'window-row-{index}', discovery_run_id=run.id, container_id='window-container', name='window-container', image=f'window:{index}', )) session.commit() invalidate_discovery_cache() statements: list[str] = [] def capture_statement(_connection, _cursor, statement, _parameters, _context, _executemany): if statement.lstrip().upper().startswith('SELECT'): statements.append(statement) event.listen(engine, 'before_cursor_execute', capture_statement) try: latest = _latest_by_container(session) finally: event.remove(engine, 'before_cursor_execute', capture_statement) assert latest['window-container'].image == 'window:2' assert len(statements) == 1 assert 'row_number() OVER' in statements[0] def test_inventory_runs_are_paginated_and_bounded(api_client): from datetime import datetime from sqlmodel import Session from backend.app.db.models import DiscoveryRun from backend.app.db.session import engine with Session(engine) as session: for index in range(6): session.add(DiscoveryRun( id=f'page-run-{index}', status='success', started_at=datetime(2026, 8, 21, 12, index), )) session.commit() first = api_client.get('/api/inventory/runs?limit=2&offset=0') second = api_client.get('/api/inventory/runs?limit=2&offset=2') invalid = api_client.get('/api/inventory/runs?limit=501') assert first.status_code == 200 assert second.status_code == 200 assert invalid.status_code == 422 assert len(first.json()) == 2 assert len(second.json()) == 2 assert {item['id'] for item in first.json()}.isdisjoint({item['id'] for item in second.json()}) def test_discovery_retention_removes_old_run_summaries(api_client): from datetime import datetime from sqlmodel import Session, select from backend.app.db.models import DiscoveryRun from backend.app.db.session import engine from backend.app.services.inventory_discovery import prune_discovery_history with Session(engine) as session: for index in range(8): session.add(DiscoveryRun( id=f'retention-run-{index}', status='failed', started_at=datetime(2026, 8, 21, 13, index), )) session.commit() result = prune_discovery_history(session, 3) retained = session.exec( select(DiscoveryRun).where(DiscoveryRun.id.startswith('retention-run-')) ).all() assert result['runsRemoved'] >= 5 assert {run.id for run in retained} == {'retention-run-5', 'retention-run-6', 'retention-run-7'} def test_discovery_hides_appops_deploy_artifacts_from_operator_lists(api_client, monkeypatch): containers = [ {'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}, {'id': 'appops-legacy', 'name': 'unraid-appops-gateway', 'image': 'unraid-appops-gateway:legacy', 'state': 'exited', 'status': 'exited'}, {'id': 'appops-current', 'name': 'unraid-appops-gateway-1216', 'image': 'unraid-appops-gateway:current', 'state': 'running', 'status': 'running'}, {'id': 'appops-pre', 'name': 'unraid-appops-gateway-1216-pre-20260628T005759Z', 'image': 'unraid-appops-gateway:old', 'state': 'exited', 'status': 'exited'}, ] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': containers}, raising=True) api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/discovered-containers').json() inventory_containers = api_client.get('/api/inventory/containers').json() apps = api_client.get('/api/inventory/apps?include_unmanaged=true').json() suggestions = api_client.get('/api/inventory/suggestions').json() run_detail = api_client.get('/api/inventory/runs').json()[0] historical = api_client.get(f"/api/inventory/runs/{run_detail['id']}").json()['containers'] for collection in [discovered, inventory_containers, suggestions]: names = {item['name'] for item in collection} assert 'binhex-sonarr' in names assert 'unraid-appops-gateway' not in names assert 'unraid-appops-gateway-1216' not in names assert 'unraid-appops-gateway-1216-pre-20260628T005759Z' not in names app_names = {item['name'] for item in apps} assert 'Sonarr' in app_names assert 'unraid-appops-gateway' not in app_names assert 'unraid-appops-gateway-1216' not in app_names assert 'unraid-appops-gateway-1216-pre-20260628T005759Z' not in app_names assert {item['name'] for item in historical} >= { 'unraid-appops-gateway', 'unraid-appops-gateway-1216', 'unraid-appops-gateway-1216-pre-20260628T005759Z', } def test_inventory_review_coalesces_recreated_container_ids_by_unique_name(api_client, monkeypatch): snapshots = [ {'status': 'ok', 'containers': [{'id': 'sonarr-old-id', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}]}, {'status': 'ok', 'containers': [{'id': 'sonarr-new-id', 'name': 'binhex-sonarr', 'image': 'sonarr:v2', 'state': 'running', 'status': 'running'}]}, ] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: snapshots.pop(0), raising=True) api_client.post('/api/inventory/discover') api_client.post('/api/inventory/discover') review = api_client.get('/api/inventory/discovered-containers').json() inventory_apps = api_client.get('/api/inventory/apps?include_unmanaged=true').json() runs = api_client.get('/api/inventory/runs').json() latest_run = api_client.get(f"/api/inventory/runs/{runs[0]['id']}").json() assert len(review) == 1 assert review[0]['containerId'] == 'sonarr-new-id' assert review[0]['image'] == 'sonarr:v2' assert len([item for item in inventory_apps if item['name'] == 'Sonarr' and item['id'].startswith('discovered-')]) == 1 assert any(item['containerId'] == 'sonarr-new-id' for item in latest_run['containers']) def test_discovery_reconciles_same_name_mapping_after_container_recreation(api_client, monkeypatch): snapshots = [ {'status': 'ok', 'containers': [{'id': 'sonarr-old-id', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}]}, {'status': 'ok', 'containers': [{'id': 'sonarr-new-id', 'name': 'binhex-sonarr', 'image': 'sonarr:v2', 'state': 'running', 'status': 'running'}]}, ] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: snapshots.pop(0), raising=True) api_client.post('/api/inventory/discover') suggestion = api_client.get('/api/inventory/suggestions').json()[0] api_client.post('/api/inventory/import', json={'discoveredContainerId': suggestion['id']}) api_client.post('/api/inventory/review/complete', json={}) result = api_client.post('/api/inventory/discover').json() mappings = api_client.get('/api/inventory/mappings').json() setup = api_client.get('/api/setup/status').json() audit = api_client.get('/api/audit').json() assert result['reconciledMappings'] == 1 assert result['duplicateMappingsRemoved'] == 0 assert len([item for item in mappings if item['containerName'] == 'binhex-sonarr']) == 1 assert any(item['containerId'] == 'sonarr-new-id' and item['appId'] == 'sonarr' for item in mappings) assert setup['summary']['appsNeedReview'] == 0 assert any(item['action_id'] == 'inventory.mapping.reconcile' for item in audit) def test_discovery_ignores_port_mount_and_network_order_changes(api_client, monkeypatch): first = { 'status': 'ok', 'containers': [{ 'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running', 'ports': ['8989:8989', '9898:9898'], 'mounts': [ {'source': '/config', 'destination': '/config', 'type': 'bind'}, {'source': '/media', 'destination': '/media', 'type': 'bind'}, ], 'networks': ['bridge', 'media'], }], } second = { 'status': 'ok', 'containers': [{ **first['containers'][0], 'ports': list(reversed(first['containers'][0]['ports'])), 'mounts': list(reversed(first['containers'][0]['mounts'])), 'networks': list(reversed(first['containers'][0]['networks'])), }], } snapshots = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: snapshots.pop(0), raising=True) api_client.post('/api/inventory/discover') result = api_client.post('/api/inventory/discover').json() current = api_client.get('/api/inventory/discovered-containers').json()[0] assert result['changedItems'] == 0 assert current['importStatus'] == 'discovered' def test_complete_inventory_review_marks_mapped_drift_reviewed(api_client, monkeypatch): first = {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}]} second = {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v2', 'state': 'running', 'status': 'running'}]} calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) api_client.post('/api/inventory/discover') suggestion = api_client.get('/api/inventory/suggestions').json()[0] api_client.post('/api/inventory/import', json={'discoveredContainerId': suggestion['id']}) api_client.post('/api/inventory/review/complete', json={}) api_client.post('/api/inventory/discover') before = api_client.get('/api/setup/status').json() changed = api_client.get('/api/inventory/discovered-containers').json()[0] assert changed['importStatus'] == 'changed' assert before['summary']['appsNeedReview'] == 0 payload = api_client.post('/api/inventory/review/complete', json={}).json() after = api_client.get('/api/inventory/discovered-containers').json()[0] assert payload['success'] is True assert payload['driftReviewed'] == 1 assert after['importStatus'] == 'managed' def test_inventory_drift_acknowledge_and_create_incident_are_metadata_only(api_client, monkeypatch): runs = [ {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v1', 'state': 'running', 'status': 'running'}]}, {'status': 'ok', 'containers': [{'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'sonarr:v2', 'state': 'running', 'status': 'running'}]}, ] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: runs.pop(0), raising=True) api_client.post('/api/inventory/discover') api_client.post('/api/inventory/discover') drift = api_client.get('/api/inventory/drift').json() image_drift = next(item for item in drift if item['type'] == 'image_changed') assert image_drift['driftId'].startswith('drift-') assert image_drift['status'] == 'open' incident = api_client.post(f"/api/inventory/drift/{image_drift['driftId']}/incident").json() assert incident['success'] is True assert incident['incident']['id'] == f"incident-{image_drift['driftId']}" incidents = api_client.get('/api/incidents').json() assert any(item['id'] == incident['incident']['id'] for item in incidents) status_update = api_client.post(f"/api/incidents/{incident['incident']['id']}/status", json={'status': 'resolved'}).json() assert status_update['success'] is True assert all(item['id'] != incident['incident']['id'] for item in api_client.get('/api/incidents').json()) acknowledged = api_client.post(f"/api/inventory/drift/{image_drift['driftId']}/acknowledge").json() assert acknowledged['success'] is True assert all(item['driftId'] != image_drift['driftId'] for item in api_client.get('/api/inventory/drift').json()) def test_discovery_first_inventory_endpoints_show_known_and_unknown(api_client, monkeypatch): def fake_test_connection(self): return {'status': 'connected'} def fake_list_containers(self): return { 'status': 'ok', 'containers': [ {'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'binhex/arch-sonarr', 'state': 'running', 'status': 'running', 'ports': ['8989:8989']}, {'id': 'custom-container', 'name': 'bespoke-worker', 'image': 'local/bespoke', 'state': 'running', 'status': 'running', 'ports': ['4567:4567']}, ], } monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', fake_test_connection, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', fake_list_containers, raising=True) api_client.post('/api/inventory/discover') containers = api_client.get('/api/inventory/containers').json() apps = api_client.get('/api/inventory/apps?include_unmanaged=true').json() app_cards = api_client.get('/api/apps?include_discovered=true').json() unknown = api_client.get('/api/inventory/unknown').json() unmapped = api_client.get('/api/inventory/unmapped').json() assert {item['containerId'] for item in containers} == {'sonarr-container', 'custom-container'} custom_container = next(item for item in containers if item['containerId'] == 'custom-container') assert custom_container['appType'] == 'unknown' assert custom_container['policyProfile'] == 'generic_unknown' assert custom_container['connectorType'] == 'none' assert 'container.logs' in custom_container['actionsAvailable'] assert 'container.restart' in custom_container['actionsAvailable'] sonarr_app = next(item for item in apps if item['primary_container'] == 'binhex-sonarr') custom_app = next(item for item in apps if item['primary_container'] == 'bespoke-worker') assert sonarr_app['connector_type'] == 'sonarr' assert custom_app['app_type'] == 'unknown' assert custom_app['connector_type'] == 'none' assert custom_app['lifecycle_state'] == 'unmapped' assert custom_app['source'] == 'portainer' assert any(item['id'].startswith('discovered-') and item['app_type'] == 'unknown' for item in app_cards) assert [item['containerId'] for item in unknown] == ['custom-container'] assert [item['containerId'] for item in unmapped] == ['custom-container'] def test_connector_candidates_are_filtered_and_secret_safe(api_client, monkeypatch): def fake_test_connection(self): return {'status': 'connected'} def fake_list_containers(self): return { 'status': 'ok', 'containers': [ { 'id': 'sonarr-container', 'name': 'binhex-sonarr', 'image': 'binhex/arch-sonarr', 'state': 'running', 'status': 'running', 'ports': ['8989:8989'], 'env': {'SONARR_API_KEY': 'super-secret'}, }, {'id': 'custom-container', 'name': 'bespoke-worker', 'image': 'local/bespoke', 'state': 'running', 'status': 'running'}, ], } monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', fake_test_connection, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', fake_list_containers, raising=True) api_client.post('/api/inventory/discover') candidates = api_client.get('/api/connectors/sonarr/candidates').json() assert len(candidates) == 1 assert candidates[0]['connectorId'] == 'sonarr' assert candidates[0]['discoveredContainerId'].startswith('dc-') assert candidates[0]['containerId'] == 'sonarr-container' assert candidates[0]['appType'] == 'sonarr' assert candidates[0]['suggestedAppId'] == 'sonarr' assert candidates[0]['connectorStatus'] == 'missing_config' assert candidates[0]['missingSecrets'] == ['SONARR_URL', 'SONARR_API_KEY'] assert 'super-secret' not in str(candidates) assert 'custom-container' not in str(candidates) def test_connector_bind_and_unbind_keep_safe_defaults(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'radarr-container', 'name': 'binhex-radarr', 'image': 'binhex/arch-radarr', 'state': 'running', 'status': 'running', 'ports': ['7878:7878']}]}, raising=True, ) api_client.post('/api/inventory/discover') candidate = api_client.get('/api/connectors/radarr/candidates').json()[0] bound = api_client.post('/api/connectors/radarr/bind', json={'discoveredContainerId': candidate['discoveredContainerId']}).json() assert bound['success'] is True assert bound['message'] == 'connector_bound' assert bound['app']['id'] == 'radarr' assert bound['app']['voiceEnabled'] is False assert bound['app']['needsReview'] is True assert bound['binding']['connectorId'] == 'radarr' assert bound['binding']['containerId'] == 'radarr-container' bindings = api_client.get('/api/connectors/radarr/bindings').json() assert len(bindings) == 1 assert bindings[0]['appId'] == 'radarr' assert bindings[0]['connectorStatus'] == 'missing_config' from backend.app.services import inventory_discovery latest_calls = 0 original_latest = inventory_discovery._latest_by_container def counted_latest(session): nonlocal latest_calls latest_calls += 1 return original_latest(session) monkeypatch.setattr(inventory_discovery, '_latest_by_container', counted_latest) api_client.get('/api/connectors/radarr/bindings') assert latest_calls == 1 unbound = api_client.post('/api/connectors/radarr/unbind', json={'appId': 'radarr'}).json() assert unbound['success'] is True assert unbound['message'] == 'connector_unbound' assert unbound['app']['connectorStatus'] == 'disconnected' assert unbound['app']['voiceEnabled'] is False assert unbound['app']['needsReview'] is True mappings = api_client.get('/api/inventory/mappings').json() assert any(mapping['appId'] == 'radarr' and mapping['containerId'] == 'radarr-container' for mapping in mappings) def test_discovered_synthetic_app_detail_is_ui_ready(api_client, monkeypatch): monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'x1', 'name': 'unknown-tool', 'image': 'local/tool', 'state': 'running', 'status': 'running', 'ports': ['8080:8080'], 'mounts': [{'Destination': '/config'}], 'networks': ['bridge']}]}, raising=True, ) api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/unknown').json()[0] detail = api_client.get(f"/api/apps/discovered-{discovered['id']}").json() assert detail['id'] == f"discovered-{discovered['id']}" assert detail['app_type'] == 'unknown' assert detail['lifecycle_state'] == 'unmapped' assert detail['primary_container'] == 'unknown-tool' assert detail['connector_type'] == 'none' assert detail['voiceEnabled'] is False assert 'Confirm app type' in detail['setup_checklist'] assert detail['discovery']['containerId'] == 'x1' def test_ignore_marks_container_without_deleting(api_client, monkeypatch): def fake_test_connection(self): return {'status': 'connected'} def fake_list_containers(self): return {'status': 'ok', 'containers': [{'id': 'x1', 'name': 'unknown-tool', 'image': 'local/tool'}]} monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', fake_test_connection, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', fake_list_containers, raising=True) api_client.post('/api/inventory/discover') discovered = api_client.get('/api/inventory/discovered-containers').json()[0] response = api_client.post('/api/inventory/ignore', json={'discoveredContainerId': discovered['id']}) assert response.status_code == 200 assert response.json()['success'] is True updated = api_client.get('/api/inventory/discovered-containers').json()[0] assert updated['importStatus'] == 'ignored' def test_drift_detects_image_port_missing_and_new(api_client, monkeypatch): from backend.app.services.inventory_discovery import run_discovery first = { 'status': 'ok', 'containers': [ {'id': 'sonarr-container', 'name': 'sonarr', 'image': 'sonarr:v1', 'ports': ['8989:8989'], 'state': 'running'}, {'id': 'radarr-container', 'name': 'radarr', 'image': 'radarr:v1', 'ports': ['7878:7878'], 'state': 'running'}, ], } second = { 'status': 'ok', 'containers': [ {'id': 'sonarr-container', 'name': 'sonarr', 'image': 'sonarr:v2', 'ports': ['8990:8989'], 'state': 'running'}, {'id': 'prowlarr-container', 'name': 'prowlarr', 'image': 'prowlarr:v1', 'ports': ['9696:9696'], 'state': 'running'}, ], } calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) run_discovery() run_discovery() drift = api_client.get('/api/inventory/drift').json() types = {item['type'] for item in drift} assert {'image_changed', 'port_changed', 'missing_container', 'new_container'}.issubset(types) def test_discovery_does_not_repeat_missing_rows(api_client, monkeypatch): from sqlmodel import Session, select from backend.app.db.models import DiscoveredContainer from backend.app.db.session import engine from backend.app.services.inventory_discovery import run_discovery calls = [ {'status': 'ok', 'containers': [{'id': 'gone', 'name': 'gone-app', 'image': 'local/gone', 'state': 'running'}]}, {'status': 'ok', 'containers': []}, {'status': 'ok', 'containers': []}, ] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) run_discovery() run_discovery() run_discovery() with Session(engine) as session: rows = session.exec(select(DiscoveredContainer).where(DiscoveredContainer.container_id == 'gone')).all() assert len(rows) == 2 assert sum(row.import_status == 'missing' for row in rows) == 1 def test_discovery_history_retention_keeps_recent_run_details(api_client, monkeypatch): from sqlmodel import Session, select from backend.app.db.models import DiscoveredContainer from backend.app.db.session import engine from backend.app.services.inventory_discovery import prune_discovery_history, run_discovery monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr( 'backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: {'status': 'ok', 'containers': [{'id': 'stable', 'name': 'stable-app', 'image': 'local/stable', 'state': 'running'}]}, raising=True, ) run_discovery() run_discovery() run_discovery() with Session(engine) as session: result = prune_discovery_history(session, keep_runs=2) rows = session.exec(select(DiscoveredContainer).where(DiscoveredContainer.container_id == 'stable')).all() assert result['removed'] == 1 assert result['retainedRuns'] == 2 assert len(rows) == 2 def test_inventory_baseline_is_persisted_audited_and_drives_drift(api_client, monkeypatch): runs = [ { 'status': 'ok', 'containers': [ {'id': 'sonarr-container', 'name': 'sonarr', 'image': 'sonarr:v1', 'ports': ['8989:8989'], 'state': 'running', 'status': 'running'}, {'id': 'radarr-container', 'name': 'radarr', 'image': 'radarr:v1', 'ports': ['7878:7878'], 'state': 'running', 'status': 'running'}, ], }, { 'status': 'ok', 'containers': [ {'id': 'sonarr-container', 'name': 'sonarr', 'image': 'sonarr:v2', 'ports': ['8990:8989'], 'state': 'running', 'status': 'running'}, {'id': 'prowlarr-container', 'name': 'prowlarr', 'image': 'prowlarr:v1', 'ports': ['9696:9696'], 'state': 'running', 'status': 'running'}, ], }, ] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: runs.pop(0), raising=True) api_client.post('/api/inventory/discover') baseline = api_client.post('/api/inventory/baseline/update').json() assert baseline['success'] is True assert baseline['data']['containerCount'] == 2 assert baseline['auditId'].startswith('audit-') assert api_client.get('/api/inventory/drift').json() == [] api_client.post('/api/inventory/discover') drift = api_client.get('/api/inventory/drift').json() assert {'image_changed', 'port_changed', 'missing_container', 'new_container'}.issubset({item['type'] for item in drift}) refreshed = api_client.post('/api/inventory/baseline/update').json() assert refreshed['success'] is True assert refreshed['data']['containerCount'] == 2 assert api_client.get('/api/inventory/drift').json() == [] audit = api_client.get('/api/audit').json() assert any(item['action_id'] == 'inventory.baseline.update' for item in audit) def test_inventory_baseline_requires_live_inventory(api_client): result = api_client.post('/api/inventory/baseline/update').json() assert result['success'] is False assert result['status'] == 'failed' assert 'discovery' in result['message'].lower() def test_drift_ignores_port_order_only_changes(api_client, monkeypatch): from backend.app.services.inventory_discovery import run_discovery first = { 'status': 'ok', 'containers': [ {'id': 'jdownloader', 'name': 'JDownloader2', 'image': 'jdownloader:v1', 'ports': ['7807:5800', '7907:5900'], 'state': 'running'}, ], } second = { 'status': 'ok', 'containers': [ {'id': 'jdownloader', 'name': 'JDownloader2', 'image': 'jdownloader:v1', 'ports': ['7907:5900', '7807:5800'], 'state': 'running'}, ], } calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) run_discovery() run_discovery() drift = api_client.get('/api/inventory/drift').json() assert not any(item['type'] == 'port_changed' for item in drift) def test_drift_suppresses_recreated_container_missing_new_pair(api_client, monkeypatch): from backend.app.services.inventory_discovery import run_discovery first = { 'status': 'ok', 'containers': [ {'id': 'infrabinder-old', 'name': 'infrabinder-local', 'image': 'infrabinder-local:latest', 'ports': ['1215:1215'], 'state': 'running'}, ], } second = { 'status': 'ok', 'containers': [ {'id': 'infrabinder-new', 'name': 'infrabinder-local', 'image': 'infrabinder-local:latest', 'ports': ['1215:1215'], 'state': 'running'}, ], } calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) run_discovery() run_discovery() drift = api_client.get('/api/inventory/drift').json() assert not any(item['type'] in {'missing_container', 'new_container'} for item in drift) def test_drift_ignores_upgrade_rollback_containers(api_client, monkeypatch): from backend.app.services.inventory_discovery import run_discovery first = { 'status': 'ok', 'containers': [ {'id': 'appops-rollback', 'name': 'unraid-appops-gateway-1216-pre-20260627T040601Z', 'image': 'appops:old', 'ports': ['1216:1216'], 'state': 'running', 'status': 'healthy'}, ], } second = { 'status': 'ok', 'containers': [ {'id': 'appops-rollback', 'name': 'unraid-appops-gateway-1216-pre-20260627T040601Z', 'image': 'appops:old', 'ports': [], 'state': 'exited', 'status': 'unhealthy'}, ], } calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) run_discovery() run_discovery() drift = api_client.get('/api/inventory/drift').json() assert drift == [] def test_drift_ignores_canonical_appops_container_redeploy(api_client, monkeypatch): from backend.app.services.inventory_discovery import run_discovery first = { 'status': 'ok', 'containers': [ {'id': 'appops-old', 'name': 'unraid-appops-gateway-1216', 'image': 'appops:old', 'ports': ['1216:1216'], 'state': 'running'}, ], } second = { 'status': 'ok', 'containers': [ {'id': 'appops-new', 'name': 'unraid-appops-gateway-1216', 'image': 'appops:new', 'ports': ['1216:1216'], 'state': 'running'}, ], } calls = [first, second] monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.test_connection', lambda self: {'status': 'connected'}, raising=True) monkeypatch.setattr('backend.app.connectors.portainer.PortainerConnector.list_containers', lambda self: calls.pop(0), raising=True) run_discovery() run_discovery() drift = api_client.get('/api/inventory/drift').json() assert drift == []