import argparse import subprocess import sys from dataclasses import dataclass from typing import Dict, List # edited by glg @dataclass(frozen=True) class TestStage: marker: str paths: List[str] TEST_STAGES: Dict[str, TestStage] = { "unit": TestStage( marker="unit and not perf_smoke", paths=[ "tests/p1/test_transaction_export_value_utils.py", "tests/p1/test_transaksi_value_utils.py", "tests/p1/test_dashboard_value_utils.py", "tests/p1/test_transaksi_harga_validation_service.py", "tests/p1/test_transaksi_payment_info_service.py", "tests/p1/test_transaksi_payment_state_service.py", "tests/p1/test_transaksi_payment_persist_use_case_service.py", "tests/p1/test_transaksi_autocomplete_async_use_case_service.py", "tests/p1/test_transaksi_barang_mutation_use_case_service.py", "tests/p1/test_dashboard_window_ui_service.py", "tests/p1/test_transaction_export_settlement_payment_utils.py", "tests/p1/test_transaction_export_legacy_defaults_service.py", "tests/p1/test_free_produk_outbox_drain_use_case_service.py", "tests/p1/test_printer_settings_value_service.py", "tests/p1/test_printer_settings_ui_state_service.py", "tests/p1/test_device_registration_request_builder.py", "tests/p1/test_device_registration_response_mapper.py", "tests/p1/test_device_registration_retry_policy.py", "tests/p1/test_config_dialog_view.py", "tests/test_device_registration_service.py", "tests/p1/test_transaksi_view_summary_service.py", "tests/p1/test_transaksi_async_payload_guards.py", "tests/p2/test_event_ingestion_backpressure_service.py", "tests/p2/test_event_outbox_service.py", "tests/p2/test_dr_backup_policy_service.py", "tests/p2/test_dr_restore_drill_service.py", "tests/p2/test_fleet_rollout_service.py", "tests/p2/test_fleet_rollout_gate_service.py", "tests/p2/test_dashboard_event_backpressure_delay.py", "tests/p2/test_export_runtime_metrics_service.py", "tests/p1/test_printer_diagnostic_service.py", "tests/p1/test_rawinput_scanner_service_hardening.py", "tests/test_base_service.py", # edited by glg # Regression guard: bug settlement identifier qualified harus tertangkap di stage unit. "tests/test_settlement_model.py", "tests/test_settlement_orchestrator_service.py", "tests/test_retry_and_circuit_breaker.py", "tests/test_export_upload_api_service.py", "tests/test_modules_quality_guard_v90.py", "tests/p1/test_sql_identifier_utils.py", "tests/p1/test_sql_query_builder.py", "tests/p1/test_auth_security_policy.py", "tests/p1/test_employee_seed_use_case_service.py", "tests/p2/test_event_ingestion_gateway_service.py", "tests/p2/test_fleet_rollout_orchestrator_service.py", "tests/p2/test_event_first_export_runtime_service.py", "tests/p2/test_fleet_rollout_runtime_guard_service.py", "tests/p2/test_dashboard_event_first_runtime_flow.py", "tests/p2/test_run_dr_fleet_pipeline.py", ], ), "integration": TestStage( marker="integration", paths=[ "tests/p1/test_lock_contention_soak.py", ], ), "critical_flow": TestStage( marker="critical_flow", paths=[ "tests/test_transaksi_persist_flow_service.py", "tests/test_transaksi_penjualan_controller_save_error_handling.py", "tests/test_load_transaksi_controller_load_flow.py", "tests/test_load_transaksi_controller_delete_guard.py", "tests/test_free_produk_sync_outbox_service.py", "tests/test_transaksi_enterprise_control_service.py", "tests/test_penjualan_quality_guard.py", ], ), "smoke": TestStage( marker="smoke", paths=[ "tests/p1/test_runtime_smoke.py", ], ), "gui_stability": TestStage( marker="not perf_smoke", paths=[ "tests/test_free_produk_relation_flow.py", "tests/test_gui_click_automation.py", "tests/test_pembatalan_transaksi_view_gui.py", ], ), "perf_smoke": TestStage( marker="perf_smoke", paths=[ "tests/p1/test_sync_export_perf_smoke.py", "tests/p1/test_lock_contention_soak.py", "tests/p1/test_ui_event_loop_burst.py", "tests/p1/test_event_outbox_backlog_burst_perf_smoke.py", "tests/p2/test_10000_cabang_event_outbox_perf_smoke.py", ], ), } STAGE_ORDER = [ "quality", "static_analysis", "unit", "critical_flow", "integration", "smoke", "gui_stability", "perf_smoke", ] def _run_command(args: List[str]) -> None: print("$", " ".join(args)) completed = subprocess.run(args, check=False) if completed.returncode != 0: raise SystemExit(completed.returncode) def _run_quality_stage() -> None: _run_command([sys.executable, "-m", "compileall", "-q", "pypos", "tests", "scripts"]) # edited by glg _run_command([sys.executable, "scripts/ci/run_penjualan_quality_guard.py"]) _run_command([sys.executable, "scripts/ci/run_modules_quality_guard_v90.py"]) def _run_static_analysis_stage() -> None: # edited by glg # Stage static analysis: lint/type/security scanner baseline. _run_command([sys.executable, "scripts/ci/run_static_analysis_gate.py"]) def _run_test_stage(stage_name: str) -> None: stage = TEST_STAGES.get(stage_name) if stage is None: raise SystemExit(f"Stage tidak dikenal: {stage_name}") if not stage.paths: raise SystemExit(f"Stage {stage_name} tidak memiliki daftar test path.") cmd = [ sys.executable, "-m", "pytest", "-q", "--maxfail=1", "-m", stage.marker, *stage.paths, ] _run_command(cmd) def run_stage(stage_name: str) -> None: if stage_name == "quality": _run_quality_stage() return if stage_name == "static_analysis": _run_static_analysis_stage() return _run_test_stage(stage_name) def parse_args(): parser = argparse.ArgumentParser( description=( "Jalankan quality gate staged " "(quality/static_analysis/unit/critical_flow/integration/smoke/perf_smoke/all)." ) ) parser.add_argument( "--stage", default="all", choices=["all"] + STAGE_ORDER, help="Stage yang dijalankan.", ) return parser.parse_args() def main(): args = parse_args() if args.stage == "all": for stage_name in STAGE_ORDER: print(f"\n=== QUALITY GATE STAGE: {stage_name} ===") run_stage(stage_name) print("\nSemua stage quality gate lulus.") return print(f"\n=== QUALITY GATE STAGE: {args.stage} ===") run_stage(args.stage) print("\nStage quality gate lulus.") if __name__ == "__main__": main()