import json import os import sys import types import unittest from datetime import datetime, timezone from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[2] WORKER_DIR = ( ROOT / "latam_lambda_tasks" / "LATAM-Tenants-Monitor-Fanout-Worker-Task" ) sys.path.insert(0, str(WORKER_DIR)) fake_boto3 = types.ModuleType("boto3") fake_boto3.client = lambda service_name: None sys.modules.setdefault("boto3", fake_boto3) fake_pymysql = types.ModuleType("pymysql") fake_pymysql.cursors = types.SimpleNamespace(DictCursor=object) fake_pymysql.connect = lambda **kwargs: None sys.modules.setdefault("pymysql", fake_pymysql) import worker_function # noqa: E402 import cluster_metrics # noqa: E402 import tenant_processor # noqa: E402 class FakeConnection: def __init__(self): self.commits = 0 self.rollbacks = 0 def commit(self): self.commits += 1 def rollback(self): self.rollbacks += 1 def close(self): pass class CapturingCursor: def __init__(self): self.query = "" self.values = None def __enter__(self): return self def __exit__(self, exc_type, exc, traceback): return False def execute(self, query, values): self.query = query self.values = values def fetchone(self): return None class CapturingConnection(FakeConnection): def __init__(self): super().__init__() self.captured_cursor = CapturingCursor() def ping(self, reconnect=True): pass def cursor(self): return self.captured_cursor class FakeHTTPResponse: def __init__(self, payload): self.payload = payload def __enter__(self): return self def __exit__(self, exc_type, exc, traceback): return False def read(self): return json.dumps(self.payload).encode("utf-8") class WorkerFunctionTest(unittest.TestCase): def setUp(self): self.now = datetime(2026, 7, 13, tzinfo=timezone.utc) self.tenant = {"id": "42", "email": "tenant@example.com"} self.tokens = { "devops_token": "redacted", "clinic_token": "redacted", "operator_email": "", } def test_active_success_commits_then_enqueues_scoring(self): connection = FakeConnection() with ( mock.patch.object(worker_function, "process_tenant", return_value="processed"), mock.patch.object(worker_function, "enqueue_scoring_request") as enqueue, ): status = worker_function.process_active_record( connection, self.tenant, self.tokens, self.now ) self.assertEqual("processed", status) self.assertEqual(1, connection.commits) self.assertEqual(0, connection.rollbacks) enqueue.assert_called_once_with( "42", "active_candidate", "active_processed", now=self.now ) def test_active_processing_failure_is_recorded_then_scored(self): connection = FakeConnection() with ( mock.patch.object( worker_function, "process_tenant", side_effect=RuntimeError("failed") ), mock.patch.object( worker_function, "active_failure_enrichment", return_value={} ), mock.patch.object(worker_function, "upsert_active_tenant") as upsert, mock.patch.object(worker_function, "enqueue_scoring_request") as enqueue, ): status = worker_function.process_active_record( connection, self.tenant, self.tokens, self.now ) self.assertEqual("recorded_failure", status) self.assertEqual(1, connection.commits) self.assertEqual(1, connection.rollbacks) upsert.assert_called_once() enqueue.assert_called_once_with( "42", "active_candidate", "active_failed", now=self.now ) def test_scoring_failure_propagates_without_overwriting_successful_data(self): connection = FakeConnection() with ( mock.patch.object(worker_function, "process_tenant", return_value="processed"), mock.patch.object(worker_function, "upsert_active_tenant") as fallback_upsert, mock.patch.object( worker_function, "enqueue_scoring_request", side_effect=RuntimeError("SQS unavailable"), ), ): with self.assertRaisesRegex(RuntimeError, "SQS unavailable"): worker_function.process_active_record( connection, self.tenant, self.tokens, self.now ) self.assertEqual(1, connection.commits) self.assertEqual(0, connection.rollbacks) fallback_upsert.assert_not_called() def test_new_signup_does_not_load_active_api_tokens(self): record = { "body": '{"source_type":"new_signup","tenant_id":"42",' '"tenant":{"id":"42"}}' } with ( mock.patch.object(worker_function, "token_config") as token_config, mock.patch.object( worker_function, "process_new_signup_record", return_value="processed" ) as process_new_signup, ): result = worker_function.process_record(FakeConnection(), record, self.now) token_config.assert_not_called() process_new_signup.assert_called_once() self.assertEqual("new_signup", result["source_type"]) def test_handler_returns_failed_record_for_retry(self): connection = FakeConnection() event = {"Records": [{"messageId": "message-1", "body": "{}"}]} with ( mock.patch.object( worker_function, "target_db_connect", return_value=connection ), mock.patch.object( worker_function, "process_record", side_effect=RuntimeError("SQS unavailable"), ), ): result = worker_function.lambda_handler(event, None) self.assertEqual(0, result["processed"]) self.assertEqual(1, result["failed"]) self.assertEqual( [{"itemIdentifier": "message-1"}], result["batchItemFailures"] ) def test_ten_clusters_skip_all_clinic_qps_calls(self): clusters = [ {"cluster_id": str(index), "service_plan": "dedicated"} for index in range(10) ] with ( mock.patch.object( cluster_metrics, "fetch_devops_clusters", return_value=(10, clusters), ), mock.patch.object(cluster_metrics, "fetch_clinic_cluster_by_id") as lookup, mock.patch.object(cluster_metrics, "cluster_qps_by_timestamp") as qps, ): result = cluster_metrics.get_tenant_cluster_stats( "42", "devops-token", None ) self.assertEqual(10, result["cluster_count"]) self.assertEqual(10, result["dedicated_cluster_count"]) self.assertIsNone(result["tenant_avg_qps"]) lookup.assert_not_called() qps.assert_not_called() def test_fewer_than_ten_clusters_still_require_clinic_token(self): with mock.patch.object( cluster_metrics, "fetch_devops_clusters", return_value=(1, [{"cluster_id": "cluster-1"}]), ): with self.assertRaisesRegex(RuntimeError, "CLINIC_API_KEY"): cluster_metrics.get_tenant_cluster_stats( "42", "devops-token", None ) def test_partial_cluster_failure_does_not_turn_unknown_qps_into_zero(self): clusters = [{"cluster_id": "one"}, {"cluster_id": "two"}] with ( mock.patch.object( cluster_metrics, "fetch_devops_clusters", return_value=(2, clusters), ), mock.patch.object( cluster_metrics, "fetch_clinic_cluster_by_id", side_effect=lambda token, cluster_id: {"clusterID": cluster_id}, ), mock.patch.object( cluster_metrics, "cluster_qps_by_timestamp", side_effect=[{"1": 12.0}, RuntimeError("Clinic unavailable")], ), ): result = cluster_metrics.get_tenant_cluster_stats( "42", "devops-token", "clinic-token" ) self.assertIsNone(result["tenant_avg_qps"]) self.assertEqual(1, len(result["errors"])) def test_active_enrichments_fail_independently(self): connection = FakeConnection() existing = { "cluster_count": 3, "starter_cluster_count": 1, "tenant_avg_qps": 2.5, "entity_ai_category": "company_non_ai", "icp_categories": '["fintech"]', "business_description": "Existing description", "official_website": "https://example.com", } with ( mock.patch.object( tenant_processor, "get_existing_active_tenant_enrichment", return_value=existing, ), mock.patch.object( tenant_processor, "get_domain_intelligence", side_effect=RuntimeError("Qwen unavailable"), ), mock.patch.object( tenant_processor, "get_tenant_cluster_stats", side_effect=RuntimeError("DevOps unavailable"), ), mock.patch.object( tenant_processor, "find_valid_credit_card", return_value={"has_credit_card": True, "billing_config_id": 7}, ) as billing, mock.patch.object( tenant_processor, "fetch_credit_summary", return_value={ "active_total_credits": 10, "latest_credit_grant_days_ago": 1, }, ), mock.patch.object(tenant_processor, "upsert_active_tenant") as upsert, ): status = tenant_processor.process_tenant( connection, self.tenant, self.tokens, self.now ) self.assertEqual("processed_with_errors", status) billing.assert_called_once() enrichment = upsert.call_args.args[2] self.assertEqual(3, enrichment["cluster_count"]) self.assertTrue(enrichment["has_credit_card"]) self.assertEqual(["fintech"], enrichment["icp_categories"]) self.assertIn("domain intelligence failed", upsert.call_args.args[3]) self.assertIn("cluster stats failed", upsert.call_args.args[3]) def test_cached_billing_config_without_card_triggers_rediscovery(self): configs = [ {"status": "active", "billing_config_id": 7}, {"status": "active", "billing_config_id": 8}, ] with ( mock.patch.object(tenant_processor, "devops_get", return_value=configs), mock.patch.object( tenant_processor, "find_valid_credit_card_in_billing_config", side_effect=[ None, {"has_credit_card": True, "billing_config_id": 8}, ], ) as lookup, ): result = tenant_processor.find_valid_credit_card( "42", "devops-token", billing_config_id=7, now=self.now ) self.assertTrue(result["has_credit_card"]) self.assertEqual(8, result["billing_config_id"]) self.assertEqual(2, lookup.call_count) def test_new_signup_upsert_preserves_existing_score_and_priority(self): connection = CapturingConnection() tenant_processor.upsert_new_signup_into_active_tenants( connection, self.tenant, { "email_domain": "example.com", "entity_ai_category": "company_non_ai", "business_description": "Example", "official_website": "https://example.com", }, ) query = connection.captured_cursor.query self.assertIn( "total_score = COALESCE(VALUES(total_score), total_score)", query ) self.assertIn("priority = COALESCE(VALUES(priority), priority)", query) def test_icp_categories_are_canonical_multi_select_values(self): self.assertEqual( ["fintech", "e_commerce", "ai", "igaming"], tenant_processor.normalize_icp_categories( ["AI", "E-Commerce", "fintech", "i-gaming", "unsupported"] ), ) def test_individual_domain_cannot_retain_icp_categories(self): self.assertEqual( [], tenant_processor.eligible_icp_categories( "individual", ["fintech", "ai"] ), ) def test_domain_intelligence_persists_icp_categories_as_json(self): connection = CapturingConnection() tenant_processor.upsert_domain_intelligence( connection, "example.com", { "entity_ai_category": "company_non_ai", "icp_categories": ["retail", "ecommerce"], "business_description": "Example retailer", "official_website": "https://example.com", }, "test", ) cursor = connection.captured_cursor self.assertIn("icp_categories", cursor.query) self.assertEqual( ["retail", "e_commerce"], json.loads(cursor.values["icp_categories"]), ) def test_domain_cache_only_reuses_current_qwen_classifier_version(self): connection = CapturingConnection() result = tenant_processor.get_cached_domain_intelligence( connection, "example.com" ) self.assertIsNone(result) cursor = connection.captured_cursor self.assertIn("AND source = %s", cursor.query) self.assertEqual( ("example.com", "qwen_web_search_v1"), cursor.values, ) def test_native_dashscope_endpoint_builds_workspace_responses_url(self): self.assertEqual( "https://workspace-1.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1/responses", tenant_processor.qwen_responses_url( "https://dashscope-intl.aliyuncs.com/api/v1", "workspace-1", "ap-southeast-1", ), ) def test_qwen_endpoint_rejects_non_alibaba_cloud_host(self): with self.assertRaisesRegex(RuntimeError, "Alibaba Cloud host"): tenant_processor.qwen_responses_url( "https://example.com/compatible-mode/v1", "workspace-1", "ap-southeast-1", ) def test_qwen_domain_classification_uses_web_search_without_reasoning(self): model_output = { "entity_ai_category": "company_non_ai", "icp_categories": ["fintech"], "business_description": "Example provides payment infrastructure.", "official_website": "https://example.com", } response = { "status": "completed", "output": [ { "type": "web_search_call", "status": "completed", }, { "type": "message", "status": "completed", "content": [ { "type": "output_text", "text": f"```json\n{json.dumps(model_output)}\n```", } ], }, ], } environment = { "API_KEY": "test-key", "WORKSPACE_ID": "workspace-1", "DASHSCOPE_ENDPOINT": "https://dashscope-intl.aliyuncs.com/api/v1", "REGION": "ap-southeast-1", "QWEN_WEB_SEARCH_MODEL": "qwen3.8-max", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( tenant_processor.urllib.request, "urlopen", return_value=FakeHTTPResponse(response), ) as urlopen, ): result = tenant_processor.classify_domain_with_web_search("example.com") request = urlopen.call_args.args[0] request_body = json.loads(request.data.decode("utf-8")) self.assertEqual( "https://workspace-1.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1/responses", request.full_url, ) self.assertEqual("qwen3.8-max", request_body["model"]) self.assertEqual([{"type": "web_search"}], request_body["tools"]) self.assertEqual({"effort": "none"}, request_body["reasoning"]) self.assertIn("an .ai domain is not sufficient evidence", request_body["input"]) self.assertEqual("company_non_ai", result["entity_ai_category"]) self.assertEqual(["fintech"], result["icp_categories"]) def test_incomplete_qwen_response_is_returned_as_unknown(self): response = { "status": "incomplete", "output_text": "Still searching", } environment = { "API_KEY": "test-key", "WORKSPACE_ID": "workspace-1", "DASHSCOPE_ENDPOINT": "https://dashscope-intl.aliyuncs.com/api/v1", "REGION": "ap-southeast-1", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( tenant_processor.urllib.request, "urlopen", return_value=FakeHTTPResponse(response), ), ): result = tenant_processor.classify_domain_with_web_search("example.com") self.assertEqual("unknown", result["entity_ai_category"]) self.assertEqual([], result["icp_categories"]) self.assertIn("status was incomplete", result["business_description"]) if __name__ == "__main__": unittest.main()