"""Unit tests for runner._extract_insert_id_target. Guards the idempotent-seed behavior added to fix PK-collision flake on the shared ci-test TiDB (see PR `fix(e2e-api): idempotent setup via pre-DELETE by primary key`). If a prior run's teardown didn't execute, setup INSERTs against hardcoded PKs (e.g. 'uli002-quote-000001') would otherwise fail with 1062. Run with: pytest tests/e2e-api/test_runner_idempotency.py """ from __future__ import annotations import sys import types from pathlib import Path # Make the e2e-api helpers importable without pulling in the real # config/api/mysql-connector modules (they require env and the driver). _E2E_API_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(_E2E_API_DIR)) sys.modules.setdefault("config", types.ModuleType("config")) sys.modules.setdefault("api", types.ModuleType("api")) sys.modules.setdefault("mysql", types.ModuleType("mysql")) sys.modules.setdefault("mysql.connector", types.ModuleType("mysql.connector")) import pytest # noqa: E402 from runner import TestRunner # noqa: E402 @pytest.mark.parametrize( "sql, expected_table", [ # Happy path: common shapes used by ULI / ORD / OLI / CTR cases. ("INSERT INTO Quote (Id, Name, AccountId) VALUES (%s, %s, %s)", "Quote"), ("INSERT INTO quote_product__c (Id, Quote__c) VALUES (%s, %s)", "quote_product__c"), ("INSERT INTO `Order` (Id, AccountId, Status) VALUES (%s, %s, %s)", "`Order`"), ("insert into Account ( Id , Name ) VALUES (%s, %s)", "Account"), ], ) def test_extracts_table_for_insert_with_id_pk(sql: str, expected_table: str) -> None: assert TestRunner._extract_insert_id_target(sql) == expected_table @pytest.mark.parametrize( "sql", [ # Already idempotent — don't double-clean. "INSERT IGNORE INTO Quote (Id, Name) VALUES (%s, %s)", # Id isn't the first column; we can't safely derive it from params[0]. "INSERT INTO Quote (Name, Id) VALUES (%s, %s)", # Non-INSERT: nothing to pre-clean. "SELECT * FROM Quote", "DELETE FROM Quote WHERE Id = %s", "UPDATE Quote SET Name = %s WHERE Id = %s", # Empty / malformed input: must not blow up. "", ], ) def test_skips_non_target_sql(sql: str) -> None: assert TestRunner._extract_insert_id_target(sql) is None