90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
import json
|
|
import sys
|
|
import subprocess
|
|
import os
|
|
import shutil
|
|
|
|
def get_tofu_binary():
|
|
# setup-opentofu installs a wrapper script as 'tofu' that mirrors output to console and sets action outputs.
|
|
# This causes massive log buffer overflow when running 'tofu show -json'.
|
|
# We try to find the real binary 'tofu-bin' located next to the wrapper.
|
|
tofu_path = shutil.which("tofu")
|
|
if tofu_path:
|
|
tofu_dir = os.path.dirname(tofu_path)
|
|
tofu_bin_path = os.path.join(tofu_dir, "tofu-bin")
|
|
if os.path.exists(tofu_bin_path):
|
|
return tofu_bin_path
|
|
return "tofu"
|
|
|
|
def validate_plan(plan_arg):
|
|
# Determine if input is json file or binary plan
|
|
# If it ends with .json, treat as file. If not (e.g. tfplan), run tofu show.
|
|
|
|
plan = {}
|
|
|
|
if plan_arg.endswith('.json'):
|
|
try:
|
|
with open(plan_arg, 'r') as f:
|
|
plan = json.load(f)
|
|
except FileNotFoundError:
|
|
print(f"Error: File {plan_arg} not found.")
|
|
sys.exit(1)
|
|
else:
|
|
print(f"Processing binary plan: {plan_arg}")
|
|
try:
|
|
# Stream the JSON output directly to json.load to save memory
|
|
# Helper: Redirect stderr to DEVNULL to avoid cluttering runner logs or buffer issues
|
|
tofu_cmd = get_tofu_binary()
|
|
with subprocess.Popen([tofu_cmd, "show", "-json", plan_arg], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) as proc:
|
|
if proc.stdout:
|
|
plan = json.load(proc.stdout)
|
|
else:
|
|
print("Error: No output from tofu show")
|
|
sys.exit(1)
|
|
|
|
if proc.wait() != 0:
|
|
print("Error: tofu show failed")
|
|
sys.exit(1)
|
|
|
|
except Exception as e:
|
|
print(f"Error capturing plan JSON: {e}")
|
|
sys.exit(1)
|
|
|
|
# planned_values contains the state AFTER the plan is applied
|
|
root_module = plan.get('planned_values', {}).get('root_module', {})
|
|
resources = root_module.get('resources', [])
|
|
|
|
# Check 1: Gitea Server must verify exist (VMID 114)
|
|
# Resource address: proxmox_lxc.lxc["114"]
|
|
gitea_exists = False
|
|
for res in resources:
|
|
if res.get('address') == 'proxmox_lxc.lxc["114"]':
|
|
gitea_exists = True
|
|
break
|
|
|
|
if not gitea_exists:
|
|
print("ERROR: Validation Failed! Gitea server (proxmox_lxc.lxc[\"114\"]) is missing in the planned state (it might be deleted).")
|
|
sys.exit(1)
|
|
|
|
# Check 2: At least one runner must remain
|
|
# Runner address pattern: proxmox_lxc.runners[...]
|
|
# We check if any resource starting with proxmox_lxc.runners exists in planned_values
|
|
runner_count = 0
|
|
for res in resources:
|
|
if res.get('address', '').startswith('proxmox_lxc.runners'):
|
|
runner_count += 1
|
|
|
|
if runner_count < 1:
|
|
print("ERROR: Validation Failed! No Gitea runners (proxmox_lxc.runners) found in the planned state.")
|
|
sys.exit(1)
|
|
|
|
print(f"Validation Successful: Gitea server exists and {runner_count} runners will be present.")
|
|
sys.exit(0)
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("Usage: python validate_plan.py <plan.json>")
|
|
sys.exit(1)
|
|
|
|
validate_plan(sys.argv[1])
|