103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from block import __main__
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
|
|
data = {
|
|
"hd_score_m1": 1093.0,
|
|
"hd_score_g1": 0.0,
|
|
"cluster_size_users_v2": 1.0,
|
|
"target_connected_30_sum": 0.0,
|
|
"email_cnt": 1.0,
|
|
"rejected_app_count": 0.0,
|
|
"app_dt_day_cnt": 1.0,
|
|
"hd_score_iso_m2": 1001.0,
|
|
"hd_score_g2": 0.0
|
|
}
|
|
|
|
|
|
def s4_data(**overrides):
|
|
values = {
|
|
**data,
|
|
"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,
|
|
}
|
|
values.update(overrides)
|
|
return values
|
|
|
|
|
|
class TestBlock(unittest.TestCase):
|
|
def test_main_success(self):
|
|
blockResult = __main__(**data)
|
|
|
|
self.assertIsInstance(blockResult, dict, "Result should be a dictionary.")
|
|
self.assertIn("hd_score_s1", blockResult, "Result dictionary should contain 'hd_score_s1' if success.")
|
|
|
|
def test_s4_triggers_for_direct_new_valid_ip_velocity(self):
|
|
block_result = __main__(**s4_data())
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 1191)
|
|
|
|
def test_s4_score_caps_at_1200(self):
|
|
block_result = __main__(**s4_data(
|
|
input_ip_distinct_ssn_24h=30,
|
|
input_ip_distinct_zip_24h=30,
|
|
))
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 1200)
|
|
|
|
def test_s4_requires_direct_new(self):
|
|
block_result = __main__(**s4_data(application_customer_type="Returning"))
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 0)
|
|
|
|
def test_s4_rejects_invalid_ip(self):
|
|
block_result = __main__(**s4_data(input_ip_address="0.0.0.0"))
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 0)
|
|
|
|
def test_s4_rejects_blocked_connection_type(self):
|
|
block_result = __main__(**s4_data(input_ip_connection_type="Mobile Wireless"))
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 0)
|
|
|
|
def test_s4_rejects_blocked_isp(self):
|
|
block_result = __main__(**s4_data(input_ip_isp="Cloudflare Inc."))
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 0)
|
|
|
|
def test_s4_requires_distinct_ssn_and_zip_thresholds(self):
|
|
block_result = __main__(**s4_data(input_ip_distinct_zip_24h=1))
|
|
|
|
self.assertEqual(block_result["hd_score_s4"], 0)
|
|
|
|
def test_s4_schema_contract(self):
|
|
request_schema = json.loads((BASE_DIR / "request_schema.json").read_text())
|
|
response_schema = json.loads((BASE_DIR / "response_schema.json").read_text())
|
|
|
|
for field in (
|
|
"application_customer_type",
|
|
"input_ip_address",
|
|
"input_ip_connection_type",
|
|
"input_ip_isp",
|
|
):
|
|
self.assertEqual(request_schema["properties"][field]["type"], ["string", "null"])
|
|
|
|
for field in (
|
|
"input_ip_distinct_ssn_24h",
|
|
"input_ip_distinct_zip_24h",
|
|
):
|
|
self.assertEqual(request_schema["properties"][field]["type"], ["number", "null"])
|
|
|
|
self.assertEqual(response_schema["properties"]["hd_score_s4"]["type"], ["number", "null"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|