import argparse import datetime import json import re import uuid from typing import Any, Dict, List from collections import OrderedDict import yaml from environs import Env from models.enhanced_operation import EnhancedOperation from models.openapi import OpenAPISpec, generate_coefficient_table from models.test_case import Case, CaseManager, convert_case_to_multistep_style from utils.get_github_info import find_swagger_by_partial_commit_id from utils.openapi_expander import unwrap_refs from utils.postprocessing_biz import postprocess_cases env = Env() env.read_env() GITHUB_TOKEN = env.str("GITHUB_TOKEN") def load_config(config_file: str) -> dict: with open(config_file, "r") as f: return yaml.safe_load(f) def get_db_path(db_url: str) -> str: """Extract file path from database URL.""" if db_url.startswith("sqlite:///"): return db_url.replace("sqlite:///", "") return db_url def get_db_type(db_url: str) -> str: """Get database type from URL.""" if db_url.startswith("sqlite:///"): return "sqlite" elif db_url.startswith("mysql://"): return "tidb" return "unknown" def generate_output_filename() -> str: now = datetime.datetime.now() return f"sqlite:///output/{now.strftime('%Y%m%d%H%M%S')}.db" def prepare_spec( spec_json: Dict[str, Any], coeff_csv: str, input_cases_json: str ) -> List[EnhancedOperation]: parsed_spec = unwrap_refs(spec_json) with open("output/unwrapped_spec.json", "w") as f: json.dump(parsed_spec, f, indent=2) spec = OpenAPISpec.from_dict(parsed_spec) coeff_table = generate_coefficient_table(coeff_csv) spec.add_coefficient_table(coeff_table) with open(input_cases_json, "r") as f: cases = json.load(f) spec.add_cases(cases) ops = [] for path, path_item in spec.spec.paths.items(): available_methods = [m for m, _ in path_item.iter_methods()] for method, operation in path_item.iter_methods(): operation_coeff = spec.coefficient_table.get(operation.operationId, {}) # type: ignore cases = spec.pre_defined_cases.get( # type: ignore operation.operationId, {"cases": [], "multi_step_cases": []} # type: ignore ) operation.security = ( spec.spec.security if operation.security is None else ( operation.security if not spec.spec.security else operation.security + spec.spec.security ) ) op = EnhancedOperation.from_openapi( path, method, operation, available_methods, operation_coeff, cases["cases"], ) ops.append(op) return ops def main(): parser = argparse.ArgumentParser( description="Generate test cases from OpenAPI spec" ) parser.add_argument( "-c", "--config", required=True, help="Path to configuration file" ) parser.add_argument("--local-spec", help="Path to OpenAPI spec JSON file") parser.add_argument("--swagger-url", help="URL to swagger JSON file") parser.add_argument("--coeff", help="Path to coefficient CSV file") parser.add_argument("--cases", help="Path to input cases JSON file") parser.add_argument("--output", help="Path to output database file") parser.add_argument("--dry-run", action="store_true", help="Perform a dry run") parser.add_argument( "--use-allowlist", action="store_true", help="Use allowlist when generating cases", ) parser.add_argument("--allowlist", help="Path to allowlist YAML file") args = parser.parse_args() config = load_config(args.config) # Override config with command line arguments if args.local_spec: config["local_spec"] = args.local_spec if args.swagger_url: config["swagger_url"] = args.swagger_url if args.coeff: config["coeff"] = args.coeff if args.cases: config["cases"] = args.cases if args.output: config["output"] = args.output if args.dry_run: config["dry_run"] = args.dry_run if args.use_allowlist: config["use_allowlist"] = args.use_allowlist if args.allowlist: config["allowlist"] = args.allowlist allowlist = None if config.get("use_allowlist") and config.get("allowlist"): allowlist = load_config(config["allowlist"]) metadata = { "trace_id": str(uuid.uuid4()), "service_name": "dedicated_openapi", "protocol": "OpenAPI", "spec_hash": "0" * 32, "spec_commit_id": "0" * 40, "spec_pr_url": "placeholder", "generator_commit_id": "0" * 40, "generator_pr_url": "placeholder", "trigger_type": "manual", "enhanced_operations": [], } if config.get("local_spec") and not config.get("swagger_url"): with open(config["local_spec"], "r") as f: swagger_json = json.load(f) else: pattern = r"https://github.com/(.*)/(.*)/blob/(?:(\w{40})|([^/]+))/(.*)" match = re.match(pattern, config["swagger_url"]) if match: owner, repo, commit, branch, file_path = match.groups() else: raise ValueError("Invalid swagger URL") swagger_info = find_swagger_by_partial_commit_id( owner, repo, file_path, GITHUB_TOKEN, commit ) swagger_json = json.loads(swagger_info["file_content"]) metadata["spec_commit_id"] = swagger_info["commit_id"] metadata["spec_pr_url"] = swagger_info["pr_url"] metadata["spec_hash"] = swagger_info["md5_hash"] metadata["trigger_type"] = "manual_test" config["output"] = f"sqlite:///output/{swagger_info['commit_id'][:7]}.db" ops = prepare_spec(swagger_json, config["coeff"], config["cases"]) if config["output"] == "none" or not config["output"]: config["output"] = generate_output_filename() # Use in-memory SQLite for processing memory_db_path = ":memory:" # Create case manager with in-memory database case_manager = CaseManager(memory_db_path) try: if config.get("dry_run", False): # Pass the case_manager instead of creating new connection for operation in ops: cases = operation.generate_cases( dry_run=True, use_allowlist=config.get("use_allowlist", False), allowlist=allowlist, ) for c in cases: case_manager.add_or_update_case(Case.from_dict(c)) else: for operation in ops: cases = operation.generate_cases( dry_run=False, use_allowlist=config.get("use_allowlist", False), allowlist=allowlist, ) for c in cases: case_manager.add_or_update_case(Case.from_dict(c)) # Process cases while keeping connection open postprocess_cases(case_manager) # Determine target database type and path db_type = get_db_type(config["output"]) target_path = get_db_path(config["output"]) case_manager.set_metadata(metadata) case_manager.save_to_sqlite(target_path) case_manager.save_to_yaml(target_path) case_manager.save_to_tidb() multistep_style_auditlog = temporary_auditlog_to_multistep_style(case_manager) class CustomDumper(yaml.SafeDumper): def ignore_aliases(self, data): return True def represent_scalar(self, tag, value, style=None): if isinstance(value, str) and "\n" in value: return yaml.ScalarNode(tag, value, style="|") return super().represent_scalar(tag, value, style) yaml.add_representer( OrderedDict, yaml.representer.SafeRepresenter.represent_dict, Dumper=CustomDumper, ) with open("output/auditlog_multistep_style.yaml", "w", encoding="utf-8") as f: yaml.dump(multistep_style_auditlog, f, sort_keys=False, default_flow_style=False, allow_unicode=True, Dumper=CustomDumper) finally: # Close the connection only after all operations are complete case_manager.close() def temporary_auditlog_to_multistep_style(case_manager: CaseManager) -> dict: # prepare steps create_cluster_step = { "case_id": "create_cluster", "operation_id": "ClusterService_CreateCluster", "description": "Create a new cluster", "case_type": "typical_positive", "request_type": "http", "request": { "auth": { "auth_type": "digest", "content": {"private_key": "placeholder", "public_key": "placeholder"} }, "method": "POST", "path": "https://dedicated.dev.tidbapi.com/v1beta1/clusters", "path_params": {}, "query_params": {}, "headers": {}, "body": { "displayName": "auditlog-test", "labels": { "tidb.cloud/project": "{projectId}" }, "port": 4000, "regionId": "{regionId}", "rootPassword": "31415926", "tidbNodeSetting": { "nodeSpecKey": "2C8G", "tidbNodeGroups": [ { "nodeCount": 1 } ] }, "tikvNodeSetting": { "nodeCount": 3, "nodeSpecKey": "2C8G", "storageSizeGi": 200, "storageType": "Basic" } }, "body_params": { "regionId": "region_id", "projectId": "project_id" } }, "expect": { "status_code": 200, "cel": "body.state == \"CREATING\"" }, "step_type": "defined", "max_retries": 0, "delay_after": 0, "name": "create_cluster", "continue_on_unexpected": False, "save": { "placeholder": [ {"key": "cluster_1", "eval": "body.clusterId"} ] }, "stage": "prepare" } polling_get_cluster_step = { "case_id": "wait_for_cluster_active", "operation_id": "ClusterService_GetCluster", "description": "Poll the cluster state until it becomes ACTIVE to proceed with the update operation.", "case_type": "supplimental_positive", "request_type": "http", "request": { "auth": { "auth_type": "digest", "content": {"private_key": "placeholder", "public_key": "placeholder"} }, "method": "GET", "path": "https://dedicated.dev.tidbapi.com/v1beta1/clusters/{clusterId}", "path_params": { "clusterId": "cluster_1" }, "query_params": {}, "headers": {}, "body": {} }, "expect": { "status_code": 200, "cel": "body.state == \"ACTIVE\"" }, "step_type": "defined", "max_retries": 60, "delay_after": 60, "name": "wait_for_cluster_active", "continue_on_unexpected": False, "save": None, "stage": "teardown" } delete_cluster_step = { "case_id": "delete_cluster", "operation_id": "ClusterService_DeleteCluster", "description": "Delete the cluster", "case_type": "supplimental_positive", "request_type": "http", "request": { "auth": { "auth_type": "digest", "content": {"private_key": "placeholder", "public_key": "placeholder"} }, "method": "DELETE", "path": "https://dedicated.dev.tidbapi.com/v1beta1/clusters/{clusterId}", "path_params": { "clusterId": "cluster_1" }, "query_params": {}, "headers": {}, "body": {} }, "expect": { "status_code": 200, "cel": "" }, "step_type": "defined", "max_retries": 0, "delay_after": 0, "name": "delete_cluster", "continue_on_unexpected": False, "save": None, "stage": "teardown" } # create_auditlog_config_step = { # "case_id": "create_auditlog_config", # "operation_id": "DatabaseAuditLogService_CreateAuditLogConfig", # "case_type": "typical_positive", # "request_type": "http", # "request": { # "auth": { # "auth_type": "digest", # "content": {"private_key": "placeholder", "public_key": "placeholder"} # }, # "method": "POST", # "path": "https://dedicated.dev.tidbapi.com/v1beta1/clusters/{auditLogConfig.clusterId}/auditLogConfig", # "path_params": { # "auditLogConfig.clusterId": "cluster_1" # }, # "query_params": {}, # "headers": {}, # "body": { # "enabled": True, # "bucketUri": "placeholder", # "bucketRegionId": "", # "awsRoleArn": "", # "azureSasToken": "placeholder", # "bucketWriteCheck": {}, # "bucketManager": "CUSTOMER" # } # }, # "expect": { # "status_code": 200, # "cel": "" # }, # "step_type": "defined", # "max_retries": 0, # "delay_after": 0, # "name": "create_auditlog_config", # "continue_on_unexpected": False, # "save": None # } create_auditlog_filter_rule_step = { "case_id": "create_auditlog_filter_rule", "operation_id": "DatabaseAuditLogService_CreateAuditLogFilterRule", "description": "Create an auditlog filter rule", "case_type": "typical_positive", "request_type": "http", "request": { "auth": { "auth_type": "digest", "content": {"private_key": "placeholder", "public_key": "placeholder"} }, "method": "POST", "path": "https://dedicated.dev.tidbapi.com/v1beta1/clusters/{auditLogFilterRule.clusterId}/auditLogFilterRules", "path_params": { "auditLogFilterRule.clusterId": "cluster_1" }, "query_params": {}, "headers": {}, "body": { "userExpr": "root", "dbExpr": "test", "tableExpr": "test", "accessTypeList": [ "Alter Table", "Drop Table" ] } }, "expect": { "status_code": 200, "cel": "" }, "step_type": "defined", "max_retries": 0, "delay_after": 0, "name": "create_auditlog_filter_rule", "continue_on_unexpected": False, "save": { "placeholder": [ { "key": "auditlogfilterrule_1", "eval": "body.auditLogFilterRuleId" } ] }, "stage": "prepare" } steps = [] steps.append(create_cluster_step) steps.append(polling_get_cluster_step) audlitlog_operations = [ "DatabaseAuditLogService_GetAuditLogConfig", "DatabaseAuditLogService_UpdateAuditLogConfig", "DatabaseAuditLogService_CreateAuditLogConfig", "DatabaseAuditLogService_ListAuditLogFilterRules", "DatabaseAuditLogService_GetAuditLogFilterRule", "DatabaseAuditLogService_DeleteAuditLogFilterRule", "DatabaseAuditLogService_CreateAuditLogFilterRule", # "DatabaseAuditLogService_ShowObjectStorageAccessIamPrincipal", # "DatabaseAuditLogService_QueryAuditLogFiles", # "DatabaseAuditLogService_GenerateAuditLogFileDownloadAddress", ] get_auditlog_config_cases = case_manager.get_cases_by_prefix(audlitlog_operations[0], "expected_status", ascending=False) for c in get_auditlog_config_cases: steps.append(convert_case_to_multistep_style(c)) update_auditlog_config_cases = case_manager.get_cases_by_prefix(audlitlog_operations[1], "expected_status", ascending=False) for c in update_auditlog_config_cases: steps.append(convert_case_to_multistep_style(c)) create_auditlog_config_cases = case_manager.get_cases_by_prefix(audlitlog_operations[2], "expected_status", ascending=False) for c in create_auditlog_config_cases: steps.append(convert_case_to_multistep_style(c)) list_auditlog_filter_rules_cases = case_manager.get_cases_by_prefix(audlitlog_operations[3], "expected_status", ascending=False) for c in list_auditlog_filter_rules_cases: steps.append(convert_case_to_multistep_style(c)) steps.append(create_auditlog_filter_rule_step) get_auditlog_filter_rule_cases = case_manager.get_cases_by_prefix(audlitlog_operations[4], "expected_status", ascending=False) for c in get_auditlog_filter_rule_cases: steps.append(convert_case_to_multistep_style(c)) delete_auditlog_filter_rule_cases = case_manager.get_cases_by_prefix(audlitlog_operations[5], "expected_status", ascending=False) for c in delete_auditlog_filter_rule_cases: steps.append(convert_case_to_multistep_style(c)) create_auditlog_filter_rule_cases = case_manager.get_cases_by_prefix(audlitlog_operations[6], "expected_status", ascending=False) for c in create_auditlog_filter_rule_cases: steps.append(convert_case_to_multistep_style(c)) steps.append(delete_cluster_step) ordered_steps = [] generation_id = case_manager.metadata["trace_id"] for i, step in enumerate(steps): if step.get("meta") and step["meta"]["category"] == "INVALID_METHOD": continue if step["request"]["path"].startswith("/"): step["request"]["path"] = "https://dedicated.dev.tidbapi.com/v1beta1" + step["request"]["path"] ordered_steps.append( OrderedDict( [ ("case_id", step["case_id"]), ("operation_id", step["operation_id"]), ("generation_id", generation_id), ("order", i + 1), ("description", step["description"]), ("case_type", step["case_type"]), ("request_type", step["request_type"]), ("request", json.dumps(step["request"], indent=2)), ("expect", json.dumps(step["expect"], indent=2)), ("step_type", step["step_type"]), ("max_retries", step["max_retries"]), ("delay_after", step["delay_after"]), ("name", step["name"]), ("continue_on_unexpected", step["continue_on_unexpected"]), ("save", json.dumps(step["save"], indent=2)), ("stage", step["stage"]), ("meta", json.dumps(step.get("meta", {}), indent=2)), ] ) ) return ordered_steps if __name__ == "__main__": main()