service-finder/backend/app/tests/test_validation_engine.py

# /opt/docker/dev/service_finder/backend/app/tests/test_validation_engine.py
"""
Test script for ValidationEngine.

Tests:
    1. get_rules() — fetches dynamic rules from DB
    2. process_staging_score() with admin override
    3. process_staging_score() with user weight
    4. process_staging_score() with robot base score
    5. Auto-promotion when threshold reached

Usage:
    docker exec -i sf_api python -m app.tests.test_validation_engine
"""
import asyncio
import logging
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from app.database import AsyncSessionLocal
from app.models import (
    ServiceStaging,
    ValidationRule,
    UserStats,
)
from app.models.identity.social import ProviderValidation
from app.services.validation_engine import ValidationEngine
from sqlalchemy import select, delete

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s')
logger = logging.getLogger("TestValidationEngine")

PASS = 0
FAIL = 0


def assert_eq(actual, expected, label: str):
    global PASS, FAIL
    if actual == expected:
        PASS += 1
        logger.info(f"  ✅ {label}: {actual} == {expected}")
    else:
        FAIL += 1
        logger.error(f"  ❌ {label}: {actual} != {expected}")


async def test_get_rules():
    """Test that get_rules() fetches all 4 base rules from the DB."""
    logger.info("\n📋 Test: get_rules()")
    engine = ValidationEngine()
    async with AsyncSessionLocal() as db:
        rules = await engine.get_rules(db)
        assert_eq(rules.get("BOT_BASE_SCORE"), 20, "BOT_BASE_SCORE")
        assert_eq(rules.get("FIRST_ENTRY_SCORE"), 40, "FIRST_ENTRY_SCORE")
        assert_eq(rules.get("USER_CONFIRM_BASE"), 20, "USER_CONFIRM_BASE")
        assert_eq(rules.get("AUTO_APPROVE_THRESHOLD"), 80, "AUTO_APPROVE_THRESHOLD")
        assert_eq(len(rules), 4, "rule count")


async def test_admin_override():
    """Test that admin override sets score=100."""
    logger.info("\n📋 Test: admin override (score=100)")
    engine = ValidationEngine()
    async with AsyncSessionLocal() as db:
        staging = ServiceStaging(
            name="Test Admin Garage",
            city="Budapest",
            fingerprint="test_admin_fp_001",
            status="pending",
            validation_level=0,
        )
        db.add(staging)
        await db.flush()

        new_level = await engine.process_staging_score(
            db, staging, is_admin=True
        )
        assert_eq(new_level, 100, "admin validation_level")
        assert_eq(staging.validation_level, 100, "staging.validation_level")

        # Cleanup
        await db.rollback()


async def test_user_weight():
    """Test that user weight is calculated from UserStats.current_level."""
    logger.info("\n📋 Test: user weight calculation")
    engine = ValidationEngine()
    async with AsyncSessionLocal() as db:
        # Create a staging entry
        staging = ServiceStaging(
            name="Test User Garage",
            city="Debrecen",
            fingerprint="test_user_fp_001",
            status="pending",
            validation_level=0,
        )
        db.add(staging)
        await db.flush()

        # Test with user_id that doesn't exist (should use base weight)
        new_level = await engine.process_staging_score(
            db, staging, user_id=999999
        )
        # USER_CONFIRM_BASE=20, no UserStats -> base=20
        assert_eq(new_level, 20, "user weight without stats (base=20)")

        staging.validation_level = 0  # reset

        # Create UserStats for user 1 (if exists)
        stmt = select(UserStats).where(UserStats.user_id == 1)
        result = await db.execute(stmt)
        stats = result.scalar_one_or_none()
        if stats:
            level = stats.current_level or 1
            expected = 20 + (level * 5)
            new_level = await engine.process_staging_score(
                db, staging, user_id=1
            )
            assert_eq(new_level, expected, f"user weight with level={level}")

        await db.rollback()


async def test_robot_base_score():
    """Test that robot (no user_id) uses BOT_BASE_SCORE=20."""
    logger.info("\n📋 Test: robot base score (BOT_BASE_SCORE=20)")
    engine = ValidationEngine()
    async with AsyncSessionLocal() as db:
        staging = ServiceStaging(
            name="Test Robot Garage",
            city="Szeged",
            fingerprint="test_robot_fp_001",
            status="pending",
            validation_level=0,
        )
        db.add(staging)
        await db.flush()

        new_level = await engine.process_staging_score(db, staging)
        assert_eq(new_level, 20, "robot base score")
        assert_eq(staging.validation_level, 20, "staging.validation_level")

        await db.rollback()


async def test_auto_promote():
    """Test that staging is auto-promoted when validation_level >= 80."""
    logger.info("\n📋 Test: auto-promotion at threshold >= 80")
    engine = ValidationEngine()
    async with AsyncSessionLocal() as db:
        staging = ServiceStaging(
            name="Test Promote Garage",
            city="Miskolc",
            postal_code="3500",
            full_address="3500 Miskolc, Test Street 1",
            fingerprint="test_promote_fp_001",
            status="pending",
            validation_level=40,
            trust_score=20,
        )
        db.add(staging)
        await db.flush()

        # First pass: admin adds 40 -> 80 (threshold reached)
        new_level = await engine.process_staging_score(
            db, staging, is_admin=True
        )
        # Admin gives 100, but capped at 100
        assert_eq(new_level, 100, "auto-promote validation_level")
        # Status should be 'approved' now
        assert_eq(staging.status, "approved", "staging.status after promotion")

        await db.rollback()


async def main():
    logger.info("=" * 60)
    logger.info("🔍 ValidationEngine Test Suite")
    logger.info("=" * 60)

    await test_get_rules()
    await test_admin_override()
    await test_user_weight()
    await test_robot_base_score()
    await test_auto_promote()

    logger.info("\n" + "=" * 60)
    logger.info(f"📊 Results: {PASS} passed, {FAIL} failed")
    logger.info("=" * 60)

    if FAIL > 0:
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())