blocks-transformer/test_block.py

import json
import unittest
from pathlib import Path
from block import __main__


BASE_DIR = Path(__file__).resolve().parent

data = [{
            "application_key": "A3CDD39F-10F8-40B0-A4C9-0E1558B75131",
            "hd_score_m1": 1101.0,
            "hd_score_iso_m2": 1113,
            "cluster_size": 10,
            "cluster_size_users_v2": 3,
            "target_connected_30_sum": 0.0,
            "email_cnt": 3,
            "rejected_app_count": 6.0,
            "app_dt_day_cnt": 7,
            "hd_score_m2": 1188,
            "hd_score_m2_connected_max": 1197.0,
            "hd_score_m2_connected_avg": 1184.888889,
            "applicant_age_connected_max": 60.0,
            "applicant_age_connected_avg": 52.44444444,
            "account_tel_first_seen_min_conn": 879.0,
            "account_tel_first_seen_max_conn": 989.0,
            "account_tel_first_seen_avg_conn": 949.6666667,
            "ssn_hash_first_seen_min_conn": 5.0,
            "ssn_hash_first_seen_avg_conn": 58.0,
            "account_login_first_seen_min_conn": 0.0,
            "digital_id_first_seen_max_conn": 2652.0,
            "true_ip_first_seen_min_conn": 1857.0,
            "true_ip_first_seen_max_conn": 1967.0,
            "dist_em_ip_ref_km_min_conn": 17.43689023,
            "pct_acc_email_attr_challenged_1_conn": 0.0,
            "account_login_first_seen_range_conn": 2313.0,
            "account_login_first_seen_stddev_conn": 1042.4994,
            "cpu_clock_range_conn": 9054.0,
            "summary_risk_score_max_conn": 14.0
        }]


class TestBlock(unittest.TestCase):
    def test_main_returns_scores(self):
        block_result = __main__(data)
        self.assertIsInstance(block_result, dict)
        self.assertIn("hd_score_g1", block_result)
        self.assertIn("hd_score_g2", block_result)

    def test_main_passes_velocity_rule_fields_through(self):
        row = dict(data[0])
        row.update({
            "cluster_size": 1,
            "application_customer_type": "Direct New",
            "input_ip_address": "203.0.113.10",
            "input_ip_connection_type": "broadband",
            "input_ip_isp": "local isp",
            "input_ip_distinct_ssn_24h": 3,
            "input_ip_distinct_zip_24h": 2,
        })

        block_result = __main__([row])

        self.assertEqual(block_result["application_customer_type"], "Direct New")
        self.assertEqual(block_result["input_ip_address"], "203.0.113.10")
        self.assertEqual(block_result["input_ip_connection_type"], "broadband")
        self.assertEqual(block_result["input_ip_isp"], "local isp")
        self.assertEqual(block_result["input_ip_distinct_ssn_24h"], 3)
        self.assertEqual(block_result["input_ip_distinct_zip_24h"], 2)

    def test_response_schema_includes_velocity_rule_fields(self):
        schema = json.loads((BASE_DIR / "response_schema.json").read_text())

        expected_types = {
            "application_customer_type": ["string", "null"],
            "input_ip_address": ["string", "null"],
            "input_ip_connection_type": ["string", "null"],
            "input_ip_isp": ["string", "null"],
            "input_ip_distinct_ssn_24h": ["number", "null"],
            "input_ip_distinct_zip_24h": ["number", "null"],
        }
        for field, expected_type in expected_types.items():
            self.assertEqual(schema["properties"][field]["type"], expected_type)


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