kcg-ai-monitoring/prediction/tests/test_dark_suspicion_params.py

"""Phase 2 PoC #1 — dark_suspicion params 외부화 동치성 테스트.

이 파일은 pandas 미설치 환경에서도 실행 가능하도록 구성한다.
`_merge_default_params` 와 DEFAULT_PARAMS 상수 자체만 단독 검증.

`compute_dark_suspicion` 전체 E2E 는 pandas 가 설치된 prediction 환경에서
수동으로 한 사이클 실행하여 신·구 diff=0 을 확인한다 (seed SQL 안내 참조).
"""
from __future__ import annotations

import importlib
import json
import os
import sys
import types
import unittest

# pandas 미설치 환경 우회 — algorithms.dark_vessel 이 pandas 를 top-level import
# 하므로, 그 import 를 stub 으로 대체해 DEFAULT_PARAMS 와 _merge_default_params 만
# 추출한다.
if 'pandas' not in sys.modules:
    pd_stub = types.ModuleType('pandas')
    pd_stub.DataFrame = type('DataFrame', (), {})  # annotation 용 dummy
    pd_stub.Timestamp = type('Timestamp', (), {})
    sys.modules['pandas'] = pd_stub

# pydantic_settings stub (다른 테스트와 동일 관용)
if 'pydantic_settings' not in sys.modules:
    stub = types.ModuleType('pydantic_settings')

    class _S:
        def __init__(self, **kw):
            for name, value in self.__class__.__dict__.items():
                if name.isupper():
                    setattr(self, name, kw.get(name, value))

    stub.BaseSettings = _S
    sys.modules['pydantic_settings'] = stub

# algorithms.location 도 top-level 의 haversine_nm import 가 있으므로 stub
if 'algorithms' not in sys.modules:
    pkg = types.ModuleType('algorithms')
    pkg.__path__ = [os.path.join(os.path.dirname(__file__), '..', 'algorithms')]
    sys.modules['algorithms'] = pkg

if 'algorithms.location' not in sys.modules:
    loc = types.ModuleType('algorithms.location')
    loc.haversine_nm = lambda a, b, c, d: 0.0  # pragma: no cover
    sys.modules['algorithms.location'] = loc

# 이제 dark_vessel 의 DEFAULT_PARAMS 와 _merge_default_params 만 조용히 import
dv = importlib.import_module('algorithms.dark_vessel')


class DarkSuspicionParamsTest(unittest.TestCase):

    def test_default_params_shape(self):
        """DEFAULT_PARAMS 는 11개 패턴 + tier_thresholds + sog_thresholds 를 포함한다."""
        p = dv.DARK_SUSPICION_DEFAULT_PARAMS
        self.assertIn('weights', p)
        self.assertIn('tier_thresholds', p)
        self.assertEqual(p['tier_thresholds'], {'critical': 70, 'high': 50, 'watch': 30})
        # 11 패턴 기본 가중치 키
        weights = p['weights']
        for key in [
            'P1_moving_off', 'P1_slow_moving_off',
            'P2_sensitive_zone', 'P2_special_zone',
            'P3_repeat_high', 'P3_repeat_low', 'P3_recent_dark',
            'P4_distance_anomaly',
            'P5_daytime_fishing_off',
            'P6_teleport_before_gap',
            'P7_unpermitted',
            'P8_very_long_gap', 'P8_long_gap',
            'P9_fishing_vessel_dark', 'P9_cargo_natural_gap',
            'P10_underway_deliberate', 'P10_anchored_natural',
            'P11_heading_cog_mismatch',
            'out_of_coverage',
        ]:
            self.assertIn(key, weights, f'weights.{key} missing')

    def test_merge_none_returns_default_reference(self):
        """params=None 이면 DEFAULT 그대로 사용 (Phase 2 이전과 동일 동작)."""
        self.assertIs(dv._merge_default_params(None), dv.DARK_SUSPICION_DEFAULT_PARAMS)

    def test_merge_empty_dict_returns_default_equivalent(self):
        """params={} 면 DEFAULT 와 key-level 완전 동일."""
        merged = dv._merge_default_params({})
        self.assertEqual(merged, dv.DARK_SUSPICION_DEFAULT_PARAMS)

    def test_merge_override_replaces_only_given_keys(self):
        """override 는 해당 key 만 교체, 나머지는 DEFAULT 유지."""
        override = {'tier_thresholds': {'critical': 80}}
        merged = dv._merge_default_params(override)
        # critical 만 교체됨
        self.assertEqual(merged['tier_thresholds']['critical'], 80)
        # high/watch 는 DEFAULT 유지
        self.assertEqual(merged['tier_thresholds']['high'], 50)
        self.assertEqual(merged['tier_thresholds']['watch'], 30)
        # weights 같은 다른 최상위 키는 DEFAULT 유지
        self.assertEqual(
            merged['weights']['P1_moving_off'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['weights']['P1_moving_off'],
        )
        # override 가 DEFAULT 를 변조하지 않는다 (불변성)
        self.assertEqual(
            dv.DARK_SUSPICION_DEFAULT_PARAMS['tier_thresholds']['critical'], 70,
        )

    def test_seed_sql_values_match_python_default(self):
        """seed SQL 의 params JSONB 가 Python DEFAULT 와 1:1 일치하는지 정적 검증."""
        seed_path = os.path.join(
            os.path.dirname(__file__), '..',
            'models_core', 'seeds', 'v1_dark_suspicion.sql',
        )
        with open(seed_path, 'r', encoding='utf-8') as f:
            sql = f.read()

        # $json$...$json$ 블록에서 JSON 추출
        start = sql.index('$json$') + len('$json$')
        end = sql.index('$json$', start)
        raw = sql[start:end].strip()
        params = json.loads(raw)

        self.assertEqual(
            params['tier_thresholds'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['tier_thresholds'],
        )
        self.assertEqual(
            params['weights'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['weights'],
        )
        self.assertEqual(
            params['sog_thresholds'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['sog_thresholds'],
        )
        self.assertEqual(
            params['repeat_thresholds'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['repeat_thresholds'],
        )
        self.assertEqual(
            params['gap_min_thresholds'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['gap_min_thresholds'],
        )
        self.assertEqual(
            params['heading_cog_mismatch_deg'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['heading_cog_mismatch_deg'],
        )
        self.assertEqual(
            params['p4_distance_multiplier'],
            dv.DARK_SUSPICION_DEFAULT_PARAMS['p4_distance_multiplier'],
        )
        self.assertEqual(
            list(params['p5_daytime_range']),
            list(dv.DARK_SUSPICION_DEFAULT_PARAMS['p5_daytime_range']),
        )


if __name__ == '__main__':
    unittest.main()