geointel/backend/tests/test_mdk_bathymetry_acquisition.py

from __future__ import annotations

import io
from pathlib import Path
from uuid import uuid4

import pytest

from app.core.config import Settings
from app.schemas.bathymetry import MdkBathymetryAcquireRequest
from app.schemas.operations import VectorSelectionBBox
from app.services.mdk_bathymetry_acquisition_service import MdkBathymetryAcquisitionService

CAPABILITIES_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
<WCS_Capabilities version="1.0.0" xmlns="http://www.opengis.net/wcs">
  <ContentMetadata>
    <CoverageOfferingBrief>
      <name>depth_model_20m_lat</name>
      <label>Belgian Continental Shelf depth model</label>
    </CoverageOfferingBrief>
  </ContentMetadata>
</WCS_Capabilities>
"""


class FakeResponse:
    def __init__(self, content: bytes, content_type: str = "application/xml") -> None:
        self._stream = io.BytesIO(content)
        self.headers = {"Content-Type": content_type}

    def read(self, limit: int = -1) -> bytes:
        return self._stream.read(limit)

    def __enter__(self):
        return self

    def __exit__(self, *args):
        return False


def _payload(**overrides) -> MdkBathymetryAcquireRequest:
    values = {
        "bbox": VectorSelectionBBox(min_x=2.5, min_y=51.3, max_x=2.6, max_y=51.4),
        "force_refresh": True,
    }
    values.update(overrides)
    return MdkBathymetryAcquireRequest(**values)


def _settings(**overrides) -> Settings:
    values = {
        "mdk_bathymetry_acquisition_enabled": True,
        "mdk_bathymetry_coverage_id": "depth_model_20m_lat",
    }
    values.update(overrides)
    return Settings(**values)


def test_acquisition_fails_closed_when_disabled() -> None:
    settings = _settings(mdk_bathymetry_acquisition_enabled=False)

    with pytest.raises(Exception) as exc_info:
        MdkBathymetryAcquisitionService.acquire(None, uuid4(), _payload(), settings=settings)

    assert getattr(exc_info.value, "code", None) == "MDK_BATHYMETRY_ACQUISITION_DISABLED"


def test_acquisition_fails_closed_without_coverage_id() -> None:
    settings = _settings(mdk_bathymetry_coverage_id=None)

    with pytest.raises(Exception) as exc_info:
        MdkBathymetryAcquisitionService.acquire(None, uuid4(), _payload(), settings=settings)

    assert getattr(exc_info.value, "code", None) == "MDK_BATHYMETRY_COVERAGE_NOT_CONFIGURED"


def test_acquisition_rejects_oversized_bbox() -> None:
    settings = _settings(mdk_bathymetry_max_bbox_deg2=0.001)

    with pytest.raises(Exception) as exc_info:
        MdkBathymetryAcquisitionService.acquire(None, uuid4(), _payload(), settings=settings)

    assert getattr(exc_info.value, "code", None) == "MDK_BATHYMETRY_BBOX_TOO_LARGE"


def test_acquisition_requires_reachable_probe() -> None:
    settings = _settings()

    def failing_opener(request, timeout=None):
        raise OSError("connection refused")

    with pytest.raises(Exception) as exc_info:
        MdkBathymetryAcquisitionService.acquire(None, uuid4(), _payload(), settings=settings, opener=failing_opener)

    assert getattr(exc_info.value, "code", None) == "MDK_BATHYMETRY_ENDPOINT_NOT_READY"


def test_acquisition_requires_advertised_coverage_id() -> None:
    settings = _settings(mdk_bathymetry_coverage_id="not_advertised_coverage")

    def opener(request, timeout=None):
        return FakeResponse(CAPABILITIES_XML)

    with pytest.raises(Exception) as exc_info:
        MdkBathymetryAcquisitionService.acquire(None, uuid4(), _payload(), settings=settings, opener=opener)

    assert getattr(exc_info.value, "code", None) == "MDK_BATHYMETRY_COVERAGE_NOT_ADVERTISED"


def test_acquisition_rejects_non_geotiff_coverage_response() -> None:
    settings = _settings()
    responses = []

    def opener(request, timeout=None):
        url = request.full_url if hasattr(request, "full_url") else str(request)
        responses.append(url)
        if "GetCapabilities" in url:
            return FakeResponse(CAPABILITIES_XML)
        return FakeResponse(b"<ServiceExceptionReport>boom</ServiceExceptionReport>", "application/xml")

    with pytest.raises(Exception) as exc_info:
        MdkBathymetryAcquisitionService.acquire(None, uuid4(), _payload(), settings=settings, opener=opener)

    assert getattr(exc_info.value, "code", None) == "MDK_BATHYMETRY_INVALID_RESPONSE"
    assert any("GetCoverage" in url for url in responses)
    coverage_urls = [url for url in responses if "GetCoverage" in url]
    assert "coverage=depth_model_20m_lat" in coverage_urls[0]
    assert "format=GeoTIFF" in coverage_urls[0]


def test_get_coverage_url_is_bounded_and_pinned() -> None:
    settings = _settings()
    bbox = [2.5, 51.3, 2.6, 51.4]

    url = MdkBathymetryAcquisitionService._get_coverage_url(settings, "depth_model_20m_lat", bbox)

    assert url.startswith("https://")
    assert "request=GetCoverage" in url
    assert "version=1.0.0" in url
    assert "crs=EPSG%3A4326" in url or "crs=EPSG:4326" in url
    width, height = MdkBathymetryAcquisitionService._pixel_dimensions(bbox)
    assert 1 <= width <= MdkBathymetryAcquisitionService.MAX_PIXELS_PER_SIDE
    assert 1 <= height <= MdkBathymetryAcquisitionService.MAX_PIXELS_PER_SIDE


def test_source_module_never_disables_tls_verification() -> None:
    source = (
        Path(__file__).resolve().parents[1] / "app" / "services" / "mdk_bathymetry_acquisition_service.py"
    ).read_text(encoding="utf-8")

    assert "_create_unverified_context" not in source
    assert "CERT_NONE" not in source
    assert "check_hostname = False" not in source