diff --git a/backend/app/api/setup_health.py b/backend/app/api/setup_health.py new file mode 100644 index 00000000..cfb543cb --- /dev/null +++ b/backend/app/api/setup_health.py @@ -0,0 +1,154 @@ +"""REST surface for Setup Health — "how set up is this panel". + +Mounted at /api/v1/setup-health (registered in app/__init__.py). Admin-only. + +Contract: + GET / -> {'items': [...], 'summary': {critical_open, recommended_open, + ok, total, score}} + +Each ``item`` is a doctor-style check dict (key/title/status/detail/ +repairable/repair_ref) plus the setup extras ``severity``/``scope``/``why``/ +``fix`` (fix is a deep-link ``{kind:'link', to}`` or a repair ``{kind:'repair', +ref}``). The doctor report carries the same items as its ``setup.*`` section, so +this endpoint and Monitoring → Doctor stay in lockstep. +""" +from flask import Blueprint, jsonify, request + +from ..middleware.rbac import admin_required, auth_required, get_current_user +from ..services.setup_health_service import SetupHealthService +from ..services.setup_reconcile_service import ( + DNS_BACKFILL_JOB_KIND, + URL_FIX_JOB_KIND, + SetupReconcileService, +) + +setup_health_bp = Blueprint('setup_health', __name__) + + +@setup_health_bp.route('', methods=['GET']) +@setup_health_bp.route('/', methods=['GET']) +@admin_required +def get_setup_health(): + """Evaluate the setup-health registry (panel scope) and return items + + summary. Cheap — DB/settings probes only, no host calls.""" + try: + return jsonify(SetupHealthService.evaluate(scope='panel')) + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +@setup_health_bp.route('/account', methods=['GET']) +@auth_required() +def get_account_security(): + """The requesting user's own setup-health items (scope 'user') — e.g. the + 'secure your account' nudge. Any authenticated user, about themselves.""" + try: + user = get_current_user() + if not user: + return jsonify({'error': 'User not found'}), 404 + return jsonify(SetupHealthService.evaluate(scope='user', user=user)) + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +# --------------------------------------------------------------------------- # +# Snooze / dismiss (Phase 6). Panel items require admin; personal (user-scope) +# items may be snoozed by the owning user. +# --------------------------------------------------------------------------- # + +def _snooze_guard(key): + """Resolve the requesting user and enforce admin for panel-scoped items. + Returns (user, None) when allowed, or (None, (json, status)) on error.""" + user = get_current_user() + if not user: + return None, (jsonify({'error': 'User not found'}), 404) + scope = SetupHealthService._item_scope(key) + if scope is None: + return None, (jsonify({'error': f'Unknown setup item: {key}'}), 400) + if scope == 'panel' and not user.is_admin: + return None, (jsonify({'error': 'Admin access required'}), 403) + return user, None + + +@setup_health_bp.route('/snooze', methods=['POST']) +@auth_required() +def snooze_item(): + """Snooze a setup item (mutes it; it still renders, just quietly).""" + data = request.get_json(silent=True) or {} + key = data.get('key') + if not key: + return jsonify({'error': "Body must carry an item 'key'."}), 400 + user, err = _snooze_guard(key) + if err: + return err + try: + return jsonify(SetupHealthService.snooze(key, days=data.get('days', 30), user=user)) + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +@setup_health_bp.route('/snooze', methods=['DELETE']) +@auth_required() +def unsnooze_item(): + """Clear a snooze so the item is active again.""" + data = request.get_json(silent=True) or {} + key = data.get('key') or request.args.get('key') + if not key: + return jsonify({'error': "An item 'key' is required."}), 400 + user, err = _snooze_guard(key) + if err: + return err + try: + return jsonify(SetupHealthService.unsnooze(key, user=user)) + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +# --------------------------------------------------------------------------- # +# Reconcile-on-connect (Phase 3) — preview is synchronous; apply enqueues the +# job so a large backfill/heal runs off the request thread. +# --------------------------------------------------------------------------- # + +@setup_health_bp.route('/reconcile/dns/preview', methods=['POST']) +@admin_required +def dns_backfill_preview(): + """Dry-run: the managed hosts a DNS backfill would ensure records for.""" + try: + return jsonify(SetupReconcileService.dns_backfill_preview()) + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +@setup_health_bp.route('/reconcile/dns/apply', methods=['POST']) +@admin_required +def dns_backfill_apply(): + """Enqueue the DNS backfill job (loops ensure_a_record, ownership-guarded).""" + try: + from app.jobs.service import JobService + job = JobService.enqueue(DNS_BACKFILL_JOB_KIND, payload={}, max_attempts=1) + return jsonify({'job_id': job.id}), 202 + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +@setup_health_bp.route('/reconcile/url-fix/preview', methods=['POST']) +@admin_required +def url_fix_preview(): + """Dry-run: WordPress sites whose siteurl/home still point at localhost or a + stale IP, with the per-site swap they'd get.""" + try: + return jsonify(SetupReconcileService.url_fix_preview()) + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 + + +@setup_health_bp.route('/reconcile/url-fix/apply', methods=['POST']) +@admin_required +def url_fix_apply(): + """Enqueue the WordPress URL-fix job (loops change_site_url).""" + try: + from app.jobs.service import JobService + job = JobService.enqueue(URL_FIX_JOB_KIND, payload={}, max_attempts=1) + return jsonify({'job_id': job.id}), 202 + except Exception as exc: # noqa: BLE001 + return jsonify({'error': str(exc)}), 500 diff --git a/backend/tests/test_setup_health.py b/backend/tests/test_setup_health.py new file mode 100644 index 00000000..c2eeab78 --- /dev/null +++ b/backend/tests/test_setup_health.py @@ -0,0 +1,271 @@ +"""Tests for the Setup Health registry (setup_health_service): the probe matrix +over settings fixtures, the critical-vs-recommended severity mapping (HTTPS is +never critical), and inclusion of the section in the doctor report + API.""" +import pytest + +from app.services.setup_health_service import SetupHealthService + + +# --------------------------------------------------------------------------- # +# Fixtures / helpers +# --------------------------------------------------------------------------- # + +def set_setting(key, value, vtype='string'): + from app import db + from app.models.system_settings import SystemSettings + SystemSettings.set(key=key, value=value, value_type=vtype) + db.session.commit() + + +def add_dns_provider(): + from app import db + from app.models.email import DNSProviderConfig + db.session.add(DNSProviderConfig(name='CF', provider='cloudflare', api_key='x')) + db.session.commit() + + +def add_email_provider(tested_ok=True): + from app import db + from app.models.email_provider import EmailProviderConnection + row = EmailProviderConnection(name='SMTP', provider='smtp', is_default=True, + is_active=True, uses_notifications=True, + last_test_ok=tested_ok) + db.session.add(row) + db.session.commit() + return row + + +def add_backup_policy(enabled=True): + from app import db + from app.models.backup_policy import BackupPolicy + row = BackupPolicy(target_type='application', target_id=1, enabled=enabled, + schedule_cron='0 3 * * *') + db.session.add(row) + db.session.commit() + return row + + +def items_by_key(result): + return {c['key']: c for c in result['items']} + + +# --------------------------------------------------------------------------- # +# Registry shape / normalization +# --------------------------------------------------------------------------- # + +def test_evaluate_returns_normalized_check_shape(app): + result = SetupHealthService.evaluate() + assert 'items' in result and 'summary' in result + for c in result['items']: + # doctor _check shape … + assert set(('key', 'title', 'status', 'detail', 'repairable', + 'repair_ref')).issubset(c) + assert c['status'] in ('ok', 'warn', 'fail') + # … plus setup extras + assert c['section'] == 'setup' + assert c['severity'] in ('critical', 'recommended') + assert c['scope'] == 'panel' + assert c['fix']['kind'] in ('link', 'repair') + + +def test_summary_counts_and_score(app): + result = SetupHealthService.evaluate() + s = result['summary'] + assert s['total'] == len(result['items']) + assert s['ok'] + s['critical_open'] + s['recommended_open'] == s['total'] + assert 0 <= s['score'] <= 100 + + +# --------------------------------------------------------------------------- # +# Probe matrix — each item ok / not-ok by settings state +# --------------------------------------------------------------------------- # + +def test_public_ip_ok_when_set(app): + set_setting('server_public_ip', '203.0.113.7') + c = items_by_key(SetupHealthService.evaluate())['setup.public_ip'] + assert c['status'] == 'ok' + + +def test_public_ip_recommended_when_unset_and_not_per_site(app): + # No base domain → publishing_gaps short-circuits to no_base_domain, so the + # missing IP is only a recommendation, not silent breakage. + c = items_by_key(SetupHealthService.evaluate())['setup.public_ip'] + assert c['status'] == 'warn' + assert c['severity'] == 'recommended' + + +def test_public_ip_critical_when_per_site_mode_without_ip(app): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + c = items_by_key(SetupHealthService.evaluate())['setup.public_ip'] + assert c['status'] == 'fail' + assert c['severity'] == 'critical' + + +def test_base_domain_ok_when_set(app): + set_setting('sites_base_domain', 'apps.example.com') + c = items_by_key(SetupHealthService.evaluate())['setup.base_domain'] + assert c['status'] == 'ok' + + +def test_base_domain_recommended_when_unset(app): + c = items_by_key(SetupHealthService.evaluate())['setup.base_domain'] + assert c['status'] == 'warn' + assert c['severity'] == 'recommended' + + +def test_dns_provider_ok_when_connected(app): + add_dns_provider() + c = items_by_key(SetupHealthService.evaluate())['setup.dns_provider'] + assert c['status'] == 'ok' + + +def test_dns_provider_critical_when_per_site_without_provider(app): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + c = items_by_key(SetupHealthService.evaluate())['setup.dns_provider'] + assert c['status'] == 'fail' + assert c['severity'] == 'critical' + + +def test_dns_provider_recommended_when_wildcard_without_provider(app): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'wildcard') + c = items_by_key(SetupHealthService.evaluate())['setup.dns_provider'] + assert c['status'] == 'warn' + assert c['severity'] == 'recommended' + + +def test_email_delivery_ok_when_tested(app): + add_email_provider(tested_ok=True) + c = items_by_key(SetupHealthService.evaluate())['setup.email_delivery'] + assert c['status'] == 'ok' + + +def test_email_delivery_warn_when_untested(app): + add_email_provider(tested_ok=False) + c = items_by_key(SetupHealthService.evaluate())['setup.email_delivery'] + assert c['status'] == 'warn' + assert c['severity'] == 'recommended' + + +def test_email_delivery_warn_when_absent(app): + c = items_by_key(SetupHealthService.evaluate())['setup.email_delivery'] + assert c['status'] == 'warn' + + +def test_backup_policy_ok_with_enabled_policy(app): + add_backup_policy(enabled=True) + c = items_by_key(SetupHealthService.evaluate())['setup.backup_policy'] + assert c['status'] == 'ok' + + +def test_backup_policy_warn_without_any(app): + add_backup_policy(enabled=False) + c = items_by_key(SetupHealthService.evaluate())['setup.backup_policy'] + assert c['status'] == 'warn' + + +def test_backup_offsite_warn_on_local_only(app): + c = items_by_key(SetupHealthService.evaluate())['setup.backup_offsite'] + assert c['status'] == 'warn' + assert c['severity'] == 'recommended' + + +def test_backup_offsite_ok_with_s3(app, monkeypatch): + from app.services.storage_provider_service import StorageProviderService + monkeypatch.setattr(StorageProviderService, 'get_config', + classmethod(lambda cls: {'provider': 's3'})) + c = items_by_key(SetupHealthService.evaluate())['setup.backup_offsite'] + assert c['status'] == 'ok' + + +def test_canonical_domain_ok_when_set_no_overlap(app): + set_setting('canonical_domain', 'panel.example.com') + c = items_by_key(SetupHealthService.evaluate())['setup.canonical_domain'] + assert c['status'] == 'ok' + + +def test_canonical_domain_warn_when_unset(app): + c = items_by_key(SetupHealthService.evaluate())['setup.canonical_domain'] + assert c['status'] == 'warn' + + +# --------------------------------------------------------------------------- # +# Severity mapping — HTTPS is NEVER critical (SSL optional by decree) +# --------------------------------------------------------------------------- # + +def test_wildcard_cert_absent_when_https_off(app): + # Wildcard cert item is only applicable once wildcard HTTPS is switched on. + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'wildcard') + keys = items_by_key(SetupHealthService.evaluate()) + assert 'setup.wildcard_cert' not in keys + + +def test_wildcard_cert_never_critical_when_missing(app, monkeypatch): + import sys + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'wildcard') + set_setting('sites_https_enabled', 'true', vtype='boolean') + # Pretend we're on the Linux host with the cert absent. + monkeypatch.setattr(sys, 'platform', 'linux') + import app.services.setup_health_service as m + monkeypatch.setattr(m.os.path, 'exists', lambda p: False) + c = items_by_key(SetupHealthService.evaluate())['setup.wildcard_cert'] + assert c['status'] == 'warn' # NOT 'fail' + assert c['severity'] == 'recommended' # HTTPS never critical + + +def test_no_https_item_is_ever_critical(app, monkeypatch): + """Across every settings permutation, an HTTPS/TLS item can never come back + critical — the SSL-optional invariant, asserted directly.""" + import sys + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'wildcard') + set_setting('sites_https_enabled', 'true', vtype='boolean') + monkeypatch.setattr(sys, 'platform', 'linux') + import app.services.setup_health_service as m + monkeypatch.setattr(m.os.path, 'exists', lambda p: False) + for c in SetupHealthService.evaluate()['items']: + if 'cert' in c['key'] or 'https' in c['key']: + assert c['severity'] != 'critical' + + +# --------------------------------------------------------------------------- # +# Doctor-report inclusion + API +# --------------------------------------------------------------------------- # + +def test_doctor_report_includes_setup_section(app, monkeypatch): + from app.services.doctor_service import DoctorService + # Keep the sweep cheap/offline — only care that setup.* is present. + monkeypatch.setattr(DoctorService, '_drift_checks', classmethod(lambda cls: [])) + monkeypatch.setattr(DoctorService, '_service_checks', classmethod(lambda cls: [])) + monkeypatch.setattr(DoctorService, '_dns_checks', classmethod(lambda cls: [])) + report = DoctorService.run() + setup_keys = [c['key'] for c in report['checks'] if c['key'].startswith('setup.')] + assert 'setup.public_ip' in setup_keys + assert 'setup.dns_provider' in setup_keys + + +def test_api_returns_items_and_summary(app, client, auth_headers): + res = client.get('/api/v1/setup-health', headers=auth_headers) + assert res.status_code == 200 + body = res.get_json() + assert 'items' in body and 'summary' in body + assert body['summary']['total'] == len(body['items']) + + +def test_api_requires_admin(app, client): + res = client.get('/api/v1/setup-health') + assert res.status_code in (401, 422) + + +def test_fingerprint_changes_only_with_critical_set(app): + # Fresh panel: no critical items (no per-site breakage) → empty fingerprint. + fp_clean = SetupHealthService.fingerprint() + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + fp_broken = SetupHealthService.fingerprint() + assert fp_clean != fp_broken + assert 'setup.public_ip' in fp_broken diff --git a/backend/tests/test_setup_nag.py b/backend/tests/test_setup_nag.py new file mode 100644 index 00000000..1b934677 --- /dev/null +++ b/backend/tests/test_setup_nag.py @@ -0,0 +1,212 @@ +"""Tests for the single Setup Health nag + snooze/dismiss (plan 22 Phase 6): +the weekly digest fires only while non-snoozed critical items exist and only +when the critical set changed; snoozed items still render but are muted, don't +count as open, and don't trigger the nag.""" +import pytest + +from app.services.setup_health_service import ( + NAG_JOB_KIND, + NAG_MARKER_KEY, + NAG_SCHEDULE_NAME, + PANEL_SNOOZE_KEY, + SetupHealthService, +) + + +def set_setting(key, value, vtype='string'): + from app import db + from app.models.system_settings import SystemSettings + SystemSettings.set(key=key, value=value, value_type=vtype) + db.session.commit() + + +def force_critical(): + """A settings state that yields a critical (fail) panel item: + per-site DNS mode + a base but no provider/IP → setup.dns_provider fail.""" + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + + +@pytest.fixture +def admin(app): + from werkzeug.security import generate_password_hash + from app import db + from app.models import User + u = User(email='nag@t.local', username='nagadmin', + password_hash=generate_password_hash('x'), + role=User.ROLE_ADMIN, is_active=True) + db.session.add(u) + db.session.commit() + return u + + +# --------------------------------------------------------------------------- # +# Snooze mutes an item (still renders, not counted, no nag) +# --------------------------------------------------------------------------- # + +def test_panel_snooze_mutes_item_and_drops_open_count(app): + force_critical() + before = SetupHealthService.evaluate(scope='panel') + dns = {c['key']: c for c in before['items']}['setup.dns_provider'] + assert dns['status'] == 'fail' + assert before['summary']['critical_open'] >= 1 + + SetupHealthService.snooze('setup.dns_provider', days=14) + after = SetupHealthService.evaluate(scope='panel') + dns = {c['key']: c for c in after['items']}['setup.dns_provider'] + # Still rendered, but muted … + assert dns['snoozed'] is True + assert dns['snoozed_until'] + # … and no longer counted as open. + assert after['summary']['critical_open'] == before['summary']['critical_open'] - 1 + assert after['summary']['snoozed'] >= 1 + + +def test_unsnooze_restores_open_count(app): + force_critical() + SetupHealthService.snooze('setup.dns_provider', days=14) + SetupHealthService.unsnooze('setup.dns_provider') + after = SetupHealthService.evaluate(scope='panel') + dns = {c['key']: c for c in after['items']}['setup.dns_provider'] + assert not dns.get('snoozed') + + +def test_expired_snooze_is_ignored(app): + force_critical() + # Write an already-expired snooze directly. + import json + set_setting(PANEL_SNOOZE_KEY, json.dumps({'setup.dns_provider': '2000-01-01T00:00:00'})) + after = SetupHealthService.evaluate(scope='panel') + dns = {c['key']: c for c in after['items']}['setup.dns_provider'] + assert not dns.get('snoozed') # expired → active again + + +def test_user_snooze_lives_on_the_user_row(app, admin): + SetupHealthService.snooze('setup.account_security', days=7, user=admin) + assert admin.setup_snoozes # persisted on the row, not the settings map + res = SetupHealthService.evaluate(scope='user', user=admin) + item = {c['key']: c for c in res['items']}['setup.account_security'] + assert item['snoozed'] is True + + +def test_snooze_unknown_item_errors(app): + assert 'error' in SetupHealthService.snooze('setup.nonexistent') + + +# --------------------------------------------------------------------------- # +# The nag job — critical-only, fingerprint-throttled, snooze-aware +# --------------------------------------------------------------------------- # + +def _capture_notify(monkeypatch): + import app.plugins_sdk as sdk + sent = [] + monkeypatch.setattr(sdk.notify, 'send', + lambda event, to, data=None, **kw: sent.append((event, to, data))) + return sent + + +def test_nag_fires_once_then_throttles(app, monkeypatch): + force_critical() + sent = _capture_notify(monkeypatch) + + first = SetupHealthService.run_nag_job(object()) + assert first['notified'] is True + assert len(sent) == 1 + event, to, data = sent[0] + assert event == 'setup.incomplete' + assert to == 'admins' + assert data['count'] >= 1 + + # Second run, same critical set → throttled (no re-nag). + second = SetupHealthService.run_nag_job(object()) + assert second['notified'] is False + assert second.get('reason') == 'unchanged' + assert len(sent) == 1 + + +def test_nag_does_not_fire_without_criticals(app, monkeypatch): + # No base/per-site config → no critical items. + sent = _capture_notify(monkeypatch) + res = SetupHealthService.run_nag_job(object()) + assert res['notified'] is False + assert res['critical_open'] == 0 + assert sent == [] + + +def test_snoozed_critical_does_not_trigger_nag(app, monkeypatch): + force_critical() + SetupHealthService.snooze('setup.dns_provider', days=30) + # public_ip may also be critical in per-site mode; snooze it too. + SetupHealthService.snooze('setup.public_ip', days=30) + sent = _capture_notify(monkeypatch) + res = SetupHealthService.run_nag_job(object()) + assert res['notified'] is False + assert res['critical_open'] == 0 + assert sent == [] + + +def test_nag_marker_resets_when_criticals_clear(app, monkeypatch): + from app.services.settings_service import SettingsService + force_critical() + _capture_notify(monkeypatch) + SetupHealthService.run_nag_job(object()) + assert SettingsService.get(NAG_MARKER_KEY) # marker set + + # Resolve the criticals (wildcard mode, no per-site requirement). + set_setting('sites_dns_mode', 'wildcard') + SetupHealthService.run_nag_job(object()) + assert (SettingsService.get(NAG_MARKER_KEY) or '') == '' # marker cleared + + +# --------------------------------------------------------------------------- # +# Schedule + API +# --------------------------------------------------------------------------- # + +def test_nag_schedule_is_seeded(app): + from app.jobs.builtin_handlers import seed_builtin_schedules + from app.jobs.models import ScheduledJob + seed_builtin_schedules() + row = ScheduledJob.query.filter_by(name=NAG_SCHEDULE_NAME).first() + assert row is not None + assert row.kind == NAG_JOB_KIND + assert row.interval_seconds == 604800 + + +def test_setup_incomplete_in_catalog(): + from app.notifications import catalog + entry = catalog.get('setup.incomplete') + assert entry is not None + assert entry['category'] == 'system' + + +def test_snooze_api_panel_requires_admin(app, client): + from app import db + from app.models import User + from flask_jwt_extended import create_access_token + from werkzeug.security import generate_password_hash + dev = User(email='d@t.local', username='devuser', + password_hash=generate_password_hash('x'), + role='developer', is_active=True) + db.session.add(dev) + db.session.commit() + headers = {'Authorization': f'Bearer {create_access_token(identity=dev.id)}'} + res = client.post('/api/v1/setup-health/snooze', + json={'key': 'setup.dns_provider'}, headers=headers) + assert res.status_code == 403 # panel item, non-admin + + +def test_snooze_api_user_item_allows_self(app, client): + from app import db + from app.models import User + from flask_jwt_extended import create_access_token + from werkzeug.security import generate_password_hash + dev = User(email='d2@t.local', username='devuser2', + password_hash=generate_password_hash('x'), + role='developer', is_active=True) + db.session.add(dev) + db.session.commit() + headers = {'Authorization': f'Bearer {create_access_token(identity=dev.id)}'} + res = client.post('/api/v1/setup-health/snooze', + json={'key': 'setup.account_security', 'days': 7}, headers=headers) + assert res.status_code == 200 + assert res.get_json()['scope'] == 'user' diff --git a/backend/tests/test_setup_preflight.py b/backend/tests/test_setup_preflight.py new file mode 100644 index 00000000..406f75b2 --- /dev/null +++ b/backend/tests/test_setup_preflight.py @@ -0,0 +1,125 @@ +"""Tests for creation-time preflight warnings (SiteDomainService.creation_warnings ++ the give_subdomain response): create-with-gaps returns codes, create-clean +returns empty, and the legacy `warning` string is still populated.""" +import pytest + +from app.services.site_domain_service import SiteDomainService + + +def set_setting(key, value, vtype='string'): + from app import db + from app.models.system_settings import SystemSettings + SystemSettings.set(key=key, value=value, value_type=vtype) + db.session.commit() + + +@pytest.fixture(autouse=True) +def _no_slow_dns(monkeypatch): + """Keep the resolvability probe deterministic + offline: default to 'does not + resolve' unless a test says otherwise.""" + monkeypatch.setattr(SiteDomainService, '_host_resolves', + staticmethod(lambda host, timeout=2.0: False)) + + +# --------------------------------------------------------------------------- # +# creation_warnings — codes + fix shape +# --------------------------------------------------------------------------- # + +def test_no_base_domain_yields_that_code(app): + # Force an empty base domain (testing config defaults to lvh.me). + from app.models.system_settings import SystemSettings + monkey = SystemSettings.get # noqa: F841 + set_setting('sites_base_domain', '') + # base_domain() falls back to config lvh.me even when the setting is '', so + # patch it to empty to exercise the no_base_domain gap. + import app.services.site_domain_service as m + orig = m.SiteDomainService.base_domain + m.SiteDomainService.base_domain = classmethod(lambda cls: '') + try: + warnings = SiteDomainService.creation_warnings() + finally: + m.SiteDomainService.base_domain = orig + codes = {w['code'] for w in warnings} + assert 'no_base_domain' in codes + w = next(w for w in warnings if w['code'] == 'no_base_domain') + assert w['fix'] == {'kind': 'link', 'to': '/settings/site'} + + +def test_http_only_gap_present_when_base_set_without_https(app): + set_setting('sites_base_domain', 'apps.example.com') + codes = {w['code'] for w in SiteDomainService.creation_warnings()} + assert 'http_only' in codes # base set, HTTPS off → recommended gap + + +def test_unresolved_host_adds_host_unresolved_code(app): + set_setting('sites_base_domain', 'apps.example.com') + warnings = SiteDomainService.creation_warnings('blog.apps.example.com') + codes = {w['code'] for w in warnings} + assert 'host_unresolved' in codes + w = next(w for w in warnings if w['code'] == 'host_unresolved') + assert w['fix'] == {'kind': 'link', 'to': '/monitoring/doctor'} + + +def test_resolving_host_does_not_add_unresolved(app, monkeypatch): + set_setting('sites_base_domain', 'apps.example.com') + monkeypatch.setattr(SiteDomainService, '_host_resolves', + staticmethod(lambda host, timeout=2.0: True)) + codes = {w['code'] for w in SiteDomainService.creation_warnings('blog.apps.example.com')} + assert 'host_unresolved' not in codes + + +def test_dev_host_is_never_probed(app): + set_setting('sites_base_domain', 'apps.example.com') + # lvh.me is a dev name — skipped by the resolvability probe entirely. + codes = {w['code'] for w in SiteDomainService.creation_warnings('thing.lvh.me')} + assert 'host_unresolved' not in codes + + +def test_clean_setup_returns_no_gaps(app, monkeypatch): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_https_enabled', 'true', vtype='boolean') + monkeypatch.setattr(SiteDomainService, '_host_resolves', + staticmethod(lambda host, timeout=2.0: True)) + # No canonical/panel overlap, wildcard mode default → no gaps. + warnings = SiteDomainService.creation_warnings('blog.apps.example.com') + assert warnings == [] + + +# --------------------------------------------------------------------------- # +# give_subdomain response carries both the legacy string and the array +# --------------------------------------------------------------------------- # + +@pytest.fixture +def app_row(app): + from werkzeug.security import generate_password_hash + from app import db + from app.models import User + from app.models.application import Application + u = User(email='pf@test.local', username='pfuser', + password_hash=generate_password_hash('x'), + role=User.ROLE_ADMIN, is_active=True) + db.session.add(u) + db.session.flush() + a = Application(name='pfsite', app_type='static', user_id=u.id, + root_path='/srv/pfsite') + db.session.add(a) + db.session.commit() + return a + + +def test_give_subdomain_returns_warnings_array_and_legacy_warning(app, app_row, monkeypatch): + # A real base domain so a subdomain is actually assigned; keep nginx writes + # from touching the host by stubbing the vhost writer. + set_setting('sites_base_domain', 'apps.example.com') + monkeypatch.setattr(SiteDomainService, 'write_app_vhost', + classmethod(lambda cls, a, force_type=None: {'nginx': None, 'warning': None})) + + result = SiteDomainService.give_subdomain(app_row, label='blog') + assert result['success'] is True + assert result['host'] == 'blog.apps.example.com' + # The array is present and carries structured codes … + assert isinstance(result['warnings'], list) + codes = {w['code'] for w in result['warnings']} + assert 'host_unresolved' in codes # unresolved by the autouse stub + # … and the legacy single 'warning' key still exists (None here). + assert 'warning' in result diff --git a/backend/tests/test_setup_reconcile.py b/backend/tests/test_setup_reconcile.py new file mode 100644 index 00000000..be44daff --- /dev/null +++ b/backend/tests/test_setup_reconcile.py @@ -0,0 +1,271 @@ +"""Tests for reconcile-on-connect (setup_reconcile_service): DNS backfill preview +correctness (wildcard vs per-site), the apply loop's idempotency, and the +WordPress url-fix detection matrix + apply.""" +import pytest + +from app.services.setup_reconcile_service import SetupReconcileService + + +# --------------------------------------------------------------------------- # +# Fixtures / helpers +# --------------------------------------------------------------------------- # + +def set_setting(key, value, vtype='string'): + from app import db + from app.models.system_settings import SystemSettings + SystemSettings.set(key=key, value=value, value_type=vtype) + db.session.commit() + + +@pytest.fixture +def owner(app): + from werkzeug.security import generate_password_hash + from app import db + from app.models import User + u = User(email='rec@test.local', username='recuser', + password_hash=generate_password_hash('x'), + role=User.ROLE_ADMIN, is_active=True) + db.session.add(u) + db.session.commit() + return u + + +def make_app_with_domain(owner, name, host, primary=True): + from app import db + from app.models.application import Application + from app.models.domain import Domain + a = Application(name=name, app_type='static', user_id=owner.id) + db.session.add(a) + db.session.flush() + db.session.add(Domain(name=host, application_id=a.id, is_primary=primary)) + db.session.commit() + return a + + +def add_provider(): + from app import db + from app.models.email import DNSProviderConfig + db.session.add(DNSProviderConfig(name='CF', provider='cloudflare', api_key='x')) + db.session.commit() + + +# --------------------------------------------------------------------------- # +# DNS backfill — candidate enumeration by mode +# --------------------------------------------------------------------------- # + +def test_wildcard_mode_yields_one_wildcard_record_and_skips_subdomains(app, owner): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'wildcard') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + + cands = SetupReconcileService._dns_candidates() + hosts = {c['host']: c for c in cands} + assert '*.apps.example.com' in hosts + assert hosts['*.apps.example.com']['mode'] == 'wildcard' + # The subdomain is covered by the wildcard, so it isn't a separate candidate. + assert 'blog.apps.example.com' not in hosts + + +def test_per_site_mode_yields_each_subdomain_and_no_wildcard(app, owner): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + make_app_with_domain(owner, 'shop', 'shop.apps.example.com') + + cands = SetupReconcileService._dns_candidates() + hosts = {c['host']: c for c in cands} + assert '*.apps.example.com' not in hosts + assert hosts['blog.apps.example.com']['mode'] == 'per-site' + assert hosts['shop.apps.example.com']['mode'] == 'per-site' + + +def test_custom_domain_outside_base_is_always_a_candidate(app, owner): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'wildcard') + make_app_with_domain(owner, 'store', 'store.acme.org') + + hosts = {c['host']: c for c in SetupReconcileService._dns_candidates()} + assert hosts['store.acme.org']['mode'] == 'custom' + + +def test_dev_and_ip_hosts_are_skipped(app, owner): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + make_app_with_domain(owner, 'dev', 'thing.lvh.me') + make_app_with_domain(owner, 'ip', '203.0.113.9') + + hosts = {c['host'] for c in SetupReconcileService._dns_candidates()} + assert 'thing.lvh.me' not in hosts + assert '203.0.113.9' not in hosts + + +def test_preview_reports_readiness_and_provider_coverage(app, owner, monkeypatch): + from app.services.dns_provider_service import DNSProviderService + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + set_setting('server_public_ip', '203.0.113.7') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + add_provider() + monkeypatch.setattr(DNSProviderService, 'find_zone_for_domain', + classmethod(lambda cls, d: ({'name': 'CF'}, {'id': 'z1'}))) + + preview = SetupReconcileService.dns_backfill_preview() + assert preview['ready'] is True + assert preview['has_provider'] is True + assert preview['server_ip'] == '203.0.113.7' + item = next(i for i in preview['items'] if i['host'] == 'blog.apps.example.com') + assert item['provider_covers_zone'] is True + assert item['target_ip'] == '203.0.113.7' + + +def test_preview_not_ready_without_ip(app, owner): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + add_provider() + preview = SetupReconcileService.dns_backfill_preview() + assert preview['ready'] is False # no server IP + + +# --------------------------------------------------------------------------- # +# DNS backfill — apply loop + idempotency (second run = no-ops) +# --------------------------------------------------------------------------- # + +def test_apply_creates_then_is_idempotent(app, owner, monkeypatch): + from app.services.dns_provider_service import DNSProviderService + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + set_setting('server_public_ip', '203.0.113.7') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + make_app_with_domain(owner, 'shop', 'shop.apps.example.com') + + # Stateful fake provider: first ensure creates, later ones are no-ops. + existing = set() + + def fake_ensure(cls, domain, ip): + if domain in existing: + return {'created': False, 'reason': 'exists', 'provider': 'CF'} + existing.add(domain) + return {'created': True, 'provider': 'CF', + 'record': {'type': 'A', 'name': domain, 'value': ip}} + monkeypatch.setattr(DNSProviderService, 'ensure_a_record', classmethod(fake_ensure)) + + first = SetupReconcileService.dns_backfill_apply() + assert first['applied'] == 2 + assert {r['host'] for r in first['results']} == { + 'blog.apps.example.com', 'shop.apps.example.com'} + + second = SetupReconcileService.dns_backfill_apply() + assert second['applied'] == 0 # idempotent — no new writes + assert all(r['created'] is False for r in second['results']) + + +def test_apply_without_ip_is_a_noop(app, owner): + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + res = SetupReconcileService.dns_backfill_apply() + assert res['applied'] == 0 + assert 'error' in res + + +def test_backfill_job_handler_runs_apply(app, owner, monkeypatch): + from app.services.dns_provider_service import DNSProviderService + set_setting('sites_base_domain', 'apps.example.com') + set_setting('sites_dns_mode', 'per-site') + set_setting('server_public_ip', '203.0.113.7') + make_app_with_domain(owner, 'blog', 'blog.apps.example.com') + monkeypatch.setattr(DNSProviderService, 'ensure_a_record', + classmethod(lambda cls, d, ip: {'created': True, 'provider': 'CF'})) + summary = SetupReconcileService.run_dns_backfill_job(object()) + assert summary['applied'] == 1 + + +# --------------------------------------------------------------------------- # +# WordPress url-fix — detection matrix (pure classifier) +# --------------------------------------------------------------------------- # + +@pytest.mark.parametrize('current,target,expected', [ + ('http://localhost:8300', 'https://blog.example.com', True), # localhost → domain + ('http://127.0.0.1:8300', 'https://blog.example.com', True), # loopback → domain + ('http://203.0.113.9', 'https://blog.example.com', True), # stale IP → domain + ('https://blog.example.com', 'https://blog.example.com', False), # already correct + ('https://old.example.com', 'https://blog.example.com', False), # domain → domain: leave + ('http://localhost:8300', 'http://localhost:9000', False), # no real target + ('http://localhost:8300', '', False), # no target + (None, 'https://blog.example.com', False), # unknown current +]) +def test_url_fix_needed_matrix(current, target, expected): + assert SetupReconcileService.url_fix_needed(current, target) is expected + + +# --------------------------------------------------------------------------- # +# WordPress url-fix — preview + apply with a mocked WP service +# --------------------------------------------------------------------------- # + +@pytest.fixture +def wp_site(app, owner): + """A WordPressSite on an Application with a primary managed domain.""" + from app import db + from app.models.application import Application + from app.models.domain import Domain + from app.models.wordpress_site import WordPressSite + a = Application(name='wpblog', app_type='wordpress', user_id=owner.id, + root_path='/srv/sites/wpblog', port=8300) + db.session.add(a) + db.session.flush() + db.session.add(Domain(name='blog.apps.example.com', application_id=a.id, is_primary=True)) + site = WordPressSite(application_id=a.id) + db.session.add(site) + db.session.commit() + return site + + +class _FakeWP: + """Stands in for the WordPress extension service (no wp-cli).""" + current = 'http://localhost:8300' + swapped = [] + + @classmethod + def _site_current_url(cls, app): + return cls.current + + @classmethod + def preview_url_change(cls, app, new_url): + return {'success': True, 'current_url': cls.current, 'new_url': new_url, + 'pairs': [{'search': cls.current, 'replace': new_url, 'replacements': 12}], + 'total': 12} + + @classmethod + def change_site_url(cls, app, new_url, keep_old_redirect=True): + cls.swapped.append(new_url) + cls.current = new_url # now correct → not re-detected + return {'success': True, 'old_url': 'http://localhost:8300', 'new_url': new_url} + + +def test_url_fix_preview_detects_localhost_site(app, wp_site, monkeypatch): + import app.services.wordpress_bridge as bridge + _FakeWP.current = 'http://localhost:8300' + monkeypatch.setattr(bridge, 'wordpress_service', lambda: _FakeWP) + + preview = SetupReconcileService.url_fix_preview() + assert preview['count'] == 1 + item = preview['items'][0] + assert item['current_url'] == 'http://localhost:8300' + assert item['new_url'].endswith('blog.apps.example.com') + assert item['total'] == 12 + + +def test_url_fix_apply_swaps_then_is_idempotent(app, wp_site, monkeypatch): + import app.services.wordpress_bridge as bridge + _FakeWP.current = 'http://localhost:8300' + _FakeWP.swapped = [] + monkeypatch.setattr(bridge, 'wordpress_service', lambda: _FakeWP) + + first = SetupReconcileService.url_fix_apply() + assert first['fixed'] == 1 + assert _FakeWP.swapped and _FakeWP.swapped[0].endswith('blog.apps.example.com') + + # Second run: the site is now on its target URL, so nothing is detected. + second = SetupReconcileService.url_fix_apply() + assert second['fixed'] == 0