import contextlib
import io
import importlib.util
import json
import os
import ssl
import subprocess
import sys
import tempfile
import types
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "apply_exp_bulk_baseline.py"
PACK_REFERENCE_DATA_SCRIPT = ROOT / "scripts" / "apply_pack_reference_data.py"
TIDB_EXEC_SCRIPT = ROOT / "scripts" / "tidb_exec_sql.py"
DEPLOY_BASELINE_SEED = (
    ROOT / "deployments" / "offline" / "runtime" / "seeds" / "deploy-baseline.seed.yaml"
)
REFERENCE_DATA_DIR = ROOT / "deployments" / "offline" / "runtime" / "reference-data"
REFERENCE_DATA_MANIFEST = REFERENCE_DATA_DIR / "manifest.json"
INDUSTRY_VERTICAL_REFERENCE_DATA = REFERENCE_DATA_DIR / "industry-vertical.json"
APPROVAL_EMAIL_TEMPLATES = REFERENCE_DATA_DIR / "approval-email-templates.json"
EMAIL_TEMPLATE_CATALOG = REFERENCE_DATA_DIR / "email-template-catalog.json"

spec = importlib.util.spec_from_file_location("apply_exp_bulk_baseline", SCRIPT)
bulk = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = bulk
spec.loader.exec_module(bulk)

if "pymysql" not in sys.modules:
    fake_pymysql = types.ModuleType("pymysql")
    fake_pymysql.connect = lambda **_: None
    sys.modules["pymysql"] = fake_pymysql
if "pymysql.constants" not in sys.modules:
    fake_constants = types.ModuleType("pymysql.constants")
    fake_constants.CLIENT = types.SimpleNamespace(MULTI_STATEMENTS=65536)
    sys.modules["pymysql.constants"] = fake_constants

tidb_spec = importlib.util.spec_from_file_location("tidb_exec_sql", TIDB_EXEC_SCRIPT)
tidb_exec = importlib.util.module_from_spec(tidb_spec)
sys.modules[tidb_spec.name] = tidb_exec
tidb_spec.loader.exec_module(tidb_exec)

pack_reference_data_spec = importlib.util.spec_from_file_location(
    "apply_pack_reference_data",
    PACK_REFERENCE_DATA_SCRIPT,
)
pack_reference_data = importlib.util.module_from_spec(pack_reference_data_spec)
sys.modules[pack_reference_data_spec.name] = pack_reference_data
pack_reference_data_spec.loader.exec_module(pack_reference_data)


def sample_seed():
    return {
        "inputs": {},
        "lookup_records": [
            {
                "key": "watchdog_admin_user",
                "object": "User",
                "match": {"Username": "admin@example.invalid"},
            }
        ],
        "baseline_records": [
            {
                "key": "account_1",
                "object": "Account",
                "match": {"Name": "EXP Bulk Account"},
                "fields": {"Name": "EXP Bulk Account"},
            },
            {
                "key": "campaign_1",
                "object": "Campaign",
                "match": {"Name": "EXP Bulk Campaign"},
                "fields": {"Name": "EXP Bulk Campaign", "IsActive": True},
            },
            {
                "key": "campaign_status_sent",
                "object": "CampaignMemberStatus",
                "match": {
                    "CampaignId": "{{ record.campaign_1.id }}",
                    "Label": "Sent",
                },
                "fields": {
                    "CampaignId": "{{ record.campaign_1.id }}",
                    "Label": "Sent",
                    "SortOrder": 1,
                    "IsDefault": True,
                },
            },
            {
                "key": "pricebook_1",
                "object": "Pricebook2",
                "match": {"Name": "Standard Price Book"},
                "fields": {"Name": "Standard Price Book", "IsActive": True},
            },
            {
                "key": "product_1",
                "object": "Product2",
                "match": {"ProductCode": "EXP-001"},
                "fields": {
                    "Name": "EXP Product",
                    "ProductCode": "EXP-001",
                    "IsActive": True,
                },
            },
            {
                "key": "entry_1",
                "object": "PricebookEntry",
                "match": {
                    "Pricebook2Id": "{{ record.pricebook_1.id }}",
                    "Product2Id": "{{ record.product_1.id }}",
                },
                "fields": {
                    "Pricebook2Id": "{{ record.pricebook_1.id }}",
                    "Product2Id": "{{ record.product_1.id }}",
                    "UnitPrice": 10,
                    "IsActive": True,
                },
            },
            {
                "key": "entry_2",
                "object": "PricebookEntry",
                "match": {
                    "Pricebook2Id": "{{ record.pricebook_1.id }}",
                    "Product2Id": "{{ record.product_1.id }}",
                },
                "fields": {
                    "Pricebook2Id": "{{ record.pricebook_1.id }}",
                    "Product2Id": "{{ record.product_1.id }}",
                    "UnitPrice": 20,
                    "IsActive": True,
                },
            },
            {
                "key": "country_us",
                "object": "Country_Dictionary__c",
                "match": {"Name": "US"},
                "fields": {"Name": "US", "Full_Name__c": "United States"},
            },
            {
                "key": "country_region_us",
                "object": "Countries_Region__c",
                "match": {"Name": "US"},
                "fields": {"Name": "US", "Region_Name__c": "NA"},
            },
            {
                "key": "generic_unresolved_ref",
                "object": "Unresolved_Reference_Test__c",
                "match": {"Name": "Needs missing ref"},
                "fields": {
                    "Name": "Needs missing ref",
                    "Lookup__c": "{{ record.missing_seed_fixture.id }}",
                },
            },
        ],
    }


class ExpBulkBaselineTests(unittest.TestCase):
    def test_dry_run_plan_layers_collapse_duplicates_and_defers_unresolved_refs(self):
        selected = set(bulk.DEFAULT_BULK_OBJECTS) | {"Unresolved_Reference_Test__c"}

        plan = bulk.plan_bulk_seed(sample_seed(), selected, dry_run=True)

        self.assertIn("Product2", plan["layers"][0])
        self.assertIn("Pricebook2", plan["layers"][0])
        self.assertIn("Campaign", plan["layers"][0])
        self.assertIn("Country_Dictionary__c", plan["layers"][0])
        self.assertIn("Countries_Region__c", plan["layers"][0])
        self.assertIn("PricebookEntry", plan["layers"][1])
        self.assertIn("CampaignMemberStatus", plan["layers"][1])

        unsupported = {item["object"]: item["reason"] for item in plan["unsupported"]}
        self.assertIn("Unresolved_Reference_Test__c", unsupported)
        self.assertIn("missing_seed_fixture", unsupported["Unresolved_Reference_Test__c"])

        pricebook_entries = plan["planned_by_object"]["PricebookEntry"]
        self.assertEqual(len(pricebook_entries), 1)
        self.assertEqual(pricebook_entries[0]["alias_keys"], ["entry_1", "entry_2"])
        self.assertEqual(pricebook_entries[0]["fields"]["UnitPrice"], 20)
        self.assertEqual(plan["collapsed_duplicate_records"], 1)
        self.assertEqual(plan["duplicate_match_group_count"], 1)
        self.assertEqual(plan["conflicting_duplicate_match_group_count"], 1)

        state = plan["records"]
        self.assertEqual(state["entry_1"]["id"], state["entry_2"]["id"])
        self.assertEqual(state["entry_1"]["UnitPrice"], 20)
        self.assertEqual(state["country_us"]["Full_Name__c"], "United States")

    def test_match_key_normalizes_seed_booleans_to_db_integer_flags(self):
        fields = ["Email", "IsConverted"]

        self.assertEqual(
            bulk.match_key(
                {"Email": "staging.snapshot.lead.01@example.invalid", "IsConverted": False},
                fields,
            ),
            bulk.match_key(
                {"Email": "staging.snapshot.lead.01@example.invalid", "IsConverted": 0},
                fields,
            ),
        )

    def test_plan_reports_unselected_records_as_fallback_objects(self):
        plan = bulk.plan_bulk_seed(sample_seed(), {"Account"}, dry_run=True)

        fallback = {item["object"]: item["records"] for item in plan["fallback_objects"]}
        self.assertEqual(fallback["Campaign"], 1)
        self.assertEqual(fallback["PricebookEntry"], 2)
        self.assertNotIn("Account", fallback)

    def test_selected_only_seed_payload_drops_unselected_baseline_records(self):
        seed = sample_seed()

        filtered = bulk.seed_payload_for_selected_objects(seed, {"Account"})
        plan = bulk.plan_bulk_seed(filtered, {"Account"}, dry_run=True)

        self.assertEqual([record["object"] for record in filtered["baseline_records"]], ["Account"])
        self.assertEqual(plan["unsupported"], [])
        self.assertEqual(plan["fallback_objects"], [])
        self.assertEqual(plan["layers"], [["Account"]])

    def test_record_key_selection_keeps_only_reviewed_baseline_records(self):
        seed = sample_seed()

        filtered = bulk.seed_payload_for_selected_record_keys(
            seed,
            {"account_1", "campaign_1"},
        )

        self.assertEqual(
            [record["key"] for record in filtered["baseline_records"]],
            ["account_1", "campaign_1"],
        )

    def test_record_key_selection_rejects_unknown_keys(self):
        with self.assertRaisesRegex(
            bulk.BulkSeedError, "unknown --record-keys: missing_record"
        ):
            bulk.seed_payload_for_selected_record_keys(
                sample_seed(),
                {"account_1", "missing_record"},
            )

    def test_shared_deploy_baseline_is_fully_covered_by_bulk_seed(self):
        import yaml

        seed = yaml.safe_load(DEPLOY_BASELINE_SEED.read_text(encoding="utf-8"))

        plan = bulk.plan_bulk_seed(seed, set(bulk.DEFAULT_BULK_OBJECTS), dry_run=True)

        self.assertEqual(plan["unsupported"], [])
        self.assertEqual(plan["fallback_objects"], [])
        # Governed reference data is not owned by the residual shared baseline.
        self.assertNotIn("ReportFolder", plan["planned_by_object"])
        self.assertNotIn("_System_EmailTemplate", plan["planned_by_object"])
        self.assertNotIn("Industry_Vertical__c", plan["planned_by_object"])
        self.assertEqual(sum(len(rows) for rows in plan["planned_by_object"].values()), 24)


    def test_user_without_password_uses_admin_hash_and_state_redacts_password(self):
        columns = {
            "id": "Id",
            "username": "Username",
            "email": "Email",
            "password": "Password",
            "lastname": "LastName",
            "isdeleted": "IsDeleted",
        }
        row = {
            "id": "user-id",
            "object": "User",
            "match": {"Username": "seeded@example.invalid"},
            "fields": {
                "Email": "seeded@example.invalid",
                "LastName": "Seeded",
            },
            "alias_keys": ["seeded_user"],
        }

        normalized = bulk.normalize_row_for_table("User", row, columns, None, "admin-hash")
        state = bulk.state_for_row(
            {
                **row,
                "fields": {
                    **row["fields"],
                    "Password": "plain-secret",
                },
            }
        )

        self.assertEqual(normalized["Password"], "admin-hash")
        self.assertNotIn("Password", state)

    def test_sparse_upsert_batches_do_not_null_unspecified_columns(self):
        table_columns = {
            "id": "Id",
            "name": "Name",
            "foo__c": "Foo__c",
            "isdeleted": "IsDeleted",
            "createddate": "CreatedDate",
            "lastmodifieddate": "LastModifiedDate",
        }
        normalized_rows = [
            {
                "Id": "row-1",
                "Name": "Seed Row 1",
                "Foo__c": None,
                "IsDeleted": False,
                "CreatedDate": "2026-06-14 00:00:00",
                "LastModifiedDate": "2026-06-14 00:00:00",
            },
            {
                "Id": "row-2",
                "Name": "Seed Row 2",
                "IsDeleted": False,
                "CreatedDate": "2026-06-14 00:00:00",
                "LastModifiedDate": "2026-06-14 00:00:00",
            },
        ]
        batches = bulk.build_sparse_upsert_batches(
            normalized_rows,
            [
                {"Name", "Foo__c"},
                {"Name"},
            ],
            table_columns,
        )

        shapes = {
            batch["insert_columns"]: batch["update_columns"]
            for batch in batches
        }
        self.assertEqual(len(batches), 2)
        self.assertEqual(
            shapes[("Id", "Name", "Foo__c", "IsDeleted", "CreatedDate", "LastModifiedDate")],
            ("Name", "Foo__c"),
        )
        self.assertEqual(
            shapes[("Id", "Name", "IsDeleted", "CreatedDate", "LastModifiedDate")],
            ("Name",),
        )

    def test_loader_managed_defaults_are_insert_only_unless_seed_specifies_them(self):
        table_columns = {
            "id": "Id",
            "name": "Name",
            "ownerid": "OwnerId",
            "isdeleted": "IsDeleted",
        }
        default_owner_only = {
            "Id": "row-1",
            "Name": "Seed Row 1",
            "OwnerId": "admin-user",
            "IsDeleted": False,
        }
        explicit_owner = {
            "Id": "row-2",
            "Name": "Seed Row 2",
            "OwnerId": "explicit-owner",
            "IsDeleted": False,
        }
        batches = bulk.build_sparse_upsert_batches(
            [default_owner_only, explicit_owner],
            [
                {"Name"},
                {"Name", "OwnerId"},
            ],
            table_columns,
        )

        self.assertEqual(len(batches), 2)
        self.assertEqual(batches[0]["insert_columns"], ("Id", "Name", "OwnerId", "IsDeleted"))
        self.assertEqual(batches[0]["update_columns"], ("Name",))
        self.assertEqual(batches[1]["insert_columns"], ("Id", "Name", "OwnerId", "IsDeleted"))
        self.assertEqual(batches[1]["update_columns"], ("Name", "OwnerId"))

    def test_dry_run_main_defaults_seed_apply_password_for_planning(self):
        seed = {
            "inputs": {},
            "lookup_records": [],
            "baseline_records": [
                {
                    "key": "seed_user",
                    "object": "User",
                    "match": {"Username": "seeded@example.invalid"},
                    "fields": {
                        "Username": "seeded@example.invalid",
                        "Email": "seeded@example.invalid",
                        "LastName": "Seeded",
                        "Password": "{{ env.EXP_SEED_APPLY_PASSWORD }}",
                    },
                }
            ],
        }
        old_argv = sys.argv[:]
        old_password = os.environ.pop("EXP_SEED_APPLY_PASSWORD", None)
        try:
            with tempfile.TemporaryDirectory() as temp_dir:
                temp_path = Path(temp_dir)
                seed_json = temp_path / "seed.json"
                state_file = temp_path / "state.json"
                summary_file = temp_path / "summary.json"
                seed_json.write_text(json.dumps(seed), encoding="utf-8")
                sys.argv = [
                    "apply_exp_bulk_baseline.py",
                    "--seed-json",
                    str(seed_json),
                    "--state-file",
                    str(state_file),
                    "--summary-file",
                    str(summary_file),
                    "--dry-run",
                ]

                with contextlib.redirect_stdout(io.StringIO()):
                    self.assertEqual(bulk.main(), 0)

                summary = json.loads(summary_file.read_text(encoding="utf-8"))
                state = json.loads(state_file.read_text(encoding="utf-8"))
                self.assertEqual(summary["unsupported"], [])
                self.assertTrue(summary["coverage_complete"])
                self.assertNotIn("Password", state["records"]["seed_user"])
        finally:
            sys.argv = old_argv
            if old_password is None:
                os.environ.pop("EXP_SEED_APPLY_PASSWORD", None)
            else:
                os.environ["EXP_SEED_APPLY_PASSWORD"] = old_password


    def test_shell_wrapper_dry_run_uses_deploy_owned_seed_renderer(self):
        seed = {
            "inputs": {},
            "lookup_records": [],
            "baseline_records": [
                {
                    "key": "account_1",
                    "object": "Account",
                    "match": {"Name": "EXP Shell Wrapper Account"},
                    "fields": {"Name": "EXP Shell Wrapper Account"},
                }
            ],
        }

        with tempfile.TemporaryDirectory() as temp_dir:
            temp_path = Path(temp_dir)
            seed_file = temp_path / "seed.yaml"
            output_dir = temp_path / "out"
            state_file = temp_path / "state.json"
            summary_file = temp_path / "summary.json"
            seed_file.write_text(json.dumps(seed), encoding="utf-8")

            result = subprocess.run(
                [
                    "bash",
                    str(ROOT / "scripts" / "apply_exp_bulk_baseline.sh"),
                    "--seed-file",
                    str(seed_file),
                    "--output-dir",
                    str(output_dir),
                    "--state-file",
                    str(state_file),
                    "--summary-file",
                    str(summary_file),
                    "--objects",
                    "Account",
                    "--dry-run",
                ],
                cwd=ROOT,
                text=True,
                capture_output=True,
            )

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertNotIn("watchdog/runner", result.stdout + result.stderr)
            summary = json.loads(summary_file.read_text(encoding="utf-8"))
            state = json.loads(state_file.read_text(encoding="utf-8"))
            self.assertTrue(summary["coverage_complete"])
            self.assertEqual(summary["bulk_record_count"], 1)
            self.assertEqual(
                state["records"]["account_1"]["Name"],
                "EXP Shell Wrapper Account",
            )

    def test_pack_reference_manifest_renders_all_governed_reference_data(self):
        manifest = pack_reference_data.load_json(
            REFERENCE_DATA_MANIFEST,
            label="reference-data manifest",
        )
        specs = pack_reference_data.manifest_object_specs(manifest)
        records = pack_reference_data.build_seed(
            REFERENCE_DATA_MANIFEST,
            [
                "Industry_Vertical__c",
                "Countries_Region__c",
                "Country_Dictionary__c",
                "ReportFolder",
                "_System_EmailTemplate",
                "Dated_Conversion_Rate__c",
            ],
            specs,
        )["baseline_records"]

        self.assertEqual(
            set(specs),
            {
                "Industry_Vertical__c",
                "Countries_Region__c",
                "Country_Dictionary__c",
                "ReportFolder",
                "_System_EmailTemplate",
                "Dated_Conversion_Rate__c",
            },
        )
        # industry/vertical + country/region + governed Account set + promoted catalog.
        self.assertEqual(len(records), 846)
        email_templates = [r for r in records if r["object"] == "_System_EmailTemplate"]
        report_folders = [r for r in records if r["object"] == "ReportFolder"]
        countries_region = [r for r in records if r["object"] == "Countries_Region__c"]
        country_dictionary = [r for r in records if r["object"] == "Country_Dictionary__c"]
        self.assertEqual(len(email_templates), 79)  # 5 governed Account (incl. Waiting) + 74 catalog
        self.assertEqual(len(report_folders), 11)  # Account (governed) + 10 catalog
        self.assertEqual(len(countries_region), 250)
        self.assertEqual(len(country_dictionary), 250)
        ai_records = [
            record
            for record in records
            if record["object"] == "Industry_Vertical__c"
            and record["fields"]["Industry"] == "Artificial Intelligence"
        ]
        self.assertEqual(len(ai_records), 4)
        self.assertEqual(
            [record["fields"]["Label"] for record in ai_records],
            [
                "AI \u2013 LLM & MaaS",
                "AI \u2013 Agents & Orchestration",
                "AI \u2013 Native Applications",
                "AI \u2013 Hardware & Robotics",
            ],
        )

    def test_pack_reference_manifest_requires_email_folder_with_templates(self):
        manifest = pack_reference_data.load_json(
            REFERENCE_DATA_MANIFEST,
            label="reference-data manifest",
        )
        specs = pack_reference_data.manifest_object_specs(manifest)

        with self.assertRaisesRegex(pack_reference_data.PackReferenceDataError, "requires ReportFolder"):
            pack_reference_data.validate_dependency_closure(["_System_EmailTemplate"], specs)


    def test_industry_vertical_reference_data_is_not_reintroduced_by_baseline_seed(self):
        import yaml

        seed = yaml.safe_load(DEPLOY_BASELINE_SEED.read_text(encoding="utf-8"))
        baseline_objects = {record.get("object") for record in seed["baseline_records"]}
        reference_payload = json.loads(INDUSTRY_VERTICAL_REFERENCE_DATA.read_text(encoding="utf-8"))

        self.assertEqual(reference_payload["version"], 1)
        self.assertEqual(len(reference_payload["records"]), 114)
        self.assertNotIn("Industry_Vertical__c", baseline_objects)

    def test_prod_account_approval_reference_data_is_not_reintroduced_by_baseline_seed(self):
        import yaml

        seed = yaml.safe_load(DEPLOY_BASELINE_SEED.read_text(encoding="utf-8"))
        template_keys = {
            "tmpl_account_submitted_for_approval_notice",
            "tmpl_account_approved_notice",
            "tmpl_account_rejected_notice",
            "tmpl_account_recalled_notice",
            "tmpl_account_approval_waiting",
        }
        baseline_keys = {record.get("key") for record in seed["baseline_records"]}
        baseline_objects = {record.get("object") for record in seed["baseline_records"]}
        reference_payload = json.loads(APPROVAL_EMAIL_TEMPLATES.read_text(encoding="utf-8"))
        reference_keys = {record["key"] for record in reference_payload["records"]}

        # The governed Account approval set is owned ONLY by the reference-data file...
        self.assertEqual(reference_payload["version"], 1)
        self.assertEqual(reference_keys, {"folder_account_email_templates", *template_keys})
        # ...and email templates/folders are no longer carried by the deploy baseline seed.
        self.assertTrue(template_keys.isdisjoint(baseline_keys))
        self.assertNotIn("folder_account_email_templates", baseline_keys)
        self.assertNotIn("_System_EmailTemplate", baseline_objects)
        self.assertNotIn("ReportFolder", baseline_objects)

    def test_manifest_object_specs_accepts_string_or_list_source(self):
        specs = pack_reference_data.manifest_object_specs({
            "version": 1,
            "objects": {
                "A": {"source": "a.json"},
                "B": {"source": ["b.json", "c.json"], "depends_on": ["A"]},
            },
        })
        self.assertEqual(specs["A"]["sources"], ["a.json"])
        self.assertEqual(specs["B"]["sources"], ["b.json", "c.json"])
        with self.assertRaisesRegex(pack_reference_data.PackReferenceDataError, "source file"):
            pack_reference_data.manifest_object_specs(
                {"version": 1, "objects": {"A": {"source": []}}}
            )

    def test_email_template_catalog_is_disjoint_from_governed_account_set(self):
        catalog = json.loads(EMAIL_TEMPLATE_CATALOG.read_text(encoding="utf-8"))
        approval = json.loads(APPROVAL_EMAIL_TEMPLATES.read_text(encoding="utf-8"))
        self.assertEqual(catalog["version"], 1)
        catalog_keys = {r["key"] for r in catalog["records"]}
        approval_keys = {r["key"] for r in approval["records"]}
        self.assertEqual(catalog_keys & approval_keys, set())
        self.assertNotIn("folder_account_email_templates", catalog_keys)
        templates = [r for r in catalog["records"] if r["object"] == "_System_EmailTemplate"]
        folders = [r for r in catalog["records"] if r["object"] == "ReportFolder"]
        self.assertEqual(len(templates), 74)
        self.assertEqual(len(folders), 10)

    def test_prod_reference_data_wrapper_dry_run_uses_dedicated_source(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            temp_path = Path(temp_dir)
            output_dir = temp_path / "out"
            state_file = temp_path / "state.json"
            summary_file = temp_path / "summary.json"

            result = subprocess.run(
                [
                    "bash",
                    str(ROOT / "scripts" / "apply_pack_reference_data.sh"),
                    "--manifest-file",
                    str(REFERENCE_DATA_MANIFEST),
                    "--output-dir",
                    str(output_dir),
                    "--state-file",
                    str(state_file),
                    "--summary-file",
                    str(summary_file),
                    "--objects",
                    "Industry_Vertical__c,ReportFolder,_System_EmailTemplate",
                    "--dry-run",
                ],
                cwd=ROOT,
                text=True,
                capture_output=True,
            )

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            summary = json.loads(summary_file.read_text(encoding="utf-8"))
            state = json.loads(state_file.read_text(encoding="utf-8"))
            self.assertEqual(summary["apply_engine"], "mysql-cli")
            self.assertEqual(
                summary["reference_data_objects"],
                ["Industry_Vertical__c", "ReportFolder", "_System_EmailTemplate"],
            )
            self.assertEqual(summary["reference_data_record_count"], 204)
            self.assertEqual(state["records"]["deploy_ind_vert_10_01"]["Label"], "AI \u2013 LLM & MaaS")
            self.assertEqual(
                state["records"]["tmpl_account_approved_notice"]["FolderId"],
                state["records"]["folder_account_email_templates"]["id"],
            )

    def test_bulk_wrapper_fails_before_install_when_python_has_no_pip_or_uv(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            temp_path = Path(temp_dir)
            fake_bin = temp_path / "bin"
            fake_bin.mkdir()
            invocation_log = temp_path / "python-invocations.txt"
            fake_python = fake_bin / "python3"
            fake_python.write_text(
                "#!/usr/bin/env bash\n"
                "printf '%s\\n' \"$*\" >> \"${FAKE_PYTHON_INVOCATION_LOG}\"\n"
                "exit 1\n",
                encoding="utf-8",
            )
            fake_python.chmod(0o755)
            output_dir = temp_path / "output"
            env = os.environ.copy()
            env.update(
                {
                    "FAKE_PYTHON_INVOCATION_LOG": str(invocation_log),
                    "PATH": f"{fake_bin}:/usr/bin:/bin",
                    "PYTHON_SEED_DEPS_DIR": str(temp_path / "seed-deps"),
                }
            )

            result = subprocess.run(
                [
                    "bash",
                    str(ROOT / "scripts" / "apply_exp_bulk_baseline.sh"),
                    "--seed-file",
                    str(DEPLOY_BASELINE_SEED),
                    "--output-dir",
                    str(output_dir),
                    "--dry-run",
                ],
                cwd=ROOT,
                env=env,
                text=True,
                capture_output=True,
            )

            self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertIn("missing Python dependency installer", result.stderr)
            invocations = invocation_log.read_text(encoding="utf-8").splitlines()
            self.assertIn("-m pip --version", invocations)
            self.assertFalse(any("pip install" in invocation for invocation in invocations))

    def test_host_and_staging_preflight_require_seed_dependency_installer(self):
        host_installer = (ROOT / "scripts" / "install_ec2_host_prereqs.sh").read_text(
            encoding="utf-8"
        )
        staging_workflow = (
            ROOT / ".github" / "workflows" / "deploy-staging-blue-green-environment.yml"
        ).read_text(encoding="utf-8")

        self.assertIn("python3-pip", host_installer)
        self.assertIn("python3 -m ensurepip --upgrade --user", host_installer)
        self.assertIn("python3 -m pip --version", host_installer)
        self.assertIn(
            "command -v uv >/dev/null 2>&1 || python3 -m pip --version",
            staging_workflow,
        )

    def test_python_tidb_tls_context_matches_required_without_verify_contract(self):
        for context in [bulk.tidb_tls_context(), tidb_exec.tidb_tls_context()]:
            self.assertFalse(context.check_hostname)
            self.assertEqual(context.verify_mode, ssl.CERT_NONE)


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