feat: port conflict validation, autosave on Apply, extended integration tests

Port conflict validation:
- api/port_registry.py: detect_conflicts() checks all service sections for shared port values
- api/app.py: returns HTTP 409 on port conflict after existing range validation
- webui/src/pages/Settings.jsx: JS-side detectPortConflicts() with useMemo shows inline
  conflict errors and blocks Save before the request is made; catch blocks surface server
  error messages (including 409) instead of generic fallbacks

Config autosave on Apply:
- webui/src/contexts/DraftConfigContext.jsx: new context; Settings registers flush callbacks
  per section; App calls flushAll() before applyPending() when any section is dirty
- webui/src/App.jsx: wraps tree with DraftConfigProvider, handleApply shows 'saving' banner
  state and awaits flushAll()
- webui/src/pages/Settings.jsx: registers identity + per-service flushers; propagates dirty
  state into context via setDirty; uses refs to avoid stale closures

Extended integration test coverage (114 new tests):
- tests/integration/test_config_api.py: GET/PUT config, export, import, backup lifecycle
- tests/integration/test_network_services.py: DNS records + DHCP reservations CRUD
- tests/integration/test_containers.py: list, restart, logs, stats; recovery polling
- tests/integration/test_negative_scenarios.py: error-path coverage for all endpoints
- tests/test_port_conflicts.py: 20 unit tests for port_registry.detect_conflicts()

Pre-commit hook updated to skip tests/integration/ (live-stack tests require a running
stack and must be run explicitly via `make test-integration`).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-24 04:45:47 -04:00
parent 596b06f171
commit 768571f2b7
11 changed files with 1584 additions and 12 deletions
+216
View File
@@ -0,0 +1,216 @@
"""
Network services integration tests: DNS records, DHCP leases, DHCP reservations.
Note on endpoint shapes discovered from app.py:
- DELETE /api/dns/records takes a JSON body (not a URL param)
- DELETE /api/dhcp/reservations takes JSON body with 'mac' field
- POST /api/dhcp/reservations requires 'mac' and 'ip' fields
Run with: pytest tests/integration/test_network_services.py -v
"""
import pytest
import requests
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
from conftest import API_BASE
# Test DNS hostname to use — must be cleaned up after tests
_TEST_DNS_HOSTNAME = 'inttest-dns-record'
def get(path, **kw):
return requests.get(f"{API_BASE}{path}", **kw)
def post(path, **kw):
return requests.post(f"{API_BASE}{path}", **kw)
def delete(path, **kw):
return requests.delete(f"{API_BASE}{path}", **kw)
# ---------------------------------------------------------------------------
# GET /api/dns/records
# ---------------------------------------------------------------------------
class TestDnsRecordsRead:
def test_get_dns_records_returns_200(self):
r = get('/api/dns/records')
assert r.status_code == 200
def test_get_dns_records_returns_list_or_dict(self):
# The network_manager may return a list of records or a dict keyed by hostname
data = get('/api/dns/records').json()
assert isinstance(data, (list, dict))
def test_get_dns_status_returns_200(self):
r = get('/api/dns/status')
assert r.status_code == 200
def test_get_dns_status_returns_dict(self):
data = get('/api/dns/status').json()
assert isinstance(data, dict)
# ---------------------------------------------------------------------------
# POST /api/dns/records + DELETE /api/dns/records (round-trip)
# ---------------------------------------------------------------------------
class TestDnsRecordsWrite:
"""Create a DNS A record then delete it. The test is self-cleaning."""
def test_add_dns_record_returns_non_error(self):
"""Adding a well-formed A record should not return a 4xx or 5xx."""
r = post('/api/dns/records', json={
'zone': 'cell',
'name': _TEST_DNS_HOSTNAME,
'record_type': 'A',
'value': '10.0.0.99',
})
# Accept 200 or 201; clean up regardless
try:
assert r.status_code in (200, 201), (
f"Expected 200/201 for DNS record creation, got {r.status_code}: {r.text}"
)
finally:
delete('/api/dns/records', json={'zone': 'cell', 'name': _TEST_DNS_HOSTNAME, 'record_type': 'A'})
def test_add_and_delete_dns_record_round_trip(self):
"""Create a record, verify it appears in the list, then delete it."""
add_r = post('/api/dns/records', json={
'zone': 'cell',
'name': _TEST_DNS_HOSTNAME,
'record_type': 'A',
'value': '10.0.0.98',
})
assert add_r.status_code in (200, 201), (
f"Could not create test DNS record: {add_r.text}"
)
try:
records = get('/api/dns/records').json()
if isinstance(records, list):
names = [r.get('name', r.get('hostname', '')) for r in records]
else:
names = list(records.keys())
assert any(_TEST_DNS_HOSTNAME in n for n in names), (
f"Added record '{_TEST_DNS_HOSTNAME}' not found in records: {records}"
)
finally:
del_r = delete('/api/dns/records', json={'zone': 'cell', 'name': _TEST_DNS_HOSTNAME, 'record_type': 'A'})
assert del_r.status_code in (200, 204), (
f"DNS record delete failed: {del_r.status_code} {del_r.text}"
)
def test_delete_nonexistent_dns_record_does_not_crash(self):
"""Deleting a record that doesn't exist should return 200/404, not 500."""
r = delete('/api/dns/records', json={'zone': 'cell', 'name': 'does-not-exist-xyz', 'record_type': 'A'})
assert r.status_code in (200, 404), (
f"Unexpected status {r.status_code} deleting non-existent DNS record"
)
def test_add_dns_record_missing_name_is_handled(self):
"""Omitting required fields should not cause an unhandled 500."""
r = post('/api/dns/records', json={'zone': 'cell', 'record_type': 'A', 'value': '10.0.0.97'})
assert r.status_code != 200 or 'error' in r.json()
# ---------------------------------------------------------------------------
# GET /api/dhcp/leases
# ---------------------------------------------------------------------------
class TestDhcpLeases:
def test_get_dhcp_leases_returns_200(self):
r = get('/api/dhcp/leases')
assert r.status_code == 200
def test_get_dhcp_leases_returns_list_or_dict(self):
data = get('/api/dhcp/leases').json()
assert isinstance(data, (list, dict))
# ---------------------------------------------------------------------------
# POST /api/dhcp/reservations + DELETE /api/dhcp/reservations
# ---------------------------------------------------------------------------
_TEST_MAC = 'de:ad:be:ef:11:22'
_TEST_RESERVATION_IP = '10.0.0.200'
class TestDhcpReservations:
def _cleanup(self):
delete('/api/dhcp/reservations', json={'mac': _TEST_MAC})
def test_add_dhcp_reservation_returns_non_error(self):
try:
r = post('/api/dhcp/reservations', json={
'mac': _TEST_MAC,
'ip': _TEST_RESERVATION_IP,
'hostname': 'inttest-dhcp-host',
})
assert r.status_code in (200, 201), (
f"Expected 200/201 for DHCP reservation, got {r.status_code}: {r.text}"
)
finally:
self._cleanup()
def test_add_dhcp_reservation_missing_mac_returns_400(self):
r = post('/api/dhcp/reservations', json={'ip': _TEST_RESERVATION_IP})
assert r.status_code == 400
assert 'error' in r.json()
def test_add_dhcp_reservation_missing_ip_returns_400(self):
r = post('/api/dhcp/reservations', json={'mac': _TEST_MAC})
assert r.status_code == 400
assert 'error' in r.json()
def test_add_dhcp_reservation_empty_body_returns_400(self):
r = requests.post(
f"{API_BASE}/api/dhcp/reservations",
data='',
headers={'Content-Type': 'application/json'},
)
assert r.status_code == 400
def test_delete_dhcp_reservation_missing_mac_returns_400(self):
r = delete('/api/dhcp/reservations', json={})
assert r.status_code == 400
assert 'error' in r.json()
def test_add_and_delete_dhcp_reservation_round_trip(self):
add_r = post('/api/dhcp/reservations', json={
'mac': _TEST_MAC,
'ip': _TEST_RESERVATION_IP,
})
assert add_r.status_code in (200, 201), (
f"Could not create DHCP reservation: {add_r.text}"
)
try:
del_r = delete('/api/dhcp/reservations', json={'mac': _TEST_MAC})
assert del_r.status_code in (200, 204), (
f"DHCP reservation delete failed: {del_r.status_code} {del_r.text}"
)
except Exception:
self._cleanup()
raise
# ---------------------------------------------------------------------------
# GET /api/ntp/status + GET /api/network/info
# ---------------------------------------------------------------------------
class TestNtpAndNetworkInfo:
def test_ntp_status_returns_200(self):
r = get('/api/ntp/status')
assert r.status_code == 200
def test_ntp_status_is_dict(self):
assert isinstance(get('/api/ntp/status').json(), dict)
def test_network_info_returns_200(self):
r = get('/api/network/info')
assert r.status_code == 200
def test_network_info_is_dict(self):
assert isinstance(get('/api/network/info').json(), dict)