""" Config API integration tests. Covers: - GET /api/config — shape, required fields - PUT /api/config — partial updates, validation rejections - GET /api/config/export — returns content - POST /api/config/import — valid and invalid payloads - POST /api/config/backup — creates a backup entry - GET /api/config/backups — lists backups Run with: pytest tests/integration/test_config_api.py -v """ import pytest import requests import sys import os sys.path.insert(0, os.path.dirname(__file__)) from conftest import API_BASE, _resolve_admin_pass _S = None @pytest.fixture(scope='module', autouse=True) def _auth_session(): global _S _S = requests.Session() _S.headers['Content-Type'] = 'application/json' r = _S.post(f"{API_BASE}/api/auth/login", json={'username': 'admin', 'password': _resolve_admin_pass()}) assert r.status_code == 200, f"Login failed: {{r.text}}" def get(path, **kw): return _S.get(f"{API_BASE}{path}", **kw) def put(path, **kw): return _S.put(f"{API_BASE}{path}", **kw) def post(path, **kw): return _S.post(f"{API_BASE}{path}", **kw) # --------------------------------------------------------------------------- # GET /api/config # --------------------------------------------------------------------------- class TestGetConfig: def test_get_config_returns_200(self): r = get('/api/config') assert r.status_code == 200 def test_get_config_content_type_is_json(self): r = get('/api/config') assert 'application/json' in r.headers.get('Content-Type', '') def test_get_config_has_cell_name(self): data = get('/api/config').json() assert 'cell_name' in data assert isinstance(data['cell_name'], str) assert data['cell_name'] # non-empty def test_get_config_has_domain(self): data = get('/api/config').json() assert 'domain' in data assert isinstance(data['domain'], str) def test_get_config_has_valid_ip_range(self): import ipaddress data = get('/api/config').json() assert 'ip_range' in data # Must be a parseable IPv4 CIDR net = ipaddress.ip_network(data['ip_range'], strict=False) assert net.version == 4, f"ip_range {data['ip_range']} is not IPv4" def test_get_config_has_wireguard_port(self): data = get('/api/config').json() assert 'wireguard_port' in data port = data['wireguard_port'] assert isinstance(port, int) assert 1 <= port <= 65535 def test_get_config_has_service_ips(self): data = get('/api/config').json() assert 'service_ips' in data sips = data['service_ips'] for key in ('dns', 'vip_mail', 'vip_calendar', 'vip_files', 'vip_webdav'): assert key in sips, f"service_ips missing key: {key}" def test_get_config_has_service_configs(self): data = get('/api/config').json() assert 'service_configs' in data assert isinstance(data['service_configs'], dict) # --------------------------------------------------------------------------- # PUT /api/config — positive cases # --------------------------------------------------------------------------- class TestPutConfigPositive: def test_put_config_returns_200(self): # Read current cell_name first so we can restore it safely current = get('/api/config').json() original_name = current['cell_name'] # Write back the same value — idempotent, no real change r = put('/api/config', json={'cell_name': original_name}) assert r.status_code == 200 def test_put_config_response_has_message(self): r = put('/api/config', json={'cell_name': get('/api/config').json()['cell_name']}) assert r.status_code == 200 assert 'message' in r.json() def test_put_config_update_cell_name_persists(self): original_name = get('/api/config').json()['cell_name'] new_name = original_name + '-test' try: r = put('/api/config', json={'cell_name': new_name}) assert r.status_code == 200 updated = get('/api/config').json() assert updated['cell_name'] == new_name finally: # Restore original name put('/api/config', json={'cell_name': original_name}) def test_put_config_update_domain_persists(self): original_domain = get('/api/config').json()['domain'] # Write same domain back to confirm the round-trip works without side effects r = put('/api/config', json={'domain': original_domain}) assert r.status_code == 200 assert get('/api/config').json()['domain'] == original_domain def test_put_config_valid_ip_range_accepted(self): # Use a known-valid RFC-1918 range; restore the original after original_range = get('/api/config').json()['ip_range'] r = put('/api/config', json={'ip_range': '172.20.0.0/16'}) try: assert r.status_code == 200 finally: put('/api/config', json={'ip_range': original_range}) def test_put_config_unknown_top_level_key_does_not_crash(self): # Unknown keys that are not identity fields and not service keys should # be silently ignored rather than causing a 500. r = put('/api/config', json={'totally_unknown_field_xyz': 'value'}) assert r.status_code in (200, 400), ( f"Unexpected status {r.status_code} for unknown field" ) # --------------------------------------------------------------------------- # PUT /api/config — validation rejections # --------------------------------------------------------------------------- class TestPutConfigValidation: def test_put_config_empty_body_returns_400(self): r = put('/api/config', data='') assert r.status_code == 400 def test_put_config_invalid_json_returns_400(self): r = put('/api/config', data='not valid json }{') assert r.status_code == 400 def test_put_config_ip_range_not_rfc1918_returns_400(self): # 8.8.0.0/16 is a public range — must be rejected r = put('/api/config', json={'ip_range': '8.8.0.0/16'}) assert r.status_code == 400 body = r.json() assert 'error' in body assert 'ip_range' in body['error'].lower() or 'rfc' in body['error'].lower() def test_put_config_ip_range_outside_172_16_prefix_returns_400(self): # 172.0.0.0/24 looks like a 172.x range but is NOT within 172.16.0.0/12 r = put('/api/config', json={'ip_range': '172.0.0.0/24'}) assert r.status_code == 400 def test_put_config_ip_range_malformed_returns_400(self): r = put('/api/config', json={'ip_range': 'not-an-ip'}) assert r.status_code == 400 def test_put_config_ip_range_bare_ip_returns_400(self): # Bare IP without CIDR prefix must be rejected — /32 networks are # accepted by Python but useless as a Docker subnet. r = put('/api/config', json={'ip_range': '10.0.0.1'}) assert r.status_code == 400 def test_put_config_calendar_port_zero_returns_400(self): r = put('/api/config', json={'calendar': {'port': 0}}) assert r.status_code == 400 assert 'error' in r.json() def test_put_config_calendar_port_too_high_returns_400(self): r = put('/api/config', json={'calendar': {'port': 65536}}) assert r.status_code == 400 assert 'error' in r.json() def test_put_config_files_port_negative_returns_400(self): r = put('/api/config', json={'files': {'port': -1}}) assert r.status_code == 400 def test_put_config_wireguard_address_without_prefix_returns_400(self): # wireguard.address must include prefix length r = put('/api/config', json={'wireguard': {'address': '10.0.0.1'}}) assert r.status_code == 400 assert 'error' in r.json() def test_put_config_wireguard_address_invalid_returns_400(self): r = put('/api/config', json={'wireguard': {'address': 'not-an-ip/24'}}) assert r.status_code == 400 # --------------------------------------------------------------------------- # GET /api/config/export # --------------------------------------------------------------------------- class TestConfigExport: def test_export_returns_200(self): r = get('/api/config/export') assert r.status_code == 200 def test_export_has_config_key(self): data = get('/api/config/export').json() assert 'config' in data def test_export_has_format_key(self): data = get('/api/config/export').json() assert 'format' in data def test_export_config_content_is_not_empty(self): data = get('/api/config/export').json() assert data['config'] # non-empty / non-None # --------------------------------------------------------------------------- # POST /api/config/import # --------------------------------------------------------------------------- class TestConfigImport: def test_import_missing_body_returns_400(self): r = post('/api/config/import', data='') assert r.status_code == 400 def test_import_invalid_json_returns_400(self): r = post('/api/config/import', data='{{bad json') assert r.status_code == 400 def test_import_valid_empty_config_does_not_crash(self): # Sending an empty config dict — the API should respond with 200 or a # meaningful error, not a 500 traceback. r = post('/api/config/import', json={'config': {}, 'format': 'json'}) assert r.status_code in (200, 400, 422, 500) # Confirm the response is still valid JSON r.json() def test_import_round_trips_exported_config(self): # Export current config, import it back — should succeed without errors. exported = get('/api/config/export').json() r = post('/api/config/import', json={ 'config': exported['config'], 'format': exported.get('format', 'json'), }) assert r.status_code in (200, 400), ( f"Unexpected status {r.status_code}: {r.text}" ) # --------------------------------------------------------------------------- # POST /api/config/backup + GET /api/config/backups # --------------------------------------------------------------------------- class TestConfigBackup: def test_create_backup_returns_200(self): r = post('/api/config/backup') assert r.status_code == 200 def test_create_backup_returns_backup_id(self): r = post('/api/config/backup') assert r.status_code == 200 data = r.json() assert 'backup_id' in data assert data['backup_id'] def test_list_backups_returns_200(self): r = get('/api/config/backups') assert r.status_code == 200 def test_list_backups_returns_list(self): r = get('/api/config/backups') assert isinstance(r.json(), list) def test_backup_appears_in_list_after_creation(self): # Create a backup, then verify it shows up in the list. create_r = post('/api/config/backup') assert create_r.status_code == 200 new_id = create_r.json().get('backup_id') backups = get('/api/config/backups').json() # The list may contain IDs directly or dicts with an 'id' key ids = [] for entry in backups: if isinstance(entry, str): ids.append(entry) elif isinstance(entry, dict): ids.append(entry.get('id') or entry.get('backup_id') or '') assert new_id in ids, ( f"Newly created backup '{new_id}' not found in backups list: {backups}" ) # --------------------------------------------------------------------------- # GET /api/config/pending # --------------------------------------------------------------------------- class TestConfigPending: def test_pending_returns_200(self): r = get('/api/config/pending') assert r.status_code == 200 def test_pending_has_needs_restart_field(self): data = get('/api/config/pending').json() assert 'needs_restart' in data assert isinstance(data['needs_restart'], bool) def test_pending_has_changes_list(self): data = get('/api/config/pending').json() assert 'changes' in data assert isinstance(data['changes'], list)