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 / "apac_lambda_tasks" / "APAC-Tenants-Monitor-Fanout-Worker-Task" ) VENDORED_PYMYSQL = ( ROOT / "aws-lambda-apac-copy" / "emea-source" / "unpacked" / "EMEA-Tenants-Scoring-Task" ) sys.path.insert(0, str(VENDORED_PYMYSQL)) 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) 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 def ping(self, reconnect=True): pass class CapturingCursor: def __init__(self): self.query = "" def __enter__(self): return self def __exit__(self, exc_type, exc, traceback): return False def execute(self, query, values): self.query = query 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 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": "", } self.active_route = { "tenant_bucket": "active", "hubspot_match": None, "priority_override": None, "priority_reason": None, "snapshot_status": "fresh", } def test_active_success_commits_then_enqueues_scoring(self): connection = FakeConnection() with ( mock.patch.object( worker_function, "process_tenant", return_value={"status": "processed", "route": self.active_route}, ), 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["status"]) self.assertEqual(1, connection.commits) self.assertEqual(0, connection.rollbacks) enqueue.assert_called_once_with( "42", "active_candidate", "active_processed", tenant_bucket="active", priority_override=None, priority_reason=None, hubspot_match=None, run_id=None, 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, "determine_tenant_route", return_value=self.active_route ), mock.patch.object(worker_function, "persist_routed_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["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", tenant_bucket="active", priority_override=None, priority_reason=None, hubspot_match=None, run_id=None, 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={"status": "processed", "route": self.active_route}, ), mock.patch.object(worker_function, "persist_routed_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",' '"effective_owner_sync_id":"sync-1",' '"tenant":{"id":"42"}}' } with ( mock.patch.object( worker_function, "load_effective_sales_owner", return_value={"email": "owner@pingcap.com", "source": "account"}, ), mock.patch.object(worker_function, "token_config") as token_config, mock.patch.object( worker_function, "process_new_signup_record", return_value={"status": "processed", "route": self.active_route}, ) 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( "owner@pingcap.com", process_new_signup.call_args.args[1]["effective_sales_owner_email"], ) self.assertEqual( "account", process_new_signup.call_args.args[1]["effective_sales_owner_source"], ) 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_forced_sales_owned_tenant_skips_all_enrichment_and_scoring(self): connection = FakeConnection() tenant_id = "1372813089209300444" record = { "body": ( '{"source_type":"active_candidate",' f'"tenant_id":"{tenant_id}",' '"effective_owner_sync_id":"sync-1",' f'"tenant":{{"id":"{tenant_id}"}}}}' ) } with ( mock.patch.object( worker_function, "load_effective_sales_owner", return_value={"email": "owner@pingcap.com", "source": "account"}, ), mock.patch.object( worker_function, "upsert_forced_sales_owned_tenant" ) as upsert, mock.patch.object(worker_function, "token_config") as tokens, mock.patch.object(worker_function, "process_active_record") as active, mock.patch.object(worker_function, "enqueue_scoring_request") as scoring, ): result = worker_function.process_record(connection, record, self.now) self.assertEqual("forced_sales_owned", result["status"]) self.assertEqual("sales_owned", result["tenant_bucket"]) self.assertTrue(result["sales_owned"]) self.assertEqual(1, connection.commits) self.assertEqual(0, connection.rollbacks) upsert.assert_called_once() self.assertEqual( "owner@pingcap.com", upsert.call_args.args[1]["effective_sales_owner_email"], ) self.assertEqual( "account", upsert.call_args.args[1]["effective_sales_owner_source"], ) tokens.assert_not_called() active.assert_not_called() scoring.assert_not_called() def test_forced_sales_owned_upsert_preserves_existing_data_and_sets_priority(self): connection = FakeConnection() with ( mock.patch.object( tenant_processor, "active_failure_enrichment", return_value={"cluster_count": 999}, ), mock.patch.object(tenant_processor, "persist_routed_tenant") as upsert, ): tenant_processor.upsert_forced_sales_owned_tenant( connection, self.tenant, "active_candidate" ) enrichment = upsert.call_args.args[2] self.assertEqual(999, enrichment["cluster_count"]) self.assertEqual("P0", enrichment["priority"]) self.assertEqual("active_candidate", enrichment["source_type"]) self.assertEqual("sales_owned", upsert.call_args.args[3]["tenant_bucket"]) 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", "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("OpenAI 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, "determine_tenant_route", return_value=self.active_route ), mock.patch.object(tenant_processor, "persist_routed_tenant") as upsert, ): status = tenant_processor.process_tenant( connection, self.tenant, self.tokens, self.now ) self.assertEqual("processed_with_errors", status["status"]) billing.assert_called_once() enrichment = upsert.call_args.args[2] self.assertEqual(3, enrichment["cluster_count"]) self.assertTrue(enrichment["has_credit_card"]) error_message = upsert.call_args.kwargs["error_message"] self.assertIn("domain intelligence failed", error_message) self.assertIn("cluster stats failed", error_message) 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) self.assertIn("effective_sales_owner_email", query) self.assertIn("effective_sales_owner_source", query) def test_effective_owner_is_loaded_from_exact_sync_batch(self): cursor = mock.MagicMock() cursor.__enter__.return_value = cursor cursor.fetchone.return_value = { "effective_owner_email": " OWNER@PINGCAP.COM ", "effective_owner_source": " ACCOUNT ", } connection = mock.MagicMock() connection.cursor.return_value = cursor owner = tenant_processor.load_effective_sales_owner( connection, "sync-1", "42" ) self.assertEqual( {"email": "owner@pingcap.com", "source": "account"}, owner ) self.assertEqual(("sync-1", "42"), cursor.execute.call_args.args[1]) def test_missing_effective_owner_snapshot_row_fails(self): cursor = mock.MagicMock() cursor.__enter__.return_value = cursor cursor.fetchone.return_value = None connection = mock.MagicMock() connection.cursor.return_value = cursor with self.assertRaisesRegex(RuntimeError, "snapshot row is missing"): tenant_processor.load_effective_sales_owner( connection, "sync-1", "42" ) def test_hubspot_match_wins_over_apac_sales_owner(self): hubspot_company = { "hubspot_company_id": "company-1", "synced_at": self.now, } snapshot = { "sync_id": "sync-1", "synced_at": self.now, "last_attempt_status": "failed", "portal_map": {}, "account_map": {}, "domain_map": {"example.com": [hubspot_company]}, } tenant = { **self.tenant, "effective_sales_owner_email": "sales@pingcap.com", } with ( mock.patch.object( tenant_processor, "load_hubspot_snapshot", return_value=snapshot ), mock.patch.dict( tenant_processor.os.environ, {"APAC_SALES_EMAILS": "sales@pingcap.com"}, ), ): route = tenant_processor.determine_tenant_route( FakeConnection(), tenant, {"email_domain": "example.com"} ) self.assertEqual("active", route["tenant_bucket"]) self.assertEqual("P0", route["priority_override"]) self.assertEqual("email_domain", route["hubspot_match"]["match_method"]) self.assertEqual("stale", route["snapshot_status"]) def test_apac_sales_owner_routes_to_sales_owned_when_hubspot_does_not_match(self): snapshot = { "sync_id": "sync-1", "synced_at": self.now, "last_attempt_status": "success", "portal_map": {}, "account_map": {}, "domain_map": {}, } tenant = { **self.tenant, "effective_sales_owner_email": "sales@pingcap.com", } with ( mock.patch.object( tenant_processor, "load_hubspot_snapshot", return_value=snapshot ), mock.patch.dict( tenant_processor.os.environ, {"APAC_SALES_EMAILS": "sales@pingcap.com"}, ), ): route = tenant_processor.determine_tenant_route( FakeConnection(), tenant, {"email_domain": "other.example"} ) self.assertEqual("sales_owned", route["tenant_bucket"]) self.assertIsNone(route["priority_override"]) def test_legacy_sales_owner_does_not_drive_sales_owned_route(self): snapshot = { "sync_id": "sync-1", "synced_at": self.now, "last_attempt_status": "success", "portal_map": {}, "account_map": {}, "domain_map": {}, } tenant = {**self.tenant, "sales_owner_email": "sales@pingcap.com"} with ( mock.patch.object( tenant_processor, "load_hubspot_snapshot", return_value=snapshot ), mock.patch.dict( tenant_processor.os.environ, {"APAC_SALES_EMAILS": "sales@pingcap.com"}, ), ): route = tenant_processor.determine_tenant_route( FakeConnection(), tenant, {"email_domain": "other.example"} ) self.assertEqual("active", route["tenant_bucket"]) def test_bucket_move_preserves_score_and_manual_tracking_state(self): route = {"tenant_bucket": "sales_owned", "hubspot_match": None} existing_state = { "total_score": 55, "priority": "P1", "track_step": 2, "track_update_time": self.now, "startup_program": "yes", } with ( mock.patch.object( tenant_processor, "get_existing_tenant_routing_state", return_value=existing_state, ), mock.patch.object(tenant_processor, "upsert_active_tenant") as upsert, mock.patch.object(tenant_processor, "delete_tenant_from_other_bucket") as delete, ): tenant_processor.persist_routed_tenant( FakeConnection(), self.tenant, {"email_domain": "example.com"}, route, source_type="active_candidate", ) routed_enrichment = upsert.call_args.args[2] self.assertEqual(55, routed_enrichment["total_score"]) self.assertEqual(2, routed_enrichment["track_step"]) self.assertEqual("yes", routed_enrichment["startup_program"]) self.assertEqual("sales_owned", upsert.call_args.kwargs["tenant_bucket"]) delete.assert_called_once_with(mock.ANY, "42", "sales_owned") if __name__ == "__main__": unittest.main()