bmc_hub/tests/test_migration_center.py

from datetime import datetime, timezone
from decimal import Decimal

import jwt
import pytest
from fastapi import HTTPException

from app.modules.migration_center.backend import router
from app.modules.migration_center.backend.service import (
    EconomicSnapshotRepository,
    _resolve_source_customer,
    preflight_token,
    snapshot_hash,
    verify_preflight,
)


def test_snapshot_hash_is_stable_for_key_order_and_decimal():
    first = snapshot_hash({"amount": Decimal("10.00"), "nested": {"b": 2, "a": 1}})
    second = snapshot_hash({"nested": {"a": 1, "b": 2}, "amount": Decimal("10.00")})
    assert first == second
    assert len(first) == 64


def test_normalizes_vtiger_subscription():
    item = router._normalized_crm_record(
        "vtiger",
        {
            "id": "72x123",
            "account_id": "3x44",
            "accountname": "Kunde A/S",
            "subject": "Driftsaftale",
            "total": "1.250",
            "startdate": "2026-01-01",
            "subscriptionstatus": "active",
        },
    )
    assert item["entity_type"] == "subscription"
    assert item["source_record_id"] == "72x123"
    assert item["source_customer_id"] == "3x44"
    assert item["product_name"] == "Driftsaftale"
    assert item["period_from"].isoformat() == "2026-01-01"
    assert len(item["source_hash"]) == 64


def test_source_frequency_is_normalized_to_hub_interval():
    assert router._hub_interval("monthly_3_last_day") == "monthly"
    assert router._hub_interval("Quarterly") == "quarterly"
    assert router._hub_interval("Annual") == "yearly"


def test_economic_repository_is_read_only_query(monkeypatch):
    captured = {}

    def fake_query(sql, params=None):
        captured["sql"] = sql
        captured["params"] = params
        return []

    monkeypatch.setattr(
        "app.modules.migration_center.backend.service.execute_query",
        fake_query,
    )
    assert EconomicSnapshotRepository.lines(42) == []
    normalized = " ".join(captured["sql"].split()).upper()
    assert normalized.startswith("SELECT")
    assert all(keyword not in normalized for keyword in ("INSERT INTO", "UPDATE ", "DELETE FROM"))
    assert captured["params"] == (42,)
    assert "DATE_TRUNC('MONTH', CURRENT_DATE) - INTERVAL '12 MONTHS'" in normalized


def test_preflight_token_is_bound_to_item_hash_and_user(monkeypatch):
    monkeypatch.setattr(
        "app.modules.migration_center.backend.service.settings.JWT_SECRET_KEY",
        "test-migration-secret",
    )
    item = {"id": 9, "source_hash": "a" * 64}
    user = {"id": 7}
    token = preflight_token(item, user)
    verify_preflight(token, item, user)

    with pytest.raises(HTTPException) as error:
        verify_preflight(token, {**item, "source_hash": "b" * 64}, user)
    assert error.value.status_code == 409


def test_economic_line_text_is_bounded_for_staging_columns():
    item = __import__(
        "app.modules.migration_center.backend.service",
        fromlist=["_normalize_economic_line"],
    )._normalize_economic_line(
        {
            "source_type": "booked", "source_invoice_number": "1", "invoice_id": 1,
            "line_number": 1, "invoice_line_id": 2, "customer_number": 3,
            "customer_name": "K" * 300, "description": "P" * 800,
            "product_number": "X" * 150, "line_net_amount": 100, "quantity": 1,
        }
    )
    assert len(item["customer_name"]) == 255
    assert len(item["product_name"]) == 500
    assert len(item["product_code"]) == 100


def test_customer_lock_blocker_states_are_explicit():
    assert {"unlocked", "lock_failed"} == router.MUTABLE_LOCK_STATES


def test_simply_customer_resolution_reuses_existing_staging_mapping(monkeypatch):
    monkeypatch.setattr(
        "app.modules.migration_center.backend.service.execute_query_single",
        lambda sql, params: {
            "customer_name": "Korrekt Firma A/S",
            "cvr": "12345678",
            "hub_customer_id": 44,
        },
    )
    result = _resolve_source_customer(
        {"source_system": "simply", "source_customer_id": "11x123"}
    )
    assert result["hub_customer_id"] == 44
    assert result["customer_name"] == "Korrekt Firma A/S"
    assert result["rule"] == "Eksisterende Simply-kundemapping"