mqtt-broker-stack/backend/tests/test_configgen.py

"""Config generator unit tests (spec section 64: configuration generation)."""

from __future__ import annotations

import pytest

from app.configgen import generate_config
from app.models import BrokerSettings, ListenerConfig


@pytest.fixture()
def session(db_session):
    yield db_session


def test_generate_with_one_listener(session):
    session.add(
        ListenerConfig(
            name="Production MQTT",
            port=1883,
            protocol="mqtt",
            allow_anonymous=False,
            max_connections=-1,
            tls_enabled=False,
            require_certificate=False,
            use_identity_as_username=False,
        )
    )
    session.commit()

    result = generate_config(session)
    assert "listener 1883" in result.config_text
    assert "global_plugin /usr/lib/mosquitto_dynamic_security.so" in result.config_text
    assert "allow_anonymous false" in result.config_text
    # Must NOT use options deprecated in 2.1 (spec 3.1).
    assert "per_listener_settings" not in result.config_text
    assert result.warnings == []


def test_generate_includes_global_settings(session):
    session.add(ListenerConfig(name="l", port=1883, protocol="mqtt"))
    session.add(BrokerSettings(key="max_packet_size", value="2000000", value_type="int"))
    session.commit()

    result = generate_config(session)
    assert "max_packet_size 2000000" in result.config_text


def test_generate_with_empty_model_warns_instead_of_raising(session):
    """Fresh-platform state: preview must work, not 500 (config page fix)."""
    result = generate_config(session)
    assert "global_plugin /usr/lib/mosquitto_dynamic_security.so" in result.config_text
    assert "persistence true" in result.config_text
    assert any("No enabled listeners" in w for w in result.warnings)


def test_disabled_listeners_are_omitted(session):
    session.add(ListenerConfig(name="off", port=1884, protocol="mqtt", enabled=False))
    session.add(ListenerConfig(name="on", port=1883, protocol="mqtt"))
    session.commit()

    result = generate_config(session)
    assert "listener 1883" in result.config_text
    assert "listener 1884" not in result.config_text


def test_unpublished_port_listener_warns(session):
    """Listeners on ports not published by compose warn: they are only
    reachable inside the Docker network (spec section 5)."""
    session.add(ListenerConfig(name="internal-only", port=8884, protocol="mqtt"))
    session.commit()

    result = generate_config(session)
    assert any(
        "not published to the host" in w and "8884" in w for w in result.warnings
    )


def test_published_port_listener_does_not_warn(session):
    session.add(ListenerConfig(name="prod", port=8883, protocol="mqtt"))
    session.commit()

    result = generate_config(session)
    assert not any("not published to the host" in w for w in result.warnings)


def test_tls_listener_without_pki_warns(session):
    session.add(
        ListenerConfig(
            name="tls-later",
            port=8883,
            protocol="mqtt",
            tls_enabled=True,
            tls_min_version="tlsv1.2",
        )
    )
    session.commit()

    result = generate_config(session)
    assert any("no certificate is bound" in w for w in result.warnings)
    assert "tls_version tlsv1.2" in result.config_text


def test_tls_listener_with_bound_certificate_emits_certfile(session):
    """With PKI material bound, the generator emits real cert paths (spec 30)."""
    from datetime import UTC, datetime

    from app.crypto import encrypt_secret
    from app.models import Certificate
    from app.pki import create_ca, issue_server_certificate

    ca_issued = create_ca(
        common_name="Test CA", validity_days=3650, key_algorithm="ecdsa-p256"
    )
    ca = Certificate(
        common_name="Test CA",
        certificate_pem=ca_issued.certificate_pem,
        private_key_pem=encrypt_secret(ca_issued.private_key_pem),
        expiry_date=ca_issued.not_valid_after,
        is_ca=True,
        key_algorithm="ecdsa-p256",
        serial_number=str(ca_issued.serial_number),
        dns_sans=[],
        ip_sans=[],
        fingerprint="aa:bb",
        status="issued",
    )
    session.add(ca)
    session.flush()

    leaf = issue_server_certificate(
        ca_certificate_pem=ca_issued.certificate_pem,
        ca_private_key_pem=ca_issued.private_key_pem,
        common_name="broker.local",
        validity_days=365,
    )
    cert = Certificate(
        common_name="broker.local",
        certificate_pem=leaf.certificate_pem,
        private_key_pem=encrypt_secret(leaf.private_key_pem),
        expiry_date=leaf.not_valid_after,
        is_ca=False,
        parent_id=ca.id,
        key_algorithm="ecdsa-p256",
        serial_number=str(leaf.serial_number),
        dns_sans=["broker.local"],
        ip_sans=[],
        fingerprint="cc:dd",
        status="issued",
    )
    session.add(cert)
    session.flush()

    session.add(
        ListenerConfig(
            name="tls-prod",
            port=8883,
            protocol="mqtt",
            tls_enabled=True,
            tls_min_version="tlsv1.2",
            cert_id=cert.id,
        )
    )
    session.commit()

    result = generate_config(session)
    text = result.config_text
    assert f"certfile /mosquitto/certs/{cert.id}.crt" in text
    assert f"keyfile /mosquitto/certs/{cert.id}.key" in text
    assert f"cafile /mosquitto/certs/{ca.id}.crt" in text
    assert "tls_version tlsv1.2" in text
    assert result.warnings == []