add configurable qualification scoring

2026-09-03 11:25:14 +02:00
parent 89eb7e07e6
commit 655780ff88
13 changed files with 321 additions and 4 deletions
@@ -0,0 +1,67 @@
+import json
+import os
+import sqlite3
+import threading
+import unittest
+from http.client import HTTPConnection
+from tempfile import TemporaryDirectory
+
+from app.main import create_server
+from app.scoring import DEFAULT_RULES, evaluate_score
+
+
+class ScoringEngineTests(unittest.TestCase):
+    def test_defaults_are_deterministic_and_emit_explanations_and_band(self):
+        signals = {"business": {"name": "Acme", "email": "a@acme.test", "website_domain": "acme.test"}, "website": {"classification": "healthy"}, "state": {"suppressed": False}}
+        first = evaluate_score(signals, DEFAULT_RULES)
+        self.assertEqual(first, evaluate_score(signals, DEFAULT_RULES))
+        self.assertEqual(0 <= first["score"] <= 100, True)
+        self.assertEqual(first["priority_band"], "medium")
+        self.assertTrue(all("code" in item and "reason" in item for item in first["explanations"]))
+
+    def test_disabled_and_versioned_rules_change_score_without_nondeterminism(self):
+        signals = {"business": {"name": "Acme"}, "state": {"suppressed": False}}
+        enabled = evaluate_score(signals, [{"code": "x", "name": "X", "condition_json": {"signal": "business.name", "operator": "present"}, "points": 30, "enabled": 1, "version": 1}])
+        disabled = evaluate_score(signals, [{"code": "x", "name": "X", "condition_json": {"signal": "business.name", "operator": "present"}, "points": 30, "enabled": 0, "version": 2}])
+        self.assertEqual(enabled["score"], 30)
+        self.assertEqual(disabled["score"], 0)
+
+    def test_suppression_is_ineligible_and_stale_uncertain_signals_do_not_penalize(self):
+        signals = {"business": {"name": "Acme"}, "website": {"classification": "unknown", "stale": True}, "domain": {"status": "error"}, "state": {"suppressed": True}}
+        result = evaluate_score(signals, DEFAULT_RULES)
+        self.assertFalse(result["eligible"])
+        self.assertEqual(result["priority_band"], "ineligible")
+        self.assertNotIn("negative", json.dumps(result["explanations"]).lower())
+
+
+class ScoringApiTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory(); self.db_path = self.tmp.name + "/db.sqlite"
+        os.environ["BOOTSTRAP_ADMIN_EMAIL"] = "score-owner@example.test"; os.environ["BOOTSTRAP_ADMIN_PASSWORD"] = "password"
+        self.server = create_server("127.0.0.1", 0, self.db_path); threading.Thread(target=self.server.serve_forever, daemon=True).start()
+        self.conn = HTTPConnection("127.0.0.1", self.server.server_port, timeout=4); self.cookie = None
+        self.request("POST", "/api/v1/auth/login", {"email": "score-owner@example.test", "password": "password"})
+    def tearDown(self):
+        self.server.shutdown(); self.server.server_close(); self.tmp.cleanup()
+    def request(self, method, path, payload=None):
+        body = json.dumps(payload).encode() if payload is not None else None; headers = {"Content-Type": "application/json"} if body else {}
+        if self.cookie: headers["Cookie"] = self.cookie
+        self.conn.request(method, path, body, headers); response = self.conn.getresponse(); cookie = response.getheader("Set-Cookie")
+        if cookie: self.cookie = cookie.split(";", 1)[0]
+        return response.status, json.loads(response.read() or b"{}")
+    def test_rule_crud_tenant_scope_and_recalculation_audit(self):
+        status, rules = self.request("GET", "/api/v1/score-rules"); self.assertEqual(status, 200); self.assertEqual(len(rules["items"]), len(DEFAULT_RULES))
+        status, rule = self.request("POST", "/api/v1/score-rules", {"code": "custom", "name": "Custom", "condition_json": {"signal": "business.name", "operator": "present"}, "points": 7}); self.assertEqual(status, 201)
+        self.assertEqual(self.request("PATCH", f"/api/v1/score-rules/{rule['id']}", {"enabled": False})[0], 200)
+        _, business = self.request("POST", "/api/v1/businesses", {"name": "Acme", "website": "https://acme.test", "email": "a@acme.test"})
+        status, result = self.request("POST", f"/api/v1/businesses/{business['id']}/score/recalculate", {"reason": "qa"}); self.assertEqual(status, 200)
+        self.assertIn("explanations", result); self.assertIn("eligible", result); self.assertEqual(result["business_id"], business["id"])
+        status, summary = self.request("GET", "/api/v1/scoring/summary"); self.assertEqual(status, 200); self.assertEqual(summary["businesses"], 1); self.assertGreaterEqual(summary["history_count"], 1)
+        db = sqlite3.connect(self.db_path); self.assertEqual(db.execute("SELECT COUNT(*) FROM score_history").fetchone()[0], 1); db.close()
+    def test_suppressed_business_recalculation_is_force_ineligible(self):
+        _, business = self.request("POST", "/api/v1/businesses", {"name": "Acme", "website": "https://acme.test"})
+        self.request("POST", "/api/v1/suppressions", {"kind": "domain", "value": "acme.test"})
+        status, result = self.request("POST", f"/api/v1/businesses/{business['id']}/score/recalculate", {}); self.assertEqual(status, 200); self.assertFalse(result["eligible"])
+
+
+if __name__ == "__main__": unittest.main()