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

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

pandas 미설치 환경을 우회하기 위해 dark_suspicion 테스트와 동일한 stub 패턴 사용.
"""
from __future__ import annotations

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

# pandas stub (annotation 용)
if 'pandas' not in sys.modules:
    pd_stub = types.ModuleType('pandas')
    pd_stub.DataFrame = type('DataFrame', (), {})
    pd_stub.Timestamp = type('Timestamp', (), {})
    sys.modules['pandas'] = pd_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

if 'algorithms' not in sys.modules:
    pkg = types.ModuleType('algorithms')
    pkg.__path__ = [os.path.join(os.path.dirname(__file__), '..', 'algorithms')]
    sys.modules['algorithms'] = pkg

gv = importlib.import_module('algorithms.gear_violation')


class GearViolationParamsTest(unittest.TestCase):

    def test_default_params_shape(self):
        p = gv.GEAR_VIOLATION_DEFAULT_PARAMS
        self.assertIn('scores', p)
        self.assertIn('signal_cycling', p)
        self.assertIn('gear_drift_threshold_nm', p)
        self.assertIn('fixed_gear_types', p)
        self.assertIn('fishery_code_allowed_gear', p)
        # 6 G-codes 점수 키 전부 있는지
        for k in ['G01_zone_violation', 'G02_closed_season', 'G03_unregistered_gear',
                  'G04_signal_cycling', 'G05_gear_drift', 'G06_pair_trawl']:
            self.assertIn(k, p['scores'])

    def test_default_values_match_module_constants(self):
        """DEFAULT_PARAMS 는 모듈 레벨 상수와 완전히 동일해야 한다 (SSOT 이중성 방지)."""
        p = gv.GEAR_VIOLATION_DEFAULT_PARAMS
        self.assertEqual(p['scores']['G01_zone_violation'], gv.G01_SCORE)
        self.assertEqual(p['scores']['G02_closed_season'], gv.G02_SCORE)
        self.assertEqual(p['scores']['G03_unregistered_gear'], gv.G03_SCORE)
        self.assertEqual(p['scores']['G04_signal_cycling'], gv.G04_SCORE)
        self.assertEqual(p['scores']['G05_gear_drift'], gv.G05_SCORE)
        self.assertEqual(p['scores']['G06_pair_trawl'], gv.G06_SCORE)
        self.assertEqual(p['signal_cycling']['gap_min'], gv.SIGNAL_CYCLING_GAP_MIN)
        self.assertEqual(p['signal_cycling']['min_count'], gv.SIGNAL_CYCLING_MIN_COUNT)
        self.assertAlmostEqual(p['gear_drift_threshold_nm'], gv.GEAR_DRIFT_THRESHOLD_NM)
        self.assertEqual(set(p['fixed_gear_types']), gv.FIXED_GEAR_TYPES)
        # fishery_code_allowed_gear: list ↔ set 변환 후 비교
        for key, allowed in gv.FISHERY_CODE_ALLOWED_GEAR.items():
            self.assertEqual(set(p['fishery_code_allowed_gear'][key]), allowed)

    def test_merge_none_returns_default_reference(self):
        self.assertIs(gv._merge_default_gv_params(None), gv.GEAR_VIOLATION_DEFAULT_PARAMS)

    def test_merge_override_replaces_only_given_keys(self):
        override = {'scores': {'G06_pair_trawl': 99}}
        merged = gv._merge_default_gv_params(override)
        self.assertEqual(merged['scores']['G06_pair_trawl'], 99)
        # 다른 score 는 DEFAULT 유지
        self.assertEqual(
            merged['scores']['G01_zone_violation'],
            gv.GEAR_VIOLATION_DEFAULT_PARAMS['scores']['G01_zone_violation'],
        )
        # fixed_gear_types 같은 top-level 키도 DEFAULT 유지
        self.assertEqual(
            merged['fixed_gear_types'],
            gv.GEAR_VIOLATION_DEFAULT_PARAMS['fixed_gear_types'],
        )
        # DEFAULT 는 변조되지 않음
        self.assertEqual(
            gv.GEAR_VIOLATION_DEFAULT_PARAMS['scores']['G06_pair_trawl'], 20,
        )

    def test_seed_sql_values_match_python_default(self):
        """seed SQL JSONB ↔ Python DEFAULT 1:1 정적 검증."""
        seed_path = os.path.join(
            os.path.dirname(__file__), '..',
            'models_core', 'seeds', 'v1_gear_violation.sql',
        )
        with open(seed_path, 'r', encoding='utf-8') as f:
            sql = f.read()

        start = sql.index('$json$') + len('$json$')
        end = sql.index('$json$', start)
        raw = sql[start:end].strip()
        params = json.loads(raw)

        expected = gv.GEAR_VIOLATION_DEFAULT_PARAMS
        self.assertEqual(params['scores'], expected['scores'])
        self.assertEqual(params['signal_cycling'], expected['signal_cycling'])
        self.assertAlmostEqual(
            params['gear_drift_threshold_nm'], expected['gear_drift_threshold_nm']
        )
        # list 는 순서 무관하게 set 비교 (DB 에 저장 시 어떤 순서든 상관 없음)
        self.assertEqual(set(params['fixed_gear_types']),
                         set(expected['fixed_gear_types']))
        for code, allowed in expected['fishery_code_allowed_gear'].items():
            self.assertEqual(
                set(params['fishery_code_allowed_gear'][code]), set(allowed),
                f'fishery_code_allowed_gear[{code}] mismatch',
            )


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