Files
llm/scripts/check_powershell_scripts.py
2026-07-26 17:30:04 +03:00

431 lines
24 KiB
Python

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 verify_text.count("--require-form-write-plan-composition") < 2:
failures.append("verify_1c_adapter_deployment.ps1 must pass --require-form-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 '"skipped_no_saved_state_candidate"' not in verify_text:
failures.append("verify_1c_adapter_deployment.ps1 must accept a missing safe module candidate in optional saved-state smoke mode.")
if "$RequireWrite -and $report.status -in $skipStatuses" not in verify_text:
failures.append("verify_1c_adapter_deployment.ps1 must reject all saved-state module skip statuses in strict mode.")
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 "[switch]$SkipCodeWriteSavedStateSmoke" not in deploy_text:
failures.append("deploy_1c_adapter_stack.ps1 must declare -SkipCodeWriteSavedStateSmoke.")
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 '"-SkipCodeWriteSavedStateSmoke"' not in deploy_text:
failures.append("deploy_1c_adapter_stack.ps1 must forward -SkipCodeWriteSavedStateSmoke 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())