"""Source connection enforces TLS certificate verification (issue 06).

The extract connection previously passed an empty ssl dict, which encrypted the
connection but did not verify the server certificate or hostname — a MITM exposure.
``connect`` must hand PyMySQL a verifying SSL context (system CA bundle, CERT_REQUIRED,
hostname check) when TLS is on.

Runnable without a test harness::

    .venv/bin/python tests/test_source_tls.py
"""

from __future__ import annotations

import os
import pathlib
import ssl
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))

from ingestion.extract.tidbcloud.engine import source_db
from ingestion.extract.tidbcloud.engine.config import SourceConfig


def _connect_kwargs(dsn: str) -> dict:
    """Call connect() with pymysql.connect stubbed, returning the kwargs it would use."""
    captured: dict = {}
    original = source_db.pymysql.connect
    source_db.pymysql.connect = lambda **kwargs: captured.update(kwargs) or object()
    os.environ["TEST_SRC_DSN"] = dsn
    try:
        config = SourceConfig(
            path=pathlib.Path("source.yaml"),
            data={"source": {"name": "nexuscrm", "database": "nexuscrm", "connection_env": "TEST_SRC_DSN"}},
        )
        source_db.connect(config)
    finally:
        source_db.pymysql.connect = original
        os.environ.pop("TEST_SRC_DSN", None)
    return captured


def test_tls_context_verifies_cert_and_hostname() -> None:
    kwargs = _connect_kwargs("mysql://u:p@gateway.example.tidbcloud.com:4000/nexuscrm")
    context = kwargs.get("ssl")
    assert isinstance(context, ssl.SSLContext), f"expected an SSLContext, got {context!r}"
    assert context.verify_mode == ssl.CERT_REQUIRED, "server certificate must be verified"
    assert context.check_hostname is True, "hostname must be verified"


def test_tls_can_be_disabled_explicitly() -> None:
    kwargs = _connect_kwargs("mysql://u:p@localhost:4000/nexuscrm?ssl=false")
    assert "ssl" not in kwargs, "ssl=false should disable TLS (e.g. a local plaintext MySQL)"


if __name__ == "__main__":
    cases = [value for name, value in sorted(globals().items()) if name.startswith("test_") and callable(value)]
    for case in cases:
        case()
        print(f"ok  {case.__name__}")
    print(f"\n{len(cases)} passed")