planet/backend/tests/test_vessel_enrichment.py

"""Tests for v5 enrichment + conflict promote-to-rule."""

from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock

import pytest

from app.models.vessel import AISConflictRecord, AISRawObservation
from app.models.vessel_enrichment import VesselMediaEnrichment, VesselProfileEnrichment
from app.services.vessel_ais_aggregation import aggregate_vessel_observations
from app.services.vessel_enrichment import (
    _apply_upsert,
    get_vessel_enrichment_bundle,
)


class _StoreSession:
    """Minimal AsyncSession stand-in that tracks mmsi-keyed enrichment + a strategy."""

    def __init__(self, *, profile=None, media=None, conflicts=None):
        self.profile = profile
        self.media = media
        self.conflicts = list(conflicts or [])
        self.added: list = []
        self.committed = False

    async def get(self, model, key):
        if model is VesselProfileEnrichment:
            return self.profile if self.profile and self.profile.mmsi == key else None
        if model is VesselMediaEnrichment:
            return self.media if self.media and self.media.mmsi == key else None
        return None


@pytest.mark.asyncio
async def test_enrichment_bundle_filters_expired_records():
    now = datetime.now(timezone.utc)
    fresh = VesselProfileEnrichment(
        mmsi=257123000,
        source="local_cache",
        payload={"vessel_subtype": "Container"},
        fetched_at=now - timedelta(hours=1),
        expires_at=now + timedelta(days=7),
        confidence=0.9,
    )
    expired_media = VesselMediaEnrichment(
        mmsi=257123000,
        source="vesselfinder",
        payload={"images": ["https://example.com/a.jpg"]},
        fetched_at=now - timedelta(days=30),
        expires_at=now - timedelta(days=1),
    )
    db = _StoreSession(profile=fresh, media=expired_media)

    bundle = await get_vessel_enrichment_bundle(db, 257123000)

    assert bundle["profile"]["payload"]["vessel_subtype"] == "Container"
    assert bundle["media"] is None


def test_apply_upsert_preserves_payload_and_metadata():
    record = VesselProfileEnrichment(mmsi=257123000)
    out = _apply_upsert(
        record,
        {
            "source": "vesselfinder",
            "payload": {"vessel_subtype": "Container", "operator": "Maersk"},
            "expires_at": "2026-12-31T00:00:00Z",
            "confidence": 0.85,
            "reference_url": "https://www.vesselfinder.com/vessels/257123000",
        },
    )
    assert out["payload"]["operator"] == "Maersk"
    assert out["confidence"] == 0.85
    assert record.reference_url == "https://www.vesselfinder.com/vessels/257123000"
    assert record.expires_at is not None
    assert record.expires_at.year == 2026


def _obs(*, source: str, mmsi: int, observed_at, **payload) -> AISRawObservation:
    payload = {"mmsi": mmsi, "lat": 60.0, "lon": 5.0, **payload}
    delivery_mode = "realtime_stream" if source == "aisstream_vessels" else "polling"
    transport = "websocket" if source == "aisstream_vessels" else "http"
    return AISRawObservation(
        target_schema="vessel_ais",
        source=source,
        entity_key=str(mmsi),
        delivery_mode=delivery_mode,
        transport=transport,
        message_type="PositionReport",
        observation_hash=f"{source}:{mmsi}:{observed_at.isoformat()}",
        observed_at=observed_at,
        collected_at=observed_at,
        normalized_payload=payload,
        raw_payload=payload,
        quality_flags=[],
    )


@pytest.mark.asyncio
async def test_promoted_rule_wins_during_aggregation():
    """Simulate the strategy that conflict-promote-to-rule writes."""
    now = datetime.now(timezone.utc)
    obs_a = _obs(
        source="aisstream_vessels",
        mmsi=257111000,
        observed_at=now,
        name="STREAM NAME",
        vessel_type_name="Cargo",
    )
    obs_b = _obs(
        source="barentswatch_vessels",
        mmsi=257111000,
        observed_at=now - timedelta(seconds=1),
        name="REST NAME",
        vessel_type_name="Cargo",
    )
    promoted_strategy = {
        "version": 99,
        "vessel_ais": {
            "source_priority": [],
            "field_rules": {
                "name": {"mode": "source_priority", "source_priority": ["barentswatch_vessels"]}
            },
            "freshness": {"realtime_stream_seconds": 0, "polling_seconds": 0},
            "allow_dynamic_lock": False,
        },
    }

    db = AsyncMock()
    vessels = await aggregate_vessel_observations(
        db,
        [obs_a, obs_b],
        write_conflicts=False,
        strategy=promoted_strategy,
    )
    assert vessels[0]["name"] == "REST NAME"
    assert vessels[0]["selected_reasons"]["name"] == "source_priority"
    assert vessels[0]["aggregation_strategy_version"] == 99


def test_conflict_record_holds_selected_source():
    """Sanity: the promote-to-rule API reads selected_source from this column."""
    record = AISConflictRecord(
        target_schema="vessel_ais",
        entity_key="257111000",
        field="name",
        candidates={"a": "X", "b": "Y"},
        selected_source="barentswatch_vessels",
        selected_value="Y",
        selected_reason="delivery_mode_priority",
    )
    serialized = record.to_dict()
    assert serialized["selected_source"] == "barentswatch_vessels"
    assert serialized["field"] == "name"