pic/tests/test_port_conflicts.py

"""
Unit tests for api/port_registry.py — port conflict detection.
"""
import sys
import os
import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api'))

from port_registry import PORT_FIELDS, detect_conflicts


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_config(**sections):
    """Build a minimal effective_config dict from keyword args."""
    return dict(sections)


# ---------------------------------------------------------------------------
# No-conflict cases
# ---------------------------------------------------------------------------

class TestNoConflict:

    def test_empty_config_and_patch(self):
        """Both inputs empty → no conflicts."""
        assert detect_conflicts({}, {}) == []

    def test_all_different_ports(self):
        effective = {
            'network':   {'dns_port': 53},
            'wireguard': {'port': 51820},
            'email':     {'smtp_port': 25, 'submission_port': 587,
                          'imap_port': 993, 'webmail_port': 8888},
            'calendar':  {'port': 5232},
            'files':     {'port': 8080, 'manager_port': 8082},
        }
        assert detect_conflicts(effective, {}) == []

    def test_patch_with_unique_port(self):
        """Updating a port to a value nobody else uses → no conflict."""
        effective = {
            'network':   {'dns_port': 53},
            'wireguard': {'port': 51820},
        }
        patch = {'wireguard': {'port': 9999}}
        assert detect_conflicts(effective, patch) == []

    def test_missing_sections_are_ignored(self):
        """Sections absent from both config and patch are silently skipped."""
        # Only 'network' is present; others are absent entirely.
        effective = {'network': {'dns_port': 53}}
        assert detect_conflicts(effective, {}) == []

    def test_none_and_empty_string_values_are_skipped(self):
        """None or '' port values must not be included in conflict detection."""
        effective = {
            'network':   {'dns_port': None},
            'wireguard': {'port': ''},
            'calendar':  {'port': 5232},
        }
        # No actual usable ports clash → no conflict
        assert detect_conflicts(effective, {}) == []


# ---------------------------------------------------------------------------
# Conflict detection
# ---------------------------------------------------------------------------

class TestConflictDetected:

    def test_two_sections_same_port(self):
        """Two sections sharing a port must produce one conflict entry."""
        effective = {
            'network':   {'dns_port': 5232},
            'calendar':  {'port': 5232},
        }
        result = detect_conflicts(effective, {})
        assert len(result) == 1
        assert result[0]['port'] == 5232
        slots = result[0]['conflicts']
        assert ('network', 'dns_port') in slots
        assert ('calendar', 'port') in slots

    def test_three_sections_same_port(self):
        """Three sections sharing a port → one conflict entry with 3 slots."""
        effective = {
            'network':   {'dns_port': 8080},
            'calendar':  {'port': 8080},
            'files':     {'port': 8080, 'manager_port': 9000},
        }
        result = detect_conflicts(effective, {})
        assert len(result) == 1
        assert result[0]['port'] == 8080
        assert len(result[0]['conflicts']) == 3

    def test_two_separate_conflicts(self):
        """Two distinct port values each shared by two sections."""
        effective = {
            'network':   {'dns_port': 53},
            'wireguard': {'port': 53},       # conflict on 53
            'calendar':  {'port': 8080},
            'files':     {'port': 8080},     # conflict on 8080
        }
        result = detect_conflicts(effective, {})
        ports_with_conflict = {c['port'] for c in result}
        assert 53 in ports_with_conflict
        assert 8080 in ports_with_conflict
        assert len(result) == 2

    def test_email_fields_conflict_with_other_section(self):
        """An email sub-port conflicting with another section."""
        effective = {
            'email':     {'smtp_port': 25, 'submission_port': 5232,
                          'imap_port': 993, 'webmail_port': 8888},
            'calendar':  {'port': 5232},
        }
        result = detect_conflicts(effective, {})
        assert len(result) == 1
        assert result[0]['port'] == 5232
        slots = result[0]['conflicts']
        assert ('email', 'submission_port') in slots
        assert ('calendar', 'port') in slots


# ---------------------------------------------------------------------------
# Patch overrides stored config
# ---------------------------------------------------------------------------

class TestPatchOverride:

    def test_patch_resolves_existing_conflict(self):
        """If the patch moves a port away from a conflict, no conflict remains."""
        effective = {
            'network':   {'dns_port': 5232},
            'calendar':  {'port': 5232},
        }
        # Patch moves calendar to a free port
        patch = {'calendar': {'port': 9000}}
        assert detect_conflicts(effective, patch) == []

    def test_patch_introduces_conflict(self):
        """If the patch sets a port that collides with stored config, detect it."""
        effective = {
            'network':   {'dns_port': 53},
            'calendar':  {'port': 5232},
        }
        # Patch changes calendar port to match DNS
        patch = {'calendar': {'port': 53}}
        result = detect_conflicts(effective, patch)
        assert len(result) == 1
        assert result[0]['port'] == 53
        slots = result[0]['conflicts']
        assert ('network', 'dns_port') in slots
        assert ('calendar', 'port') in slots

    def test_patch_partial_section_merges_with_stored(self):
        """A partial patch for a section merges with stored fields (not replaces)."""
        effective = {
            'email': {
                'smtp_port': 25,
                'submission_port': 587,
                'imap_port': 993,
                'webmail_port': 8888,
            },
            'calendar': {'port': 5232},
        }
        # Patch only changes imap_port; other email ports remain from stored config
        patch = {'email': {'imap_port': 5232}}
        result = detect_conflicts(effective, patch)
        assert len(result) == 1
        assert result[0]['port'] == 5232
        slots = result[0]['conflicts']
        assert ('email', 'imap_port') in slots
        assert ('calendar', 'port') in slots

    def test_patch_only_affects_patched_section(self):
        """Fields NOT in the patch are still read from effective_config."""
        effective = {
            'wireguard': {'port': 51820},
            'files':     {'port': 8080, 'manager_port': 8082},
        }
        # Patch changes files.manager_port but leaves files.port alone
        patch = {'files': {'manager_port': 51820}}
        result = detect_conflicts(effective, patch)
        assert len(result) == 1
        assert result[0]['port'] == 51820
        slots = result[0]['conflicts']
        assert ('wireguard', 'port') in slots
        assert ('files', 'manager_port') in slots


# ---------------------------------------------------------------------------
# Self-conflict: same (section, field) should not flag itself
# ---------------------------------------------------------------------------

class TestNoSelfConflict:

    def test_same_field_in_effective_and_patch_no_duplicate(self):
        """
        When the patch sets the same value as the stored config for the same
        (section, field), there must be no self-conflict.
        """
        effective = {'calendar': {'port': 5232}}
        patch = {'calendar': {'port': 5232}}  # same value, same slot
        assert detect_conflicts(effective, patch) == []

    def test_only_one_section_one_field(self):
        """A single unique port cannot conflict with itself."""
        effective = {'network': {'dns_port': 53}}
        patch = {'network': {'dns_port': 53}}
        assert detect_conflicts(effective, patch) == []


# ---------------------------------------------------------------------------
# Real-world default ports from PORT_DEFAULTS in ip_utils.py
# ---------------------------------------------------------------------------

class TestRealWorldDefaults:

    DEFAULT_CONFIG = {
        'network':   {'dns_port': 53},
        'wireguard': {'port': 51820},
        'email':     {'smtp_port': 25, 'submission_port': 587,
                      'imap_port': 993, 'webmail_port': 8888},
        'calendar':  {'port': 5232},
        'files':     {'port': 8080, 'manager_port': 8082},
    }

    def test_defaults_have_no_conflicts(self):
        """All out-of-the-box defaults must be conflict-free."""
        assert detect_conflicts(self.DEFAULT_CONFIG, {}) == []

    def test_changing_wireguard_to_dns_port_conflicts(self):
        patch = {'wireguard': {'port': 53}}
        result = detect_conflicts(self.DEFAULT_CONFIG, patch)
        assert len(result) == 1
        assert result[0]['port'] == 53

    def test_changing_files_port_to_calendar_port_conflicts(self):
        patch = {'files': {'port': 5232}}
        result = detect_conflicts(self.DEFAULT_CONFIG, patch)
        assert len(result) == 1
        assert result[0]['port'] == 5232

    def test_integer_string_ports_are_treated_as_ints(self):
        """Port values supplied as strings (as from JSON) must still work."""
        effective = {
            'network':  {'dns_port': '53'},
            'calendar': {'port': '53'},
        }
        result = detect_conflicts(effective, {})
        assert len(result) == 1
        assert result[0]['port'] == 53

    def test_non_integer_port_values_are_skipped(self):
        """Malformed values that can't be cast to int must not crash."""
        effective = {
            'network':  {'dns_port': 'bogus'},
            'calendar': {'port': 5232},
        }
        assert detect_conflicts(effective, {}) == []