from __future__ import annotations import argparse import json import shutil import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DEFAULT_PATHS = [ROOT / "scripts", ROOT / "core" / "deploy"] def iter_powershell_scripts(paths: list[Path]) -> list[Path]: scripts: list[Path] = [] for path in paths: if path.is_file() and path.suffix.lower() == ".ps1": scripts.append(path) elif path.is_dir(): scripts.extend(path.rglob("*.ps1")) return sorted(set(scripts)) def powershell_executable() -> str | None: return shutil.which("pwsh") or shutil.which("powershell") def ps_single_quoted(value: str) -> str: return "'" + value.replace("'", "''") + "'" def check_script(executable: str, path: Path) -> tuple[bool, str]: path_literal = ps_single_quoted(str(path)) command = [ executable, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ( "$ErrorActionPreference = 'Stop'; " f"$path = {path_literal}; " "$errors = $null; " "[System.Management.Automation.PSParser]::Tokenize((Get-Content -Raw -LiteralPath $path), [ref]$errors) | Out-Null; " "if ($errors) { " " foreach ($errorItem in $errors) { " " Write-Error ('{0}: {1} at line {2}, column {3}' -f $path, $errorItem.Message, $errorItem.Token.StartLine, $errorItem.Token.StartColumn); " " }; " " exit 1 " "}" ), ] result = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, check=False) output = "\n".join(part for part in [result.stdout.strip(), result.stderr.strip()] if part) return result.returncode == 0, output def run_powershell_contract_command(executable: str, args: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run( [executable, "-NoProfile", "-ExecutionPolicy", "Bypass", *args], cwd=ROOT, text=True, capture_output=True, check=False, ) def combined_output(result: subprocess.CompletedProcess[str]) -> str: return "\n".join(part for part in [result.stdout.strip(), result.stderr.strip()] if part) def check_adapter_base_id_runtime(executable: str) -> list[str]: failures: list[str] = [] duplicate_verify = run_powershell_contract_command( executable, ["-File", "scripts/verify_1c_adapter_deployment.ps1", "-BaseId", "upo_test,upo_test", "-SkipRest", "-SkipMcp"], ) duplicate_verify_output = combined_output(duplicate_verify) if duplicate_verify.returncode == 0 or "Duplicate BaseId value(s): upo_test" not in duplicate_verify_output: failures.append( "verify_1c_adapter_deployment.ps1 must fail duplicate comma-separated -BaseId values at runtime." ) multi_verify = run_powershell_contract_command( executable, ["-File", "scripts/verify_1c_adapter_deployment.ps1", "-BaseId", "upo_test,another_test", "-SkipRest", "-SkipMcp"], ) multi_verify_output = combined_output(multi_verify) if multi_verify.returncode != 0 or "--base-id upo_test another_test" not in multi_verify_output: failures.append( "verify_1c_adapter_deployment.ps1 must expand comma-separated -BaseId values before calling the persisted report validator." ) duplicate_deploy = run_powershell_contract_command( executable, ["-File", "scripts/deploy_1c_adapter_stack.ps1", "-BaseId", "upo_test,upo_test", "-SkipRest", "-SkipMcp"], ) duplicate_deploy_output = combined_output(duplicate_deploy) if duplicate_deploy.returncode == 0 or "Duplicate BaseId value(s): upo_test" not in duplicate_deploy_output: failures.append( "deploy_1c_adapter_stack.ps1 must fail duplicate comma-separated -BaseId values at runtime." ) multi_deploy = run_powershell_contract_command( executable, ["-File", "scripts/deploy_1c_adapter_stack.ps1", "-BaseId", "upo_test,another_test", "-SkipRest", "-SkipMcp"], ) multi_deploy_output = combined_output(multi_deploy) if multi_deploy.returncode != 0 or "--base-id upo_test another_test" not in multi_deploy_output: failures.append( "deploy_1c_adapter_stack.ps1 must preserve multiple normalized -BaseId values when invoking verification." ) return failures def check_adapter_verify_wiring(scripts: list[Path], executable: str) -> list[str]: script_set = set(scripts) verify_path = ROOT / "scripts" / "verify_1c_adapter_deployment.ps1" deploy_path = ROOT / "scripts" / "deploy_1c_adapter_stack.ps1" stack_path = ROOT / "scripts" / "check_1c_adapter_verification_stack.py" readiness_path = ROOT / "scripts" / "check_1c_saved_state_strict_readiness.py" prepare_copy_sql_path = ROOT / "scripts" / "prepare_1c_saved_state_copy_sql.py" prepare_cleanup_sql_path = ROOT / "scripts" / "prepare_1c_saved_state_cleanup_sql.py" verify_copy_path = ROOT / "scripts" / "verify_1c_saved_state_copy.py" execute_copy_sql_path = ROOT / "scripts" / "execute_1c_saved_state_copy_sql.ps1" if verify_path not in script_set and deploy_path not in script_set: return [] failures: list[str] = [] if not verify_path.exists(): failures.append("scripts/verify_1c_adapter_deployment.ps1 is missing.") return failures if not deploy_path.exists(): failures.append("scripts/deploy_1c_adapter_stack.ps1 is missing.") return failures if not stack_path.exists(): failures.append("scripts/check_1c_adapter_verification_stack.py is missing.") return failures if not readiness_path.exists(): failures.append("scripts/check_1c_saved_state_strict_readiness.py is missing.") if not prepare_copy_sql_path.exists(): failures.append("scripts/prepare_1c_saved_state_copy_sql.py is missing.") if not prepare_cleanup_sql_path.exists(): failures.append("scripts/prepare_1c_saved_state_cleanup_sql.py is missing.") if not verify_copy_path.exists(): failures.append("scripts/verify_1c_saved_state_copy.py is missing.") return failures if not execute_copy_sql_path.exists(): failures.append("scripts/execute_1c_saved_state_copy_sql.ps1 is missing.") return failures verify_text = verify_path.read_text(encoding="utf-8", errors="replace") deploy_text = deploy_path.read_text(encoding="utf-8", errors="replace") stack_text = stack_path.read_text(encoding="utf-8", errors="replace") readiness_text = readiness_path.read_text(encoding="utf-8", errors="replace") prepare_copy_sql_text = prepare_copy_sql_path.read_text(encoding="utf-8", errors="replace") if prepare_copy_sql_path.exists() else "" prepare_cleanup_sql_text = prepare_cleanup_sql_path.read_text(encoding="utf-8", errors="replace") if prepare_cleanup_sql_path.exists() else "" verify_copy_text = verify_copy_path.read_text(encoding="utf-8", errors="replace") if verify_copy_path.exists() else "" execute_copy_sql_text = execute_copy_sql_path.read_text(encoding="utf-8", errors="replace") if execute_copy_sql_path.exists() else "" if "[switch]$RequireSelectorChainWritePlanComposition" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must declare -RequireSelectorChainWritePlanComposition.") if "[switch]$RequireSavedStateWriteSmoke" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must declare -RequireSavedStateWriteSmoke.") if "[string]$SavedStateTable" not in verify_text or '[ValidateSet("ConfigSave", "ConfigCASSave")]' not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must declare -SavedStateTable with ConfigSave/ConfigCASSave validation.") if verify_text.count("$SavedStateTable") < 4: failures.append("verify_1c_adapter_deployment.ps1 must use -SavedStateTable for copy plan and saved-state smoke commands.") if verify_text.count("--require-write-plan-composition") < 2: failures.append("verify_1c_adapter_deployment.ps1 must pass --require-write-plan-composition to both REST and MCP selector-chain smoke commands.") if "--allow-empty-saved-state" not in verify_text or "if (-not $RequireSavedStateWriteSmoke)" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must allow empty saved-state only when -RequireSavedStateWriteSmoke is not set.") if "function Get-DuplicateValues" not in verify_text or "Duplicate BaseId value(s)" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must reject duplicate -BaseId values before writing reports.") if "function Normalize-BaseIds" not in verify_text or '-split ","' not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must split comma-separated -BaseId values before verification.") if "function Assert-SelectorChainReport" not in verify_text or verify_text.count("Assert-SelectorChainReport") < 3: failures.append("verify_1c_adapter_deployment.ps1 must validate both persisted selector-chain JSON reports after smoke commands.") if "working_state" not in verify_text or 'did not use working state' not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must validate selector-chain working_state=working in persisted reports.") for function_name in ( "Assert-WritePlanSafetyReport", "Assert-WritePreflightReport", "Assert-WriteRollbackSafetyReport", "Assert-SavedStateDiffReport", "Assert-SavedStateChangesReport", "Assert-SavedStateFormWriteReport", "Assert-SavedStateModuleWriteReport", ): if f"function {function_name}" not in verify_text or verify_text.count(function_name) < 2: failures.append(f"verify_1c_adapter_deployment.ps1 must validate reports with {function_name}.") if "scripts/check_1c_verify_reports.py" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must run the offline persisted report validator.") if "scripts/plan_1c_saved_state_copy.py" not in verify_text or "saved-state-copy-plan.json" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must generate saved-state-copy-plan.json before persisted report validation.") if "scripts/prepare_1c_saved_state_copy_sql.py" not in verify_text or "prepare-saved-state-copy-sql.json" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must generate prepare-saved-state-copy-sql.json before persisted report validation.") if "scripts/prepare_1c_saved_state_cleanup_sql.py" not in verify_text or "cleanup-saved-state-copy-sql.json" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must generate cleanup-saved-state-copy-sql.json before persisted report validation.") if "scripts/check_1c_saved_state_strict_readiness.py" not in verify_text or "saved-state-strict-readiness.json" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must generate saved-state-strict-readiness.json before persisted report validation.") for flag in ( "--skip-rest", "--skip-mcp", "--skip-write-plan-safety-smoke", "--skip-write-rollback-safety-smoke", "--skip-saved-state-diff-smoke", "--skip-saved-state-write-smoke", "--require-saved-state-write-smoke", "--require-selector-chain-write-plan-composition", "--rest-adapter-url", "--mcp-url", "--saved-state-table", ): if flag not in verify_text: failures.append(f"verify_1c_adapter_deployment.ps1 must forward {flag} to scripts/check_1c_verify_reports.py.") verify_reports_self_test = subprocess.run( [sys.executable, "scripts/check_1c_verify_reports.py", "--self-test", "--json"], cwd=ROOT, text=True, capture_output=True, check=False, ) if verify_reports_self_test.returncode != 0: output = "\n".join(part for part in [verify_reports_self_test.stdout.strip(), verify_reports_self_test.stderr.strip()] if part) failures.append(f"scripts/check_1c_verify_reports.py --self-test failed: {output}") else: try: self_test_report = json.loads(verify_reports_self_test.stdout) except json.JSONDecodeError as exc: failures.append(f"scripts/check_1c_verify_reports.py --self-test --json returned invalid JSON: {exc}") else: expected_self_test_codes = { "strict_skip_failure_codes": { "selector_chain_composition_required", "saved_state_form_write_required", "saved_state_module_write_required", }, "coverage_failure_codes": { "selector_chain_write_plan_evidence_missing", "selector_chain_next_method_unexpected", "selector_chain_write_plan_target_not_boolean", "selector_chain_working_state_unexpected", }, "consistency_failure_codes": { "selector_chain_write_plan_target_not_composed", "selector_chain_composed_status_unexpected", }, "safety_failure_codes": { "write_plan_safety_check_field_unexpected", "write_plan_safety_check_missing", }, "rollback_safety_failure_codes": { "write_rollback_safety_check_field_unexpected", "write_rollback_safety_check_missing", }, "saved_state_diff_failure_codes": { "saved_state_diff_check_field_unexpected", "saved_state_diff_check_missing", }, "schema_failure_codes": { "report_schema_unexpected", }, "identity_failure_codes": { "report_base_id_unexpected", "report_transport_unexpected", }, "endpoint_failure_codes": { "report_endpoint_url_unexpected", }, "staleness_failure_codes": { "report_stale", }, "duplicate_failure_codes": { "duplicate_base_id", }, "saved_state_failure_codes": { "saved_state_form_route_write_plan_not_allowed", "saved_state_form_route_field_missing", "saved_state_module_rollback_missing", }, "saved_state_strict_readiness_failure_codes": { "saved_state_strict_readiness_required", "saved_state_strict_readiness_table_unexpected", }, "saved_state_copy_plan_failure_codes": { "saved_state_copy_plan_source_family_invalid", "saved_state_copy_plan_status_unexpected", "saved_state_copy_plan_source_row_table_unexpected", "saved_state_copy_plan_target_collisions_present", }, "saved_state_table_failure_codes": { "saved_state_table_mismatch", "saved_state_table_unexpected", }, } for field, expected_codes in expected_self_test_codes.items(): actual_codes = set(self_test_report.get(field) or []) missing_codes = sorted(expected_codes - actual_codes) if missing_codes: failures.append(f"scripts/check_1c_verify_reports.py --self-test must cover {field}: missing {missing_codes}.") if "ConvertFrom-Json" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must parse selector-chain JSON reports with ConvertFrom-Json.") if "--saved-state-table" not in readiness_text or "saved_state_table" not in readiness_text: failures.append("check_1c_saved_state_strict_readiness.py must support --saved-state-table and include it in reports.") for token in ("source_family.valid", "target_collisions.status", "sql_write_performed", "BEGIN TRANSACTION", "THROW 51001"): if token not in prepare_copy_sql_text: failures.append(f"prepare_1c_saved_state_copy_sql.py must include guarded SQL generation token: {token}.") for token in ("onec_saved_state_cleanup_sql_plan.v1", "DELETE t FROM", "BinarySHA1", "THROW 51102", "sql_write_performed"): if token not in prepare_cleanup_sql_text: failures.append(f"prepare_1c_saved_state_cleanup_sql.py must include guarded cleanup SQL token: {token}.") for token in ("onec_saved_state_copy_verify.v1", "blocked_missing_target_rows", "BinarySHA1", "sql_write_performed", "--require-ready"): if token not in verify_copy_text: failures.append(f"verify_1c_saved_state_copy.py must include read-only verification token: {token}.") for token in ( "[switch]$IUnderstandThisWritesToSql", "Refusing to execute SQL without -IUnderstandThisWritesToSql", "onec_saved_state_copy_sql_execution.v1", "onec_saved_state_copy_sql_plan.v1", "sql_execution_attempted", "sql_write_performed", "Get-FileHash", "scripts/verify_1c_saved_state_copy.py", "--require-ready", ): if token not in execute_copy_sql_text: failures.append(f"execute_1c_saved_state_copy_sql.ps1 must include guarded execution token: {token}.") if "[switch]$RequireSelectorChainWritePlanComposition" not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must declare -RequireSelectorChainWritePlanComposition.") if "[switch]$RequireSavedStateWriteSmoke" not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must declare -RequireSavedStateWriteSmoke.") if "[switch]$SkipWriteRollbackSafetySmoke" not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must declare -SkipWriteRollbackSafetySmoke.") if "[switch]$SkipSavedStateDiffSmoke" not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must declare -SkipSavedStateDiffSmoke.") if "[string]$SavedStateTable" not in deploy_text or '[ValidateSet("ConfigSave", "ConfigCASSave")]' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must declare -SavedStateTable with ConfigSave/ConfigCASSave validation.") if '"-RequireSelectorChainWritePlanComposition"' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must forward -RequireSelectorChainWritePlanComposition to verify_1c_adapter_deployment.ps1.") if '"-RequireSavedStateWriteSmoke"' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must forward -RequireSavedStateWriteSmoke to verify_1c_adapter_deployment.ps1.") if '"-SavedStateTable"' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must forward -SavedStateTable to verify_1c_adapter_deployment.ps1.") if '"-SkipWriteRollbackSafetySmoke"' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must forward -SkipWriteRollbackSafetySmoke to verify_1c_adapter_deployment.ps1.") if '"-SkipSavedStateDiffSmoke"' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must forward -SkipSavedStateDiffSmoke to verify_1c_adapter_deployment.ps1.") if "function Get-DuplicateValues" not in deploy_text or "Duplicate BaseId value(s)" not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must reject duplicate -BaseId values before invoking verification.") if "function Normalize-BaseIds" not in deploy_text or '-split ","' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must split comma-separated -BaseId values before invoking verification.") if '$baseIds -join ","' not in deploy_text: failures.append("deploy_1c_adapter_stack.ps1 must pass normalized BaseId values to nested verification as a comma-separated argument.") for flag in ("--rest-adapter-url", "--mcp-url", "--saved-state-table", "--max-report-age-seconds"): if flag not in stack_text: failures.append(f"check_1c_adapter_verification_stack.py must pass {flag} to scripts/check_1c_verify_reports.py.") for script_name in ( "scripts/smoke_1c_write_plan_safety.py", "scripts/smoke_1c_write_preflight.py", "scripts/smoke_1c_write_rollback_safety.py", "scripts/smoke_1c_saved_state_diff.py", "scripts/smoke_1c_saved_state_changes.py", "scripts/smoke_1c_saved_state_write_routes.py", "scripts/smoke_1c_saved_state_module_write.py", "scripts/check_1c_saved_state_strict_readiness.py", "scripts/plan_1c_saved_state_copy.py", "scripts/prepare_1c_saved_state_copy_sql.py", "scripts/prepare_1c_saved_state_cleanup_sql.py", "scripts/verify_1c_saved_state_copy.py", ): if script_name not in stack_text: failures.append(f"check_1c_adapter_verification_stack.py must py_compile {script_name}.") if 'nargs="+"' not in stack_text or "*args.base_id" not in stack_text: failures.append("check_1c_adapter_verification_stack.py must support multiple --base-id values and forward them to scripts/check_1c_verify_reports.py.") if "duplicate_base_id" not in stack_text: failures.append("check_1c_adapter_verification_stack.py must reject duplicate --base-id values.") failures.extend(check_adapter_base_id_runtime(executable)) return failures def main() -> int: parser = argparse.ArgumentParser(description="Check PowerShell script syntax with PSParser.") parser.add_argument("paths", nargs="*", type=Path, help="Files or directories to scan. Defaults to scripts/ and core/deploy/.") args = parser.parse_args() executable = powershell_executable() if not executable: print("PowerShell executable not found.", file=sys.stderr) return 1 paths = [path if path.is_absolute() else ROOT / path for path in args.paths] if args.paths else DEFAULT_PATHS scripts = iter_powershell_scripts(paths) if not scripts: print("No PowerShell scripts found.") return 0 failures = [] for script in scripts: ok, output = check_script(executable, script) if not ok: failures.append((script, output)) contract_failures = check_adapter_verify_wiring(scripts, executable) if failures or contract_failures: print("PowerShell script check failed:", file=sys.stderr) for script, output in failures: print(f"- {script.relative_to(ROOT)}", file=sys.stderr) if output: print(output, file=sys.stderr) for failure in contract_failures: print(f"- adapter verify wiring: {failure}", file=sys.stderr) return 1 print(f"Validated {len(scripts)} PowerShell script(s).") return 0 if __name__ == "__main__": raise SystemExit(main())