Programming and Automation

This comprehensive guide covers programming and automation, including exam topics, example scripts, core components, monitoring tools, and much more.
Check out the headings to find what you’re looking for!

In a Nutshell

Automation through programming streamlines IT processes, improves efficiency, and reduces errors. Scripts, programming languages, and automation tools are deployed strategically to achieve this.

Key Overview

Repetitive tasks such as backups, user management, and deployments can be automated through scripting languages like Bash, PowerShell, or Python. Scheduling mechanisms (Cron, Task Scheduler), monitoring, logging, and security checks round out the picture.

Vendor-specific tools often come with their own scripting components (for example, VMware PowerCLI). Frequently, you’ll also need to analyze existing code and adapt it carefully to integrate automation into established systems.

You’ll regularly need to modify files you’ve created or received. Sometimes you’ll want to replace numerous “formatters” or other elements via variables, or search for and remove a link. That’s where simple Python, Bash, or PowerShell scripts come in handy.

Exam-Relevant Key Points

Selecting the right programming language for the task

Choosing the right language is crucial for automation success. Bash suits simple system tasks on Linux, while Python is preferred for complex logic and cross-platform solutions. PowerShell is the natural choice for Windows environments with deep API integration.

Automation reduces manual intervention and errors

Automated processes minimize human error and deliver consistent results. Recurring tasks like backups, user provisioning, and deployments execute reliably and with full auditability.

Scripting languages for system tasks and process logic

Scripts enable automation of system tasks (file management, process control) and complex process logic (conditionals, loops, error handling). They support rapid development and straightforward customization.

Platform-specific tools: Bash (Linux), PowerShell (Windows)

Bash is the standard shell scripting and system automation tool on Linux. PowerShell offers object-based processing and deep integration with Windows systems and APIs. Both are relevant for IT certification exams covering system administration.

Monitoring through logs, exit codes, and triggers

Automated processes must be monitored. Logging documents all actions, exit codes signal success or failure, and triggers enable event-driven automation. This ensures transparency and fault tolerance.

Sensitive processes require logging and permission checks

For sensitive automation (user rights, system changes), logging and permission validation are essential. Only authorized users should execute critical scripts, and all actions must be logged.

Reduced effort for maintenance and system upkeep

Automation significantly cuts manual effort in maintenance and system upkeep. Once written, scripts can be reused many times and easily adjusted, saving time and resources.

Document and version all changes

All modifications to automation scripts must be documented and version controlled. This ensures traceability, enables rollbacks, and supports team collaboration.

Core Components

1. Language selection

The right choice depends on the task and platform. Python works well for complex logic and cross-platform needs, Bash for Linux system tasks, and PowerShell for Windows environments.

2. Platform-specific scripting tools

Each platform offers its own automation tools: Bash on Linux, PowerShell on Windows, Cron for scheduling on Linux, Task Scheduler on Windows. These enable direct system integration.

3. Define automation goals

Before you start, establish clear objectives: which tasks should be automated? What efficiency gains are expected? What security requirements apply?

4. Conditional execution and loops

Automation often requires conditional logic (if/else) and loops (for/while) for recurring processes. These control structures enable flexible and adaptive automation.

5. Error handling and logging

Robust automation needs comprehensive error handling (try/catch, exit codes) and detailed logging. This ensures stability and auditability.

6. Scheduled tasks (Cron, Task Scheduler)

Cron on Linux and Task Scheduler on Windows enable time-based script execution. This is ideal for regular maintenance like backups or cleanup.

7. Integration with existing systems

Automation scripts must fit into your existing infrastructure. This covers APIs, databases, file systems, and other IT components.

8. Version control for scripts

Keep all automation scripts under version control (Git). This tracks changes, supports collaboration, and enables safe rollbacks.

9. Security checks and permission models

Automation demands strict security: permission checks, role management, secure credential handling, and audit logging for sensitive operations.

10. Monitor automated workflows

Automated processes need oversight: success validation, error alerts, performance monitoring, and alarms when things deviate.

Simple Practical Example (Python Backup)

import shutil

shutil.copy('/home/user/data.db', '/backup/data.db')

Explanation: This script automatically copies a file to a backup directory and can run regularly via a Cron job.

Why shutil instead of basic copy commands?

shutil stands for “shell utilities” and is a Python standard library module for high-level file operations. It has several advantages over bare copy commands:

Benefits of shutil.copy():

• Cross-platform: Works identically on Windows, Linux, and macOS
• Error handling: Raises clean exceptions for problems (permissions, missing paths)
• Metadata: Preserves file permissions and metadata (with shutil.copy2())
• Safe: Checks automatically for overwrite issues

Alternatives and why they fall short:

# ❌ Bad: OS-specific
import os
os.system('cp /home/user/data.db /backup/data.db')  # Linux only
os.system('copy C:\\data.db C:\\backup\\data.db')   # Windows only

# ❌ Bad: No error handling
f = open('/home/user/data.db', 'rb')
data = f.read()
f.close()
f = open('/backup/data.db', 'wb')
f.write(data)
f.close()

# ✅ Good: shutil.copy()
import shutil
shutil.copy('/home/user/data.db', '/backup/data.db')

shutil.copy() versus shutil.copy2():

• copy(): Copies content and basic permissions
• copy2(): Copies content and all metadata (timestamps, permissions) — usually the better choice for backups

Advantages and disadvantages

Advantages

• Error prevention through automated processes
• Time savings on repetitive tasks
• Traceability through logging
• Cross-platform capability with portable scripting languages

Disadvantages

• Errors in scripts can cause significant damage
• Requires testing and proper logging
• Inconsistent behavior across different environments

Common exam questions (with brief answers)

1. When should you use Bash instead of Python? For simple system tasks on Linux (file operations, process management, pipes).
2. What are typical automation tasks? Backups, user provisioning, log archival, deployment.
3. How do you make automation secure? Logging, permission checks, code reviews, evaluating exit statuses.
4. What is a cron job? A time-triggered task on Linux.
5. What are PowerShell’s advantages over Bash? Object-based architecture and deep integration with Windows and APIs.

Free response

This topic combines language choice, platform dependencies, and automation goals. Exams often focus on reading scripts, extending them, and correctly implementing error handling and logging.

Learning strategy for this topic

1. Understanding fundamentals: automate a simple backup locally

Solution:

# Bash backup script
#!/bin/bash
SOURCE="/home/user/documents"
TARGET="/backup/documents_$(date +%Y%m%d)"
mkdir -p "$TARGET"
cp -r "$SOURCE"/* "$TARGET/"
echo "Backup completed: $TARGET" >> /var/log/backup.log

Set up cron: 0 2 * * * /path/to/backup.sh

2. Deeper practice: outline an automation workflow (user provisioning + permissions)

Solution:

# Python user provisioning with permission checks
import subprocess
import logging

logging.basicConfig(filename='user_management.log', level=logging.INFO)

def create_user(username, groups):
    try:
        # Create user
        subprocess.run(['useradd', '-m', username], check=True)
        
        # Assign groups
        for group in groups:
            subprocess.run(['usermod', '-aG', group, username], check=True)
        
        logging.info(f"User {username} created with groups: {groups}")
        print(f"User {username} successfully created")
        
    except subprocess.CalledProcessError as e:
        logging.error(f"Error creating user {username}: {e}")
        print(f"Error: {e}")

# Usage
create_user("newuser", ["sudo", "developers"])

3. Exam-focused training: analyze and enhance scripts

Solution:

• Analysis: Check scripts for error handling, logging, and security
• Enhancement: Evaluate exit codes, validate parameters, improve error messages
• Example improvement:

import sys
import os

def validate_path(path):
    if not os.path.exists(path):
        print(f"Error: path {path} does not exist")
        sys.exit(1)
    return True

# Usage
if len(sys.argv) < 2:
    print("Usage: script.py <path>")
    sys.exit(1)

validate_path(sys.argv[1])
# Rest of script...

4. Preventing errors: add logging and permission checks

Solution:

import logging
import os
import sys
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/automation.log'),
        logging.StreamHandler()
    ]
)

def check_permissions(required_uid=0):
    """Check if script runs with correct permissions"""
    if os.geteuid() != required_uid:
        logging.error(f"Script must run as UID {required_uid}")
        sys.exit(1)

def safe_operation(operation_func):
    """Decorator for safe operations"""
    def wrapper(*args, **kwargs):
        try:
            result = operation_func(*args, **kwargs)
            logging.info(f"Operation {operation_func.__name__} succeeded")
            return result
        except Exception as e:
            logging.error(f"Error in {operation_func.__name__}: {e}")
            raise
    return wrapper

@safe_operation
def critical_operation():
    # Critical operation here
    pass

Topic analysis

• Technical core: Shell, Python, scheduling, logging
• Implementation challenges: platform differences, error prevention
• Security implications: permission management, risks of unsecured scripts
• Documentation requirements: version control, change logs
• Business value: reduced manual effort through repeatability

Current software products

Automation platforms

• Ansible: Configuration management and automation (open source)
• Jenkins: CI/CD pipelines with extensive plugin support
• GitHub Actions: Cloud-native CI/CD directly in your repository
• GitLab CI/CD: Integrated pipeline solution with Auto DevOps

Ansible: Up and Running

Scripting tools

• Visual Studio Code: Modern editor with Python and PowerShell support
• PyCharm: Python IDE with debugging and testing features
• Windows Terminal: Modern terminal environment for PowerShell
• WSL (Windows Subsystem for Linux): Linux environment on Windows

Monitoring and logging

• Prometheus: Metric collection and alerting
• Grafana: Monitoring data visualization
• ELK Stack: Elasticsearch, Logstash, Kibana for log analysis
• Nagios: Classic monitoring system

Cloud automation

• Azure Automation: PowerShell and Python runbooks in Azure
• AWS Lambda: Serverless functions for automation
• Google Cloud Functions: Event-driven automation
• Terraform: Infrastructure as Code for multi-cloud

Questions and answers

Question 1: Which language for which automation task?

Answer: Bash for simple Linux system tasks, PowerShell for Windows system integration, Python for complex logic and cross-platform solutions.

Question 2: How do you secure automation scripts?

Answer: Version control with Git, code reviews, comprehensive error handling, logging, permission checks, and regular security audits.

Question 3: What are common automation mistakes?

Answer: Missing error handling, insufficient logging, hardcoded paths, lack of tests, ignoring exit codes.

Question 4: How do you test automation scripts?

Answer: Unit tests for functions, integration tests for workflows, staging environments for production testing, monitoring in live operation.

Question 5: What are alternatives to cron?

Answer: systemd timers (modern alternative), Jenkins scheduled jobs, Kubernetes CronJobs, cloud-specific schedulers (AWS EventBridge).

Further resources

1. http://linuxcommand.org/lc3_learning_the_shell.php
2. https://learn.microsoft.com/en-us/powershell/
3. https://automatetheboringstuff.com/
4. https://docs.github.com/en/actions
5. https://crontab.guru/

Book recommendation

Deepen your knowledge of Python automation with this practical guide for beginners and advanced users. This book provides concrete solutions for real-world automation tasks and helps you develop efficient scripts.

Automate the Boring Stuff with Python: Practical Programming for Total Beginners

This book is ideal for:

• Beginners who want to learn Python for automation
• System administrators who need to automate repetitive tasks
• Developers who want to optimize their daily workflows
• Anyone looking for practical Python solutions to everyday problems

A collection of small automation scripts for Windows/Linux/Mac

Here are 15 practical Python scripts for common automation tasks across different platforms.

Linux scripts (5)

1. Resize and optimize images

from PIL import Image
import os
from pathlib import Path

def resize_images(folder_path, max_size=1920, quality=85):
    """Resizes all images in a folder"""
    for file in Path(folder_path).glob('*'):
        if file.suffix.lower() in ['.jpg', '.jpeg', '.png']:
            with Image.open(file) as img:
                img.thumbnail((max_size, max_size))
                img.save(file, quality=quality, optimize=True)
                print(f"Optimized: {file.name}")

# Usage: resize_images('/home/user/Pictures')

2. Find duplicate files in a folder

import hashlib
from pathlib import Path
from collections import defaultdict

def find_duplicates(folder_path):
    """Finds duplicate files based on MD5 hash"""
    hashes = defaultdict(list)
    
    for file in Path(folder_path).rglob('*'):
        if file.is_file():
            file_hash = hashlib.md5(file.read_bytes()).hexdigest()
            hashes[file_hash].append(file)
    
    duplicates = {h: files for h, files in hashes.items() if len(files) > 1}
    
    for hash_val, files in duplicates.items():
        print(f"Duplicates found ({len(files)} files):")
        for file in files:
            print(f"  - {file}")

# Usage: find_duplicates('/home/user/Downloads')

3. Clean up old log files automatically

import os
from pathlib import Path
from datetime import datetime, timedelta

def clean_old_logs(log_folder, days=30):
    """Deletes log files older than X days"""
    cutoff_date = datetime.now() - timedelta(days=days)
    
    for file in Path(log_folder).glob('*.log'):
        if datetime.fromtimestamp(file.stat().st_mtime) < cutoff_date:
            file.unlink()
            print(f"Deleted: {file.name}")

# Usage: clean_old_logs('/var/log/myapp', days=30)

4. Merge PDF files

from PyPDF2 import PdfMerger
from pathlib import Path

def merge_pdfs(folder_path, output_name='combined.pdf'):
    """Combines all PDFs in a folder"""
    merger = PdfMerger()
    
    for file in sorted(Path(folder_path).glob('*.pdf')):
        merger.append(file)
        print(f"Added: {file.name}")
    
    merger.write(output_name)
    merger.close()
    print(f"Created: {output_name}")

# Usage: merge_pdfs('/home/user/Documents/PDFs')

5. System monitoring and alerts

import psutil
import smtplib
from email.mime.text import MIMEText

def check_disk_usage(threshold=90):
    """Checks disk usage and sends alert when threshold is exceeded"""
    for partition in psutil.disk_partitions():
        usage = psutil.disk_usage(partition.mountpoint)
        percent = usage.percent
        
        if percent > threshold:
            print(f"ALERT: {partition.mountpoint} is {percent}% full")
            # You could send an email alert here
            # send_alert_email(partition.mountpoint, percent)

# Usage: check_disk_usage(threshold=90)

Windows scripts (5)

1. Clean up the desktop

import shutil
from pathlib import Path
from datetime import datetime

def clean_downloads():
    """Organizes files in the Downloads folder"""
    downloads = Path.home() / 'Downloads'
    
    # Create folders
    folders = {
        'Pictures': ['.jpg', '.jpeg', '.png', '.gif'],
        'Documents': ['.pdf', '.doc', '.docx', '.txt'],
        'Music': ['.mp3', '.wav', '.flac'],
        'Videos': ['.mp4', '.avi', '.mkv'],
        'Archives': ['.zip', '.rar', '.7z']
    }
    
    for folder, extensions in folders.items():
        target = downloads / folder
        target.mkdir(exist_ok=True)
        
        for file in downloads.glob('*'):
            if file.suffix.lower() in extensions and file.is_file():
                shutil.move(str(file), str(target / file.name))
                print(f"Moved: {file.name} -> {folder}")

# Usage: clean_downloads()

2. Delete temporary files

import shutil
import os
from pathlib import Path

def clean_temp_files():
    """Deletes temporary files and cache"""
    temp_folders = [
        Path(os.environ.get('TEMP', '')),
        Path(os.environ.get('TMP', '')),
        Path.home() / 'AppData' / 'Local' / 'Temp',
    ]
    
    for folder in temp_folders:
        if folder.exists():
            try:
                for file in folder.glob('*'):
                    if file.is_file():
                        file.unlink()
                print(f"Cleaned: {folder}")
            except PermissionError:
                print(f"Access denied: {folder}")

# Usage: clean_temp_files()

3. Scan WiFi networks

import subprocess

def scan_wifi_networks():
    """Displays available WiFi networks"""
    try:
        result = subprocess.run(
            ['netsh', 'wlan', 'show', 'networks'],
            capture_output=True,
            text=True
        )
        
        networks = []
        for line in result.stdout.split('\n'):
            if 'SSID' in line:
                ssid = line.split(':')[1].strip()
                networks.append(ssid)
        
        print("Available networks:")
        for network in networks:
            print(f"  - {network}")
            
    except Exception as e:
        print(f"Error: {e}")

# Usage: scan_wifi_networks()

4. Clipboard manager

import pyperclip
import time
from datetime import datetime

def clipboard_history(max_entries=10):
    """Tracks clipboard history"""
    history = []
    last_content = ""
    
    print("Clipboard monitoring started (Ctrl+C to exit)")
    
    try:
        while True:
            current = pyperclip.paste()
            
            if current != last_content and current:
                timestamp = datetime.now().strftime("%H:%M:%S")
                history.insert(0, f"[{timestamp}] {current[:50]}...")
                
                if len(history) > max_entries:
                    history.pop()
                
                print(f"\nNew entry: {current[:30]}...")
                for entry in history:
                    print(f"  {entry}")
                
                last_content = current
            
            time.sleep(2)
            
    except KeyboardInterrupt:
        print("\nClipboard monitoring stopped")

# Usage: clipboard_history()

5. Screenshot on hotkey press

import pyautogui
import keyboard
from datetime import datetime
from pathlib import Path

def screenshot_on_hotkey(hotkey='f9'):
    """Takes a screenshot when a hotkey is pressed"""
    screenshots = Path.home() / 'Desktop' / 'Screenshots'
    screenshots.mkdir(exist_ok=True)
    
    def take_screenshot():
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = screenshots / f"screenshot_{timestamp}.png"
        pyautogui.screenshot(str(filename))
        print(f"Screenshot saved: {filename}")
    
    keyboard.add_hotkey(hotkey, take_screenshot)
    print(f"Press {hotkey.upper()} for screenshot")
    keyboard.wait()  # Waits for keypresses

# Usage: screenshot_on_hotkey('f9')

Mac Scripts (5)

1. Automating Finder Organization

import shutil
from pathlib import Path

def organize_downloads():
    """Organizes Mac Downloads folder"""
    downloads = Path.home() / 'Downloads'
    
    file_types = {
        'Bilder': ['.jpg', '.jpeg', '.png', '.heic', '.gif'],
        'Dokumente': ['.pdf', '.pages', '.doc', '.docx', '.txt'],
        'Musik': ['.mp3', '.m4a', '.wav', '.aiff'],
        'Videos': ['.mp4', '.mov', '.avi', '.mkv'],
        'Archiv': ['.zip', '.rar', '.7z', '.dmg']
    }
    
    for folder, extensions in file_types.items():
        target = downloads / folder
        target.mkdir(exist_ok=True)
        
        for file in downloads.glob('*'):
            if file.suffix.lower() in extensions and file.is_file():
                shutil.move(str(file), str(target / file.name))
                print(f"Organisiert: {file.name}")

# Nutzung: organize_downloads()

2. Sending macOS Notifications

import subprocess

def send_notification(title, message):
    """Sends a macOS system notification"""
    cmd = f"""
    osascript -e 'display notification "{message}" with title "{title}"'
    """
    subprocess.run(cmd, shell=True)

# Nutzung: send_notification("Backup fertig", "Dein Backup wurde erfolgreich abgeschlossen")

3. Controlling Spotify

import subprocess

def spotify_control(action):
    """Controls Spotify using AppleScript"""
    commands = {
        'play': 'tell application "Spotify" to play',
        'pause': 'tell application "Spotify" to pause',
        'next': 'tell application "Spotify" to next track',
        'previous': 'tell application "Spotify" to previous track'
    }
    
    if action in commands:
        cmd = f'osascript -e \'tell application "Spotify" to {commands[action]}\''
        subprocess.run(cmd, shell=True)
        print(f"Spotify: {action}")
    else:
        print(f"Unbekannter Befehl: {action}")

# Nutzung: spotify_control('play')

4. Adjusting Screen Brightness

import subprocess

def set_brightness(level):
    """Sets screen brightness (0-100)"""
    if 0 <= level <= 100:
        cmd = f"""
        osascript -e 'tell application "System Events" to set brightness of every display to {level/100}'
        """
        subprocess.run(cmd, shell=True)
        print(f"Helligkeit auf {level}% gesetzt")
    else:
        print("Helligkeit muss zwischen 0 und 100 liegen")

# Nutzung: set_brightness(50)

5. Extracting Text from PDFs

import PyPDF2
from pathlib import Path

def extract_pdf_text(pdf_path):
    """Extracts text from a PDF file"""
    text = ""
    
    with open(pdf_path, 'rb') as file:
        reader = PyPDF2.PdfReader(file)
        
        for page in reader.pages:
            text += page.extract_text() + "\n"
    
    return text

def save_pdf_text_to_file(pdf_path, output_path):
    """Saves extracted text to a file"""
    text = extract_pdf_text(pdf_path)
    
    with open(output_path, 'w', encoding='utf-8') as file:
        file.write(text)
    
    print(f"Text extrahiert: {output_path}")

# Nutzung: save_pdf_text_to_file('/Users/user/Dokument.pdf', '/Users/user/Dokument.txt')

Usage Notes

1. Install required packages:

   pip install pillow pyperclip pyautogui pypdf2 psutil

2. Permissions: Some scripts require administrator privileges

3. Adjust paths: Update paths to match your local environment

4. Test first: Always test scripts with sample data before running them

5. Backups: Create backups before executing deletion operations

Bash and PowerShell Script Collection

macOS comes with Bash by default (or zsh as the default shell in newer macOS versions). Here are 15 additional scripts in native shell languages.

Linux Bash Scripts (5)

1. Automatic Backup with Rotation

#!/bin/bash
# backup.sh - Automatic backup with rotation

SOURCE_DIR="/home/user/documents"
BACKUP_DIR="/backup"
MAX_BACKUPS=7
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_NAME="backup_$DATE.tar.gz"

# Create backup
tar -czf "$BACKUP_DIR/$BACKUP_NAME" "$SOURCE_DIR"
echo "Backup erstellt: $BACKUP_NAME"

# Delete old backups (rotation)
cd "$BACKUP_DIR"
ls -t backup_*.tar.gz | tail -n +$((MAX_BACKUPS + 1)) | xargs -r rm
echo "Alte Backups bereinigt (max $MAX_BACKUPS behalten)"

2. Server Status Check

#!/bin/bash
# server_check.sh - Checks server status and services

echo "=== Server Status Check ==="
echo "Zeit: $(date)"
echo ""

# CPU load
echo "CPU-Last:"
top -bn1 | grep "Cpu(s)" | sed "s/.*, *\([0-9.]*\)%id.*/\1/" | awk '{print 100 - $1"%"}'

# Memory
echo ""
echo "Speicher:"
free -h

# Disk
echo ""
echo "Festplatte:"
df -h

# Check important services
echo ""
echo "Services:"
for service in nginx mysql docker; do
    if systemctl is-active --quiet "$service"; then
        echo "✓ $service läuft"
    else
        echo "✗ $service läuft nicht"
    fi
done

3. Analyzing Log Files

#!/bin/bash
# log_analyzer.sh - Analyzes log files for errors

LOG_FILE="/var/log/syslog"
ERROR_COUNT=$(grep -i "error" "$LOG_FILE" | wc -l)
WARNING_COUNT=$(grep -i "warning" "$LOG_FILE" | wc -l)

echo "=== Log-Analyse für $LOG_FILE ==="
echo "Fehler: $ERROR_COUNT"
echo "Warnungen: $WARNING_COUNT"
echo ""

# Show last 10 errors
if [ $ERROR_COUNT -gt 0 ]; then
    echo "Letzte 10 Fehler:"
    grep -i "error" "$LOG_FILE" | tail -n 10
fi

4. Finding Large Files

#!/bin/bash
# find_large_files.sh - Finds large files

DIRECTORY="/home/user"
MIN_SIZE="100M"  # Minimum size

echo "=== Dateien größer als $MIN_SIZE in $DIRECTORY ==="
find "$DIRECTORY" -type f -size +"$MIN_SIZE" -exec ls -lh {} \; | awk '{print $5, $9}'

5. Process Monitoring

#!/bin/bash
# process_monitor.sh - Monitors specific processes

PROCESSES=("nginx" "mysql" "redis" "docker")

echo "=== Prozess-Monitoring ==="
for process in "${PROCESSES[@]}"; do
    if pgrep -x "$process" > /dev/null; then
        PID=$(pgrep -x "$process")
        MEMORY=$(ps -p "$PID" -o %mem --no-headers)
        CPU=$(ps -p "$PID" -o %cpu --no-headers)
        echo "✓ $process (PID: $PID, RAM: $MEMORY%, CPU: $CPU%)"
    else
        echo "✗ $process läuft nicht"
    fi
done

Windows PowerShell Scripts (5)

1. Gathering System Information

# system_info.ps1 - Collects detailed system information

Write-Host "=== System-Informationen ===" -ForegroundColor Cyan

# System info
Write-Host "`nSystem:" -ForegroundColor Yellow
Get-CimInstance Win32_ComputerSystem | Select-Object Manufacturer, Model, TotalPhysicalMemory

# Operating system
Write-Host "`nBetriebssystem:" -ForegroundColor Yellow
Get-CimInstance Win32_OperatingSystem | Select-Object Caption, Version, BuildNumber

# CPU
Write-Host "`nCPU:" -ForegroundColor Yellow
Get-CimInstance Win32_Processor | Select-Object Name, NumberOfCores, MaxClockSpeed

# Disk drives
Write-Host "`nFestplatten:" -ForegroundColor Yellow
Get-CimInstance Win32_LogicalDisk | Select-Object DeviceID, Size, FreeSpace | Format-Table

2. Listing User Accounts

# user_accounts.ps1 - Lists all user accounts

Write-Host "=== User-Accounts ===" -ForegroundColor Cyan

Get-LocalUser | Select-Object Name, Enabled, LastLogon | Format-Table

# Admin accounts
Write-Host "`nAdmin-Accounts:" -ForegroundColor Yellow
Get-LocalGroupMember -Group "Administrators" | Select-Object Name, PrincipalSource

3. Checking for Windows Updates

# check_updates.ps1 - Checks for Windows updates

Write-Host "=== Windows-Updates prüfen ===" -ForegroundColor Cyan

try {
    $UpdateSession = New-Object -ComObject Microsoft.Update.Session
    $UpdateSearcher = $UpdateSession.CreateUpdateSearcher()
    $Updates = $UpdateSearcher.Search("IsInstalled=0").Updates

    if ($Updates.Count -eq 0) {
        Write-Host "Keine ausstehenden Updates" -ForegroundColor Green
    } else {
        Write-Host "$($Updates.Count) ausstehende Updates:" -ForegroundColor Yellow
        foreach ($Update in $Updates) {
            Write-Host "  - $($Update.Title)"
        }
    }
} catch {
    Write-Host "Fehler beim Prüfen der Updates: $_" -ForegroundColor Red
}

4. Monitoring Network Connections

# network_monitor.ps1 - Monitors active network connections

Write-Host "=== Aktive Netzwerk-Verbindungen ===" -ForegroundColor Cyan

Get-NetTCPConnection | Where-Object {$_.State -eq "Established"} | 
    Select-Object LocalAddress, LocalPort, RemoteAddress, RemotePort, State | 
    Format-Table

# Open ports
Write-Host "`nOffene Ports:" -ForegroundColor Yellow
Get-NetTCPConnection | Where-Object {$_.State -eq "Listening"} | 
    Select-Object LocalAddress, LocalPort, OwningProcess | 
    Format-Table

5. Analyzing Event Logs

# event_log.ps1 - Analyzes Windows event logs

Write-Host "=== Event-Log Analyse (letzte 24h) ===" -ForegroundColor Cyan

$StartDate = (Get-Date).AddDays(-1)

# Errors in system log
Write-Host "`nSystem-Fehler:" -ForegroundColor Yellow
Get-EventLog -LogName System -EntryType Error -After $StartDate | 
    Select-Object TimeGenerated, Source, Message | 
    Format-Table -Wrap

# Warnings in application log
Write-Host "`nApplication-Warnungen:" -ForegroundColor Yellow
Get-EventLog -LogName Application -EntryType Warning -After $StartDate | 
    Select-Object TimeGenerated, Source, Message | 
    Format-Table -Wrap

Mac Bash/zsh Scripts (5)

1. Mac System Information

#!/bin/bash
# mac_info.sh - Display Mac system information

echo "=== Mac System Information ==="
echo ""

# Hardware
echo "Hardware:"
system_profiler SPHardwareDataType | grep -E "Model Name|Processor|Memory"

# macOS version
echo ""
echo "macOS Version:"
sw_vers

# Disk space
echo ""
echo "Disk Space:"
df -h

# Battery status (laptops)
echo ""
echo "Battery Status:"
system_profiler SPPowerDataType | grep -E "Charge|Capacity"

2. Update Homebrew Packages

#!/bin/bash
# brew_update.sh - Update Homebrew packages

echo "=== Homebrew Update ==="
echo ""

# Update Homebrew itself
echo "Updating Homebrew..."
brew update

# Update packages
echo ""
echo "Upgrading packages..."
brew upgrade

# Clean up
echo ""
echo "Cleaning up..."
brew cleanup

echo "Update complete!"

3. Automate Finder Organization

#!/bin/bash
# finder_organize.sh - Organize Downloads folder

DOWNLOADS="$HOME/Downloads"

echo "=== Organizing Downloads ==="

# Create folders
mkdir -p "$DOWNLOADS"/{Pictures,Documents,Music,Videos,Archives}

# Move files
for file in "$DOWNLOADS"/*; do
    if [ -f "$file" ]; then
        case "${file,,}" in
            *.jpg|*.jpeg|*.png|*.gif|*.heic)
                mv "$file" "$DOWNLOADS/Pictures/"
                echo "Pictures: $(basename "$file")"
                ;;
            *.pdf|*.doc|*.docx|*.txt|*.pages)
                mv "$file" "$DOWNLOADS/Documents/"
                echo "Documents: $(basename "$file")"
                ;;
            *.mp3|*.m4a|*.wav|*.aiff)
                mv "$file" "$DOWNLOADS/Music/"
                echo "Music: $(basename "$file")"
                ;;
            *.mp4|*.mov|*.avi|*.mkv)
                mv "$file" "$DOWNLOADS/Videos/"
                echo "Videos: $(basename "$file")"
                ;;
            *.zip|*.rar|*.7z|*.dmg)
                mv "$file" "$DOWNLOADS/Archives/"
                echo "Archives: $(basename "$file")"
                ;;
        esac
    fi
done

echo "Organization complete!"

4. Manage SSH Keys

#!/bin/bash
# ssh_keys.sh - Manage SSH keys

SSH_DIR="$HOME/.ssh"

echo "=== SSH Key Management ==="
echo ""

# Check SSH directory
if [ ! -d "$SSH_DIR" ]; then
    echo "Creating SSH directory..."
    mkdir -p "$SSH_DIR"
    chmod 700 "$SSH_DIR"
fi

# List existing keys
echo "Existing keys:"
ls -la "$SSH_DIR"/*.pub 2>/dev/null || echo "No public keys found"

# Create new key (optional)
read -p "Create new SSH key? (y/n): " create_key
if [ "$create_key" = "y" ]; then
    read -p "Key name (e.g. github): " key_name
    ssh-keygen -t ed25519 -f "$SSH_DIR/$key_name" -C "$key_name"
    echo "Key created: $SSH_DIR/$key_name"
fi

5. Dock Backup and Restore

#!/bin/bash
# dock_backup.sh - Backup and restore Dock configuration

DOCK_PLIST="$HOME/Library/Preferences/com.apple.dock.plist"
BACKUP_DIR="$HOME/DockBackups"

echo "=== Dock Backup ==="

# Create backup
backup_dock() {
    mkdir -p "$BACKUP_DIR"
    DATE=$(date +%Y%m%d_%H%M%S)
    cp "$DOCK_PLIST" "$BACKUP_DIR/dock_backup_$DATE.plist"
    echo "Dock backed up: dock_backup_$DATE.plist"
}

# Restore Dock
restore_dock() {
    echo "Available backups:"
    ls -lt "$BACKUP_DIR"/*.plist
    read -p "Enter backup file: " backup_file
    cp "$backup_file" "$DOCK_PLIST"
    killall Dock
    echo "Dock restored and restarted"
}

# Menu
echo "1: Backup Dock"
echo "2: Restore Dock"
read -p "Selection: " choice

case $choice in
    1) backup_dock ;;
    2) restore_dock ;;
    *) echo "Invalid selection" ;;
esac

Notes on Running Shell Scripts

Bash scripts (Linux/Mac):

1. Make executable: chmod +x script.sh
2. Run: ./script.sh or bash script.sh
3. Check path: which bash (typically /bin/bash)

PowerShell scripts (Windows):

1. Check execution policy: Get-ExecutionPolicy
2. Change if needed: Set-ExecutionPolicy RemoteSigned
3. Run: .\script.ps1 or powershell -File script.ps1

Mac-specific:

• Default shell is zsh (since macOS Catalina)
• Bash is still available: /bin/bash
• For AppleScript integration: use osascript

Best Practices for Automation Scripts

1. Coding Standards and Naming Conventions

Python:

# ✅ Good naming conventions
def backup_database(source_path: str, backup_dir: str) -> bool:
    """Create a database backup"""
    pass

# ❌ Poor naming conventions
def db(s, d):
    pass

Bash:

# ✅ Good naming conventions
backup_database() {
    local source_path="$1"
    local backup_dir="$2"
}

# ❌ Poor naming conventions
bd() {
    s=$1
    d=$2
}

PowerShell:

# ✅ Good naming conventions (PascalCase, Verb-Noun)
function Backup-Database {
    param(
        [string]$SourcePath,
        [string]$BackupDir
    )
}

# ❌ Poor naming conventions
function bd {
    param($s, $d)
}

2. Documentation and Comments

Docstrings and comments:

def process_files(directory: str, pattern: str) -> list:
    """
    Process all files in a directory matching a pattern.

    Args:
        directory: Path to the directory
        pattern: File pattern (e.g. '*.log')

    Returns:
        List of processed files

    Raises:
        FileNotFoundError: If directory does not exist
    """
    if not os.path.exists(directory):
        raise FileNotFoundError(f"Directory not found: {directory}")
    # ... rest of function

Bash comments:

#!/bin/bash
# backup.sh - Automatic backup with rotation
# Created: 2026-07-14
# Author: Your Name
# Version: 1.0

# Configuration
SOURCE_DIR="/home/user/documents"  # Source directory
BACKUP_DIR="/backup"               # Backup directory
MAX_BACKUPS=7                      # Maximum number of backups

3. Modularization and Reusability

Functions instead of monoliths:

# ✅ Modularized
def validate_path(path: str) -> bool:
    """Check if path exists"""
    return os.path.exists(path)

def copy_file(source: str, target: str) -> bool:
    """Copy file with error handling"""
    try:
        shutil.copy(source, target)
        return True
    except Exception as e:
        log_error(f"Copy error: {e}")
        return False

def backup_files(files: list, backup_dir: str) -> bool:
    """Main backup function"""
    if not validate_path(backup_dir):
        return False

    for file in files:
        if not copy_file(file, backup_dir):
            return False

    return True

4. Separate Configuration from Code

Use config files:

# config.yaml
backup:
  source_dir: "/home/user/documents"
  backup_dir: "/backup"
  max_backups: 7
  compression: "gzip"

# script.py
import yaml

def load_config(config_file: str) -> dict:
    """Load configuration from YAML file"""
    with open(config_file, 'r') as f:
        return yaml.safe_load(f)

config = load_config('config.yaml')
backup_config = config['backup']

5. Logging and Monitoring

Structured Logging:

import logging
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('automation.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def process_data(data: str):
    """Process data with logging"""
    logger.info(f"Processing started: {len(data)} characters")
    try:
        result = transform_data(data)
        logger.info(f"Processing completed successfully")
        return result
    except Exception as e:
        logger.error(f"Processing failed: {e}")
        raise

6. Error Handling and Recovery

Robust Error Handling:

import time
from functools import wraps

def retry(max_attempts: int = 3, delay: int = 1):
    """Decorator for retry mechanism"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    logger.warning(f"Attempt {attempt + 1} failed: {e}")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2)
def download_file(url: str, target: str):
    """Download file with retry logic"""
    # Download implementation
    pass

7. Testing and Validation

Basic Tests:

def test_validate_path():
    """Test for validate_path function"""
    # Test with existing path
    assert validate_path('/tmp') == True

    # Test with non-existent path
    assert validate_path('/nonexistent/path') == False

    print("✓ validate_path tests passed")

if __name__ == '__main__':
    test_validate_path()

8. Performance Optimization

Efficient Processing:

# ❌ Inefficient (for large files)
def process_large_file_slow(filepath: str):
    with open(filepath, 'r') as f:
        content = f.read()  # Loads everything into RAM
        return process_content(content)

# ✅ Efficient (streaming)
def process_large_file_fast(filepath: str):
    with open(filepath, 'r') as f:
        for line in f:  # Line-by-line processing
            yield process_line(line)

9. Script Security

Don’t Hardcode Secrets:

# ❌ Bad - passwords in code
password = "secret123"

# ✅ Good - environment variables
import os
password = os.getenv('DB_PASSWORD')

# ✅ Better - secrets manager
from keyring import get_password
password = get_password('system', 'username')

10. Versioning and Deployment

Semantic Versioning:

# Versioning in script
VERSION="1.2.3"  # MAJOR.MINOR.PATCH

# Changelog
# 1.2.3 - Bugfix for backup rotation
# 1.2.0 - New logging function
# 1.0.0 - Initial release

Security in Automation Scripts

1. Secrets Management

Environment Variables:

# .env file (don't commit to Git!)
DB_PASSWORD=secret123
API_KEY=sk-1234567890

# script.py
import os
from dotenv import load_dotenv

load_dotenv()  # Loads .env file

db_password = os.getenv('DB_PASSWORD')
api_key = os.getenv('API_KEY')

Secrets Manager:

# AWS Secrets Manager
import boto3

def get_secret(secret_name: str) -> str:
    """Retrieve secret from AWS Secrets Manager"""
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId=secret_name)
    return response['SecretString']

# Azure Key Vault
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential

def get_azure_secret(vault_url: str, secret_name: str) -> str:
    """Retrieve secret from Azure Key Vault"""
    credential = DefaultAzureCredential()
    client = SecretClient(vault_url=vault_url, credential=credential)
    return client.get_secret(secret_name).value

2. Input Validation

Validate User Input:

import re
from pathlib import Path

def validate_filename(filename: str) -> bool:
    """Validate filename for security"""
    # Allow only alphanumeric characters, underscore, hyphen
    pattern = r'^[a-zA-Z0-9_-]+\.[a-zA-Z0-9]{3,4}$'
    return bool(re.match(pattern, filename))

def validate_path(path: str) -> bool:
    """Validate path against traversal attacks"""
    resolved_path = Path(path).resolve()
    # Check if path falls outside allowed directory
    return not any(part in ['..', '~'] for part in resolved_path.parts)

# Usage
filename = input("Filename: ")
if not validate_filename(filename):
    raise ValueError("Invalid filename")

3. Rights and Permissions

Principle of Least Privilege:

#!/bin/bash
# Run script with minimal permissions

# Check if script is running as root
if [ "$EUID" -eq 0 ]; then
    echo "This script should not be run as root"
    exit 1
fi

# Create files with secure permissions
umask 077  # Only owner has access
touch sensitive_file.txt

Python Permissions:

import os
import stat

def set_secure_permissions(filepath: str):
    """Set secure file permissions (owner only)"""
    os.chmod(filepath, stat.S_IRUSR | stat.S_IWUSR)  # 0600

def set_executable_permissions(filepath: str):
    """Set executable permissions (owner rwx)"""
    os.chmod(filepath, stat.S_IRWXU)  # 0700

4. SQL Injection Prevention

Parameterized Queries:

# ❌ UNSAFE - SQL injection possible
query = f"SELECT * FROM users WHERE name = '{username}'"
cursor.execute(query)

# ✅ SAFE - parameterized query
query = "SELECT * FROM users WHERE name = %s"
cursor.execute(query, (username,))

5. Command Injection Prevention

Secure Shell Execution:

import subprocess

# ❌ UNSAFE - command injection possible
user_input = "user; rm -rf /"
os.system(f"echo {user_input}")

# ✅ SAFE - subprocess with list
subprocess.run(['echo', user_input], check=True)

# ✅ Even safer - shlex.quote
import shlex
safe_input = shlex.quote(user_input)
subprocess.run(f"echo {safe_input}", shell=True, check=True)

6. File Operation Safety

Secure File Operations:

import os
import tempfile

def safe_file_write(filepath: str, content: str):
    """Secure write with atomic operations"""
    # Temporary file in same directory
    dir_path = os.path.dirname(filepath)
    with tempfile.NamedTemporaryFile(
        mode='w',
        dir=dir_path,
        delete=False
    ) as tmp_file:
        tmp_file.write(content)
        tmp_path = tmp_file.name
    
    # Atomic rename
    os.rename(tmp_path, filepath)

def safe_file_delete(filepath: str):
    """Secure deletion with confirmation"""
    if not os.path.exists(filepath):
        return False
    
    # Request confirmation
    response = input(f"Delete {filepath}? (y/n): ")
    if response.lower() == 'y':
        os.remove(filepath)
        return True
    return False

7. Logging Sensitive Data

Prevent Secrets in Logs:

import logging
from typing import Any

class SecureFormatter(logging.Formatter):
    """Formatter that masks sensitive data"""
    SENSITIVE_KEYS = ['password', 'token', 'secret', 'key']
    
    def format(self, record: logging.LogRecord) -> str:
        msg = super().format(record)
        for key in self.SENSITIVE_KEYS:
            msg = msg.replace(f'{key}=', f'{key}=***')
        return msg

# Usage
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(SecureFormatter())
logger.addHandler(handler)

logger.info("Login with password=secret123")
# Output: Login with password=***

8. HTTPS and TLS

Securing network connections:

import requests
import urllib3

# Enable SSL warnings (not in production!)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# ✅ SECURE - HTTPS with certificate verification
response = requests.get('https://api.example.com', verify=True)

# ❌ INSECURE - SSL verification disabled
response = requests.get('https://api.example.com', verify=False)

# ✅ SECURE - Custom CA certificate
response = requests.get(
    'https://api.example.com',
    verify='/path/to/ca-bundle.crt'
)

9. Dependency Security

Vulnerability scanning:

# Scan Python dependencies
pip install safety
safety check

# Scan npm dependencies
npm audit

# Update all dependencies
pip install --upgrade pip
pip list --outdated

10. Security Audits

Regular security checks:

import subprocess
import json

def run_security_audit():
    """Runs basic security checks"""
    checks = {
        'python_version': check_python_version,
        'dependencies': check_dependencies,
        'file_permissions': check_file_permissions,
        'hardcoded_secrets': check_hardcoded_secrets
    }
    
    results = {}
    for check_name, check_func in checks.items():
        results[check_name] = check_func()
    
    return results

def check_hardcoded_secrets():
    """Checks for hardcoded secrets in code"""
    import re
    secret_patterns = [
        r'password\s*=\s*["\'][^"\']+["\']',
        r'api_key\s*=\s*["\'][^"\']+["\']',
        r'secret\s*=\s*["\'][^"\']+["\']'
    ]
    
    findings = []
    for root, dirs, files in os.walk('.'):
        for file in files:
            if file.endswith('.py'):
                filepath = os.path.join(root, file)
                with open(filepath, 'r') as f:
                    content = f.read()
                    for pattern in secret_patterns:
                        if re.search(pattern, content):
                            findings.append(filepath)
    
    return findings

Testing Automation Scripts

1. Unit Testing for Python Scripts

Basic unit tests:

# test_backup.py
import unittest
import tempfile
import os
from backup_script import backup_files, validate_path

class TestBackupFunctions(unittest.TestCase):
    def setUp(self):
        """Setup for tests"""
        self.temp_dir = tempfile.mkdtemp()
        self.test_file = os.path.join(self.temp_dir, 'test.txt')
        with open(self.test_file, 'w') as f:
            f.write('Test content')
    
    def tearDown(self):
        """Cleanup after tests"""
        import shutil
        shutil.rmtree(self.temp_dir)
    
    def test_validate_path_existing(self):
        """Test for existing path"""
        self.assertTrue(validate_path(self.temp_dir))
    
    def test_validate_path_non_existing(self):
        """Test for non-existent path"""
        self.assertFalse(validate_path('/not/existent'))
    
    def test_backup_files_success(self):
        """Test successful backup"""
        backup_dir = os.path.join(self.temp_dir, 'backup')
        os.makedirs(backup_dir)
        
        result = backup_files([self.test_file], backup_dir)
        self.assertTrue(result)
        self.assertTrue(os.path.exists(os.path.join(backup_dir, 'test.txt')))

if __name__ == '__main__':
    unittest.main()

2. Mocking External Dependencies

Mocking filesystem operations:

from unittest.mock import patch, MagicMock
import unittest

class TestBackupWithMocking(unittest.TestCase):
    @patch('backup_script.shutil.copy')
    def test_backup_with_mock(self, mock_copy):
        """Test with mocked shutil.copy function"""
        mock_copy.return_value = True
        
        result = backup_files(['file1.txt'], '/backup')
        
        self.assertTrue(result)
        mock_copy.assert_called_once()
    
    @patch('backup_script.os.path.exists')
    def test_validate_path_mock(self, mock_exists):
        """Test with mocked os.path.exists"""
        mock_exists.return_value = True
        
        result = validate_path('/any/path')
        
        self.assertTrue(result)
        mock_exists.assert_called_once_with('/any/path')

3. Integration Testing

End-to-end tests:

# test_integration.py
import unittest
import tempfile
import os
import subprocess

class TestIntegration(unittest.TestCase):
    def test_script_execution(self):
        """Test complete script execution"""
        script_path = 'backup.sh'
        test_dir = tempfile.mkdtemp()
        
        # Execute script
        result = subprocess.run(
            ['bash', script_path, test_dir],
            capture_output=True,
            text=True
        )
        
        # Verify results
        self.assertEqual(result.returncode, 0)
        self.assertIn('Backup erstellt', result.stdout)
        
        # Cleanup
        import shutil
        shutil.rmtree(test_dir)

4. Testing Bash Scripts

Bash unit testing framework:

#!/bin/bash
# test_backup.sh - Tests for backup.sh

# Test functions
test_validate_path() {
    source backup.sh
    
    # Test with existing path
    if validate_path "/tmp"; then
        echo "✓ validate_path: existing path"
    else
        echo "✗ validate_path: existing path failed"
        exit 1
    fi
    
    # Test with non-existent path
    if ! validate_path "/not/existent"; then
        echo "✓ validate_path: non-existent path"
    else
        echo "✗ validate_path: non-existent path failed"
        exit 1
    fi
}

test_backup_function() {
    source backup.sh
    
    local test_dir=$(mktemp -d)
    local test_file="$test_dir/test.txt"
    echo "Test content" > "$test_file"
    
    # Test backup function
    backup_files "$test_file" "$test_dir/backup"
    
    if [ -f "$test_dir/backup/test.txt" ]; then
        echo "✓ backup_files: file copied successfully"
    else
        echo "✗ backup_files: file not copied"
        exit 1
    fi
    
    # Cleanup
    rm -rf "$test_dir"
}

# Run all tests
echo "=== Bash Tests ==="
test_validate_path
test_backup_function
echo "All tests passed!"

5. PowerShell Testing

Pester framework for PowerShell:

# test_backup.ps1 - Pester tests for backup.ps1

Describe "Backup Functions" {
    BeforeAll {
        . .\backup.ps1
        $testDir = Join-Path $env:TEMP "backup_test"
        New-Item -Path $testDir -ItemType Directory -Force | Out-Null
    }
    
    AfterAll {
        Remove-Item -Path $testDir -Recurse -Force -ErrorAction SilentlyContinue
    }
    
    Context "Validate-Path" {
        It "should return existing path" {
            $result = Validate-Path $testDir
            $result | Should -Be $true
        }
        
        It "should reject non-existent path" {
            $result = Validate-Path "C:\not\existent"
            $result | Should -Be $false
        }
    }
    
    Context "Backup-Files" {
        It "should copy files successfully" {
            $testFile = Join-Path $testDir "test.txt"
            "Test content" | Out-File -FilePath $testFile
            
            $backupDir = Join-Path $testDir "backup"
            $result = Backup-Files $testFile $backupDir
            
            $result | Should -Be $true
            (Test-Path (Join-Path $backupDir "test.txt")) | Should -Be $true
        }
    }
}

6. Test-Driven Development (TDD)

TDD workflow:

# 1. Write a test (it will fail)
def test_calculate_backup_size():
    """Test for backup size calculation"""
    result = calculate_backup_size('/tmp')
    assert result > 0  # Will fail—function doesn't exist yet

# 2. Implement the function (to make the test pass)
def calculate_backup_size(directory: str) -> int:
    """Calculate the total size of a directory"""
    total_size = 0
    for dirpath, dirnames, filenames in os.walk(directory):
        for filename in filenames:
            filepath = os.path.join(dirpath, filename)
            total_size += os.path.getsize(filepath)
    return total_size

# 3. Run the test (should pass now)
# 4. Refactor (while keeping tests passing)

7. Continuous Testing

Run tests automatically when files change:

# Install watchdog
pip install watchdog watchmedo

# Automatically execute tests on file changes
watchmedo shell-command \
    --patterns="*.py" \
    --recursive \
    --command='python -m pytest test_*.py -v'

8. Test Coverage

Analyze code coverage:

# Install pytest-cov
pip install pytest-cov

# Run tests with coverage report
pytest --cov=backup_script --cov-report=html

# Open the coverage report
open htmlcov/index.html

9. Property-Based Testing

Using Hypothesis in Python:

from hypothesis import given, strategies as st
import unittest

class TestBackupProperties(unittest.TestCase):
    @given(st.lists(st.text()))
    def test_backup_preserves_content(self, content_list):
        """Property test: verify backup preserves content"""
        # Test with various inputs
        for content in content_list:
            with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
                f.write(content)
                temp_file = f.name
            
            # Create backup
            backup_file = temp_file + '.backup'
            shutil.copy(temp_file, backup_file)
            
            # Verify content
            with open(backup_file, 'r') as f:
                backup_content = f.read()
            
            self.assertEqual(content, backup_content)
            
            # Cleanup
            os.unlink(temp_file)
            os.unlink(backup_file)

10. Load Testing for Scripts

Performance testing:

import time
import statistics

def test_backup_performance():
    """Test backup performance"""
    test_times = []
    
    for i in range(10):
        start_time = time.time()
        backup_files(['large_file.txt'], '/backup')
        end_time = time.time()
        
        test_times.append(end_time - start_time)
    
    avg_time = statistics.mean(test_times)
    max_time = max(test_times)
    
    print(f"Average time: {avg_time:.2f}s")
    print(f"Maximum time: {max_time:.2f}s")
    
    # Performance assertions
    assert avg_time < 5.0, "Backup is too slow"
    assert max_time < 10.0, "Backup exceeds maximum time"

Version Control for Automation Scripts

1. Git Basics for Scripts

Initialize and make your first commit:

# Initialize a Git repository
git init

# Create .gitignore
cat > .gitignore << EOF
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/

# Secrets
.env
*.key
*.pem
secrets/

# Logs
*.log

# IDE
.vscode/
.idea/
*.swp
*.swo

# OS
.DS_Store
Thumbs.db
EOF

# Add scripts
git add backup.sh backup.py config.yaml
git commit -m "Initial commit: Backup-Skripte hinzugefügt"

2. Branching Strategies

Git Flow for scripts:

# Main branch (production)
git checkout -b main

# Develop branch (development)
git checkout -b develop

# Feature branch for a new feature
git checkout -b feature/add-logging

# Commit changes
git add backup.py
git commit -m "Add logging functionality"

# Merge back to develop
git checkout develop
git merge feature/add-logging

# Delete feature branch
git branch -d feature/add-logging

3. Conventional Commit Messages

Conventional Commits:

# Format: <type>(<scope>): <description>

# Examples:
git commit -m "feat(backup): Add compression support"
git commit -m "fix(logging): Fix timestamp format"
git commit -m "docs(readme): Update installation instructions"
git commit -m "style(python): Apply PEP 8 formatting"
git commit -m "refactor(cleanup): Simplify error handling"
git commit -m "test(backup): Add unit tests for backup function"
git commit -m "chore(deps): Update dependencies"

4. Code Reviews and Pull Requests

Pull request workflow:

# Create feature branch
git checkout -b feature/backup-rotation

# Make changes and commit
git add backup.sh
git commit -m "feat(backup): Add automatic backup rotation"

# Push to remote
git push origin feature/backup-rotation

# Create a pull request (GitHub/GitLab/Bitbucket)
# Team members review the code
# Merge after approval

5. Git Hooks for Quality Assurance

Pre-commit hook:

#!/bin/bash
# .git/hooks/pre-commit

# Format Python code
autopep8 --in-place --aggressive *.py

# Lint Python code
pylint *.py

# Run tests
python -m pytest test_*.py

# Exit with success
exit 0

Pre-push hook:

#!/bin/bash
# .git/hooks/pre-push

# Run all tests
python -m pytest

# Security check
safety check

# Exit with success
exit 0

6. Versioning with Tags

Semantic Versioning:

# Major version (breaking changes)
git tag -a v2.0.0 -m "Major release: Breaking changes"
git push origin v2.0.0

# Minor version (new features, backward compatible)
git tag -a v1.2.0 -m "Minor release: New features"
git push origin v1.2.0

# Patch version (bugfixes, backward compatible)
git tag -a v1.1.1 -m "Patch release: Bug fixes"
git push origin v1.1.1

7. Maintaining a Changelog

Automatic changelog generation:

# Install conventional-changelog
npm install -g conventional-changelog-cli

# Generate changelog
conventional-changelog -p angular -i CHANGELOG.md -s

# CHANGELOG.md structure:
# ## [1.2.0] - 2026-07-14
# ### Added
# - Backup rotation support
# - Logging functionality
#
# ### Fixed
# - Path validation error
#
# ## [1.1.0] - 2026-06-01
# ### Added
# - Initial backup script release

8. Git Configuration

Git configuration for scripts:

# Configure user
git config user.name "Your Name"
git config user.email "your@email.com"

# Line endings (important for cross-platform work)
git config core.autocrlf input  # Linux/Mac
git config core.autocrlf true   # Windows

# File attributes
git config core.filemode false  # Ignore permissions

# Configure diff tool
git config diff.tool vimdiff
git config merge.tool vimdiff

9. Git Workflows for Teams

Forking workflow:

# 1. Fork the repository (GitHub/GitLab)
# 2. Clone your fork
git clone https://github.com/your-username/repo.git

# 3. Add upstream remote
git remote add upstream https://github.com/original/repo.git

# 4. Create feature branch
git checkout -b feature/new-feature

# 5. Commit changes
git add script.py
git commit -m "feat: add new feature"

# 6. Push to your fork
git push origin feature/new-feature

# 7. Create a pull request
# 8. Merge after review

# 9. Fetch upstream changes
git fetch upstream
git checkout main
git merge upstream/main

10. Git for Backup Strategies

Using Git as a backup system:

#!/bin/bash
# git_backup.sh - Automated Git backup

REPO_DIR="/path/to/repo"
BACKUP_REMOTE="git@github.com:user/backup-repo.git"

cd "$REPO_DIR"

# Stage changes
git add .

# Commit with timestamp
git commit -m "Auto-backup: $(date '+%Y-%m-%d %H:%M:%S')"

# Push to remote
git push "$BACKUP_REMOTE" main

echo "Backup successfully pushed to Git"

CI/CD Integration for Automation Scripts

1. GitHub Actions for Python Scripts

Workflow file:

# .github/workflows/automation.yml
name: Automation Scripts CI/CD

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-cov pylint safety
    
    - name: Lint with pylint
      run: pylint *.py
    
    - name: Security check
      run: safety check
    
    - name: Run tests
      run: pytest --cov=. --cov-report=xml
    
    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml

  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to production
      run: |
        echo "Deploying automation scripts to production"
        # Deployment logic here

2. GitLab CI for Bash Scripts

GitLab CI configuration:

# .gitlab-ci.yml
stages:
  - test
  - deploy

variables:
  SCRIPT_DIR: "/opt/automation"

test_scripts:
  stage: test
  image: alpine:latest
  
  before_script:
    - apk add --no-cache bash shellcheck
  
  script:
    - shellcheck *.sh
    - bash -n backup.sh
    - bash test_backup.sh
  
  artifacts:
    reports:
      junit: test-results.xml

deploy_production:
  stage: deploy
  image: alpine:latest
  only:
    - main
  
  script:
    - apk add --no-cache rsync openssh
    - mkdir -p ~/.ssh
    - echo "$SSH_PRIVATE_KEY" > ~/.ssh/id_rsa
    - chmod 600 ~/.ssh/id_rsa
    - rsync -avz --delete *.sh user@server:$SCRIPT_DIR/
    - ssh user@server "chmod +x $SCRIPT_DIR/*.sh"

3. Jenkins Pipeline

Jenkinsfile:

// Jenkinsfile
pipeline {
    agent any
    
    environment {
        PYTHON_VERSION = '3.11'
        DEPLOY_SERVER = 'user@server'
    }
    
    stages {
        stage('Checkout') {
            steps {
                checkout scm
            }
        }
        
        stage('Lint') {
            steps {
                sh 'pylint *.py || true'
                sh 'shellcheck *.sh || true'
            }
        }
        
        stage('Test') {
            steps {
                sh 'python -m pytest test_*.py --cov=. --cov-report=html'
            }
        }
        
        stage('Security Scan') {
            steps {
                sh 'safety check'
            }
        }
        
        stage('Deploy') {
            when {
                branch 'main'
            }
            steps {
                sh '''
                    rsync -avz --delete *.sh ${DEPLOY_SERVER}:/opt/automation/
                    ssh ${DEPLOY_SERVER} "chmod +x /opt/automation/*.sh"
                '''
            }
        }
    }
    
    post {
        always {
            archiveArtifacts artifacts: 'htmlcov/**', allowEmptyArchive: true
            publishHTML target: [
                reportDir: 'htmlcov',
                reportFiles: 'index.html',
                reportName: 'Coverage Report'
            ]
        }
    }
}

4. Azure DevOps Pipeline

azure-pipelines.yml:

# azure-pipelines.yml
trigger:
  branches:
    include:
    - main
    - develop

pool:
  vmImage: 'ubuntu-latest'

variables:
  pythonVersion: '3.11'

stages:
- stage: Test
  jobs:
  - job: TestScripts
    steps:
    - task: UsePythonVersion@0
      inputs:
        versionSpec: '$(pythonVersion)'
    
    - script: |
        python -m pip install --upgrade pip
        pip install pytest pytest-cov pylint safety
      displayName: 'Install dependencies'
    
    - script: |
        pylint *.py
      displayName: 'Lint Python'
    
    - script: |
        safety check
      displayName: 'Security check'
    
    - script: |
        pytest --cov=. --cov-report=xml
      displayName: 'Run tests'
    
    - task: PublishCodeCoverageResults@1
      inputs:
        codeCoverageTool: 'Cobertura'
        summaryFileLocation: '$(System.DefaultWorkingDirectory)/coverage.xml'

- stage: Deploy
  dependsOn: Test
  condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
  jobs:
  - deployment: DeployScripts
    environment: 'Production'
    strategy:
      runOnce:
        deploy:
          steps:
          - script: |
              rsync -avz --delete *.sh user@server:/opt/automation/
            displayName: 'Deploy scripts'

5. Docker-based CI/CD

Dockerfile for tests:

# Dockerfile.test
FROM python:3.11-slim

# System dependencies
RUN apt-get update && apt-get install -y \
    shellcheck \
    bash \
    && rm -rf /var/lib/apt/lists/*

# Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir pytest pytest-cov pylint safety

# Copy scripts
COPY *.py *.sh /app/
WORKDIR /app

# Test command
CMD ["pytest", "--cov=.", "--cov-report=html"]

Docker Compose for CI:

# docker-compose.test.yml
version: '3.8'

services:
  test:
    build:
      context: .
      dockerfile: Dockerfile.test
    volumes:
      - .:/app
      - ./htmlcov:/app/htmlcov
    command: pytest --cov=. --cov-report=html

6. Environment Management

Multi-environment deployment:

# .github/workflows/deploy.yml
name: Multi-Environment Deploy

on:
  push:
    branches: [ main ]

jobs:
  deploy_staging:
    runs-on: ubuntu-latest
    environment: staging
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to Staging
      run: |
        rsync -avz --delete *.sh $STAGING_SERVER:/opt/automation/
      env:
        STAGING_SERVER: ${{ secrets.STAGING_SERVER }}

  deploy_production:
    runs-on: ubuntu-latest
    environment: production
    needs: deploy_staging
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to Production
      run: |
        rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
      env:
        PROD_SERVER: ${{ secrets.PROD_SERVER }}

7. Rollback Strategies

Automatic rollback:

# .github/workflows/rollback.yml
name: Rollback Deployment

on:
  workflow_dispatch:
    inputs:
      version:
        description: 'Version to rollback to'
        required: true

jobs:
  rollback:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
      with:
        ref: ${{ github.event.inputs.version }}
    
    - name: Deploy previous version
      run: |
        rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
      env:
        PROD_SERVER: ${{ secrets.PROD_SERVER }}
    
    - name: Notify team
      run: |
        echo "Rollback to version ${{ github.event.inputs.version }} completed"

8. Monitoring in CI/CD

Pipeline Health Checks:

# .github/workflows/monitoring.yml
name: Pipeline Monitoring

on:
  schedule:
    - cron: '0 * * * *'  # Every hour

jobs:
  health_check:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Check script health
      run: |
        # Verify scripts are running
        ssh $PROD_SERVER "ps aux | grep backup.sh"
        
        # Verify logs are being written
        ssh $PROD_SERVER "tail -n 10 /var/log/automation.log"
      env:
        PROD_SERVER: ${{ secrets.PROD_SERVER }}
    
    - name: Alert on failure
      if: failure()
      run: |
        # Send alert (Slack, email, etc.)
        curl -X POST $SLACK_WEBHOOK -d '{"text":"Automation health check failed"}'
      env:
        SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}

9. Automated Testing in Pipeline

Test Matrix for Multiple Python Versions:

# .github/workflows/test-matrix.yml
name: Test Matrix

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ['3.9', '3.10', '3.11']
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest
    
    - name: Run tests
      run: pytest test_*.py

10. Secrets Management in CI/CD

Using GitHub Secrets:

# .github/workflows/secure-deploy.yml
name: Secure Deployment

on:
  push:
    branches: [ main ]

jobs:
  deploy:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Setup SSH key
      run: |
        mkdir -p ~/.ssh
        echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
        chmod 600 ~/.ssh/id_rsa
        ssh-keyscan -H ${{ secrets.DEPLOY_SERVER }} >> ~/.ssh/known_hosts
    
    - name: Deploy with secrets
      run: |
        # Load secrets from environment
        export DB_PASSWORD="${{ secrets.DB_PASSWORD }}"
        export API_KEY="${{ secrets.API_KEY }}"
        
        # Run deployment script
        rsync -avz --delete *.sh ${{ secrets.DEPLOY_SERVER }}:/opt/automation/
        ssh ${{ secrets.DEPLOY_SERVER }} "cd /opt/automation && ./deploy.sh"
      env:
        DEPLOY_SERVER: ${{ secrets.DEPLOY_SERVER }}

Containerizing Automation Scripts

1. Dockerfile for Python Scripts

Basic Dockerfile:

# Dockerfile
FROM python:3.11-slim

# Working directory
WORKDIR /app

# Copy dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy scripts
COPY backup.py .
COPY config.yaml .

# Set script as entry point
ENTRYPOINT ["python", "backup.py"]
CMD ["--config", "config.yaml"]

2. Dockerfile for Bash Scripts

Bash in Container:

# Dockerfile.bash
FROM alpine:latest

# Install Bash and tools
RUN apk add --no-cache bash rsync coreutils

# Copy scripts
COPY backup.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/backup.sh

# Copy configuration
COPY config.yaml /etc/backup/

# Working directory
WORKDIR /data

ENTRYPOINT ["/usr/local/bin/backup.sh"]

3. Multi-Stage Dockerfile

Optimized Multi-Stage Build:

# Dockerfile.multi-stage
# Build stage
FROM python:3.11-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# Runtime stage
FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY backup.py .
COPY config.yaml .

ENV PATH=/root/.local/bin:$PATH

ENTRYPOINT ["python", "backup.py"]

4. Docker Compose for Complete Environment

docker-compose.yml:

version: '3.8'

services:
  backup:
    build:
      context: .
      dockerfile: Dockerfile
    volumes:
      - ./data:/data
      - ./config:/config
      - ./logs:/logs
    environment:
      - BACKUP_DIR=/data/backups
      - SOURCE_DIR=/data/source
      - LOG_LEVEL=INFO
    restart: unless-stopped

  monitoring:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana

volumes:
  grafana-data:

5. Kubernetes Deployment

Deployment YAML:

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: automation-scripts
spec:
  replicas: 2
  selector:
    matchLabels:
      app: automation
  template:
    metadata:
      labels:
        app: automation
    spec:
      containers:
      - name: backup
        image: your-registry/backup:latest
        env:
        - name: BACKUP_DIR
          value: "/data/backups"
        - name: SOURCE_DIR
          valueFrom:
            configMapKeyRef:
              name: backup-config
              key: source_dir
        volumeMounts:
        - name: data
          mountPath: /data
        - name: config
          mountPath: /config
      volumes:
      - name: data
        persistentVolumeClaim:
          claimName: backup-pvc
      - name: config
        configMap:
          name: backup-config

6. Docker Health Checks

Health Check in Dockerfile:

# Dockerfile with health check
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backup.py health_check.py .

# Health check script
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python health_check.py || exit 1

ENTRYPOINT ["python", "backup.py"]

Health Check Script:

# health_check.py
import os
import sys

def check_health():
    """Verify the backup script is healthy"""
    # Check if configuration exists
    if not os.path.exists('config.yaml'):
        print("Config file missing")
        return False
    
    # Check if backup directory is writable
    backup_dir = os.getenv('BACKUP_DIR', '/data/backups')
    if not os.access(backup_dir, os.W_OK):
        print("Backup directory not writable")
        return False
    
    print("Health check passed")
    return True

if __name__ == '__main__':
    sys.exit(0 if check_health() else 1)

7. Container Optimization

Best Practices for Smaller Images:

# Optimized Dockerfile
FROM python:3.11-alpine AS builder

# Only necessary dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir --user --no-deps -r requirements.txt

FROM python:3.11-alpine

# Non-root user
RUN addgroup -g 1000 backup && \
    adduser -D -u 1000 -G backup backup

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY --chown=backup:backup backup.py .

USER backup
ENV PATH=/root/.local/bin:$PATH

ENTRYPOINT ["python", "backup.py"]

8. Container Security

Security Scanning and Hardening:

# Security-hardened Dockerfile
FROM python:3.11-slim

# Security updates
RUN apt-get update && \
    apt-get upgrade -y && \
    apt-get install -y --no-install-recommends \
    ca-certificates && \
    rm -rf /var/lib/apt/lists/*

# Non-root user
RUN groupadd -r backup && useradd -r -g backup backup

WORKDIR /app
COPY --chown=backup:backup requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

COPY --chown=backup:backup backup.py .

USER backup
ENTRYPOINT ["python", "backup.py"]

Security Scanning with Trivy:

# Scan image
trivy image your-registry/backup:latest

# Integrate into CI/CD
trivy image --exit-code 1 --severity HIGH,CRITICAL your-registry/backup:latest

9. Container Logging

Structured logging:

# backup.py with container logging
import logging
import json
import sys

class JSONFormatter(logging.Formatter):
    """JSON formatter for container logs"""
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
            'logger': record.name
        }
        if hasattr(record, 'extra_data'):
            log_data.update(record.extra_data)
        return json.dumps(log_data)

# Configure logging
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

10. Container Orchestration

Docker Swarm service:

# docker-stack.yml
version: '3.8'

services:
  backup:
    image: your-registry/backup:latest
    deploy:
      replicas: 3
      update_config:
        parallelism: 1
        delay: 10s
      restart_policy:
        condition: on-failure
        delay: 5s
        max_attempts: 3
    environment:
      - BACKUP_DIR=/data/backups
    volumes:
      - backup-data:/data
    networks:
      - automation-network

networks:
  automation-network:
    driver: overlay

volumes:
  backup-data:

Monitoring & Alerting for Automation Scripts

1. Prometheus Metrics Export

Python metrics with Prometheus:

# metrics_exporter.py
from prometheus_client import start_http_server, Counter, Histogram, Gauge
import time

# Define metrics
backup_counter = Counter('backup_total', 'Total number of backups', ['status'])
backup_duration = Histogram('backup_duration_seconds', 'Backup duration')
backup_size = Gauge('backup_size_bytes', 'Backup size in bytes')

def backup_with_metrics(source: str, target: str):
    """Backup with Prometheus metrics"""
    start_time = time.time()
    
    try:
        # Perform backup
        size = perform_backup(source, target)
        
        # Update metrics
        backup_counter.labels(status='success').inc()
        backup_duration.observe(time.time() - start_time)
        backup_size.set(size)
        
    except Exception as e:
        backup_counter.labels(status='failed').inc()
        raise

# Start metrics server
if __name__ == '__main__':
    start_http_server(8000)
    print("Metrics server started on port 8000")

2. Grafana Dashboard

Grafana dashboard configuration:

{
  "dashboard": {
    "title": "Automation Scripts Monitoring",
    "panels": [
      {
        "title": "Backup Success Rate",
        "targets": [
          {
            "expr": "rate(backup_total{status='success'}[5m])"
          }
        ]
      },
      {
        "title": "Backup Duration",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m]))"
          }
        ]
      },
      {
        "title": "Backup Size",
        "targets": [
          {
            "expr": "backup_size_bytes"
          }
        ]
      }
    ]
  }
}

3. Log Analysis with ELK Stack

Filebeat configuration:

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/automation/*.log
  fields:
    app: automation
  fields_under_root: true

output.elasticsearch:
  hosts: ["localhost:9200"]
  index: "automation-logs-%{+yyyy.MM.dd}"

setup.kibana:
  host: "localhost:5601"

4. Alerting with Alertmanager

Alertmanager configuration:

# alertmanager.yml
global:
  resolve_timeout: 5m

route:
  group_by: ['alertname']
  group_wait: 10s
  group_interval: 10s
  repeat_interval: 1h
  receiver: 'web.hook'

receivers:
- name: 'web.hook'
  webhook_configs:
  - url: 'http://localhost:5001/alerts'

- name: 'slack'
  slack_configs:
  - api_url: 'YOUR_SLACK_WEBHOOK_URL'
    channel: '#automation-alerts'

Prometheus alert rules:

# alerts.yml
groups:
- name: automation_alerts
  rules:
  - alert: BackupFailed
    expr: rate(backup_total{status='failed'}[5m]) > 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "Backup failed"
      description: "Backup has been failing for the last 5 minutes"

  - alert: BackupTooSlow
    expr: histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m])) > 300
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Backup too slow"
      description: "Backup duration exceeds 5 minutes"

5. Health Check Endpoints

HTTP health check:

# health_check.py
from flask import Flask, jsonify
import os
import psutil

app = Flask(__name__)

@app.route('/health')
def health_check():
    """Health check endpoint"""
    health_status = {
        'status': 'healthy',
        'checks': {}
    }
    
    # Check disk space
    disk_usage = psutil.disk_usage('/')
    health_status['checks']['disk'] = {
        'status': 'ok' if disk_usage.percent < 90 else 'warning',
        'usage_percent': disk_usage.percent
    }
    
    # Check memory
    memory = psutil.virtual_memory()
    health_status['checks']['memory'] = {
        'status': 'ok' if memory.percent < 90 else 'warning',
        'usage_percent': memory.percent
    }
    
    # Check config file
    config_exists = os.path.exists('config.yaml')
    health_status['checks']['config'] = {
        'status': 'ok' if config_exists else 'error',
        'exists': config_exists
    }
    
    # Overall status
    if any(check['status'] == 'error' for check in health_status['checks'].values()):
        health_status['status'] = 'error'
    elif any(check['status'] == 'warning' for check in health_status['checks'].values()):
        health_status['status'] = 'warning'
    
    return jsonify(health_status)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

6. Slack Integration

Send Slack alerts:

# slack_alerts.py
import requests
import json

def send_slack_alert(webhook_url: str, message: str, level: str = 'info'):
    """Send alert to Slack"""
    colors = {
        'info': '#36a64f',
        'warning': '#ff9900',
        'error': '#ff0000'
    }
    
    payload = {
        'attachments': [{
            'color': colors.get(level, '#36a64f'),
            'title': f'Automation Alert: {level.upper()}',
            'text': message,
            'footer': 'Automation Scripts',
            'ts': int(time.time())
        }]
    }
    
    response = requests.post(webhook_url, json=payload)
    return response.status_code == 200

# Usage
send_slack_alert(
    webhook_url='YOUR_SLACK_WEBHOOK',
    message='Backup failed for database',
    level='error'
)

7. Email Alerting

Email alerts with Python:

# email_alerts.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_email_alert(
    subject: str,
    message: str,
    to_email: str,
    from_email: str,
    smtp_server: str,
    smtp_port: int = 587
):
    """Send email alert"""
    msg = MIMEMultipart()
    msg['From'] = from_email
    msg['To'] = to_email
    msg['Subject'] = subject
    
    msg.attach(MIMEText(message, 'plain'))
    
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        server.login(from_email, 'YOUR_PASSWORD')
        server.send_message(msg)

# Usage
send_email_alert(
    subject='Automation Alert: Backup Failed',
    message='The backup script failed with error: ...',
    to_email='admin@example.com',
    from_email='automation@example.com',
    smtp_server='smtp.example.com'
)

8. System Monitoring with psutil

Monitor system resources:

# system_monitor.py
import psutil
import time

def monitor_system(interval: int = 60):
    """Monitors system resources"""
    while True:
        # CPU
        cpu_percent = psutil.cpu_percent(interval=1)
        
        # Memory
        memory = psutil.virtual_memory()
        
        # Disk
        disk = psutil.disk_usage('/')
        
        # Network
        network = psutil.net_io_counters()
        
        # Output
        print(f"CPU: {cpu_percent}%")
        print(f"Memory: {memory.percent}%")
        print(f"Disk: {disk.percent}%")
        print(f"Network Sent: {network.bytes_sent} bytes")
        print(f"Network Recv: {network.bytes_recv} bytes")
        print("-" * 40)
        
        # Alert on high values
        if cpu_percent > 90:
            send_alert(f"High CPU usage: {cpu_percent}%")
        if memory.percent > 90:
            send_alert(f"High memory usage: {memory.percent}%")
        if disk.percent > 90:
            send_alert(f"High disk usage: {disk.percent}%")
        
        time.sleep(interval)

9. Log Rotation and Management

logrotate configuration:

# /etc/logrotate.d/automation
/var/log/automation/*.log {
    daily
    rotate 14
    compress
    delaycompress
    missingok
    notifempty
    create 0640 automation automation
    sharedscripts
    postrotate
        systemctl reload automation-service > /dev/null 2>&1 || true
    endscript
}

10. Uptime Monitoring

Check uptime with Python:

# uptime_monitor.py
import requests
import time
from datetime import datetime

def check_uptime(url: str, interval: int = 60):
    """Monitors service uptime"""
    downtime_start = None
    total_downtime = 0
    
    while True:
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                if downtime_start:
                    downtime_duration = time.time() - downtime_start
                    total_downtime += downtime_duration
                    send_alert(f"Service is back up after {downtime_duration:.2f}s downtime")
                    downtime_start = None
                print(f"{datetime.now()}: Service is UP")
            else:
                if not downtime_start:
                    downtime_start = time.time()
                    send_alert(f"Service returned status {response.status_code}")
                print(f"{datetime.now()}: Service returned {response.status_code}")
        except requests.RequestException as e:
            if not downtime_start:
                downtime_start = time.time()
                send_alert(f"Service is down: {str(e)}")
            print(f"{datetime.now()}: Service is DOWN - {str(e)}")
        
        time.sleep(interval)

Advanced Error Handling for Automation Scripts

1. Structured Error Handling

Custom exception classes:

# exceptions.py
class AutomationError(Exception):
    """Base exception for automation scripts"""
    pass

class BackupError(AutomationError):
    """Backup-specific errors"""
    pass

class ConfigError(AutomationError):
    """Configuration errors"""
    pass

class NetworkError(AutomationError):
    """Network errors"""
    pass

class ValidationError(AutomationError):
    """Validation errors"""
    pass

2. Retry Mechanisms with Exponential Backoff

Intelligent retry logic:

# retry_handler.py
import time
import random
from functools import wraps
from typing import Callable, Type

def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2,
    jitter: bool = True,
    exceptions: tuple = (Exception,)
):
    """Decorator with exponential backoff and jitter"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = initial_delay
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries - 1:
                        raise
                    
                    # Add jitter to avoid thundering herd
                    if jitter:
                        delay = delay * (1 + random.uniform(-0.1, 0.1))
                    
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
                    time.sleep(min(delay, max_delay))
                    delay *= exponential_base
            
        return wrapper
    return decorator

# Usage
@retry_with_backoff(max_retries=5, initial_delay=2.0, exceptions=(ConnectionError, TimeoutError))
def download_file(url: str, target: str):
    """Downloads file with retry logic"""
    # Download logic
    pass

3. Circuit Breaker Pattern

Circuit breaker for external services:

# circuit_breaker.py
from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"  # Normal operation
    OPEN = "open"      # Circuit is open, requests fail immediately
    HALF_OPEN = "half_open"  # Testing if service recovered

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        expected_exception: Exception = Exception
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    def call(self, func: Callable, *args, **kwargs):
        """Executes function with circuit breaker"""
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise
    
    def _should_attempt_reset(self) -> bool:
        """Checks if circuit should attempt reset"""
        return (
            self.last_failure_time and
            time.time() - self.last_failure_time >= self.recovery_timeout
        )
    
    def _on_success(self):
        """On successful execution"""
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        """On failed execution"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage
circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

def call_external_api():
    """Call external API with circuit breaker protection"""
    return circuit_breaker.call(requests.get, "https://api.example.com")

4. Dead Letter Queue for Failed Tasks

Dead Letter Queue Implementation:

# dead_letter_queue.py
import json
import os
from datetime import datetime
from typing import Any, Dict

class DeadLetterQueue:
    def __init__(self, queue_file: str = "dead_letter_queue.json"):
        self.queue_file = queue_file
        self._load_queue()
    
    def _load_queue(self):
        """Loads queue from file"""
        if os.path.exists(self.queue_file):
            with open(self.queue_file, 'r') as f:
                self.queue = json.load(f)
        else:
            self.queue = []
    
    def _save_queue(self):
        """Persists queue to file"""
        with open(self.queue_file, 'w') as f:
            json.dump(self.queue, f, indent=2)
    
    def add(self, task: Dict[str, Any], error: Exception):
        """Adds a failed task to the queue"""
        dead_letter = {
            'task': task,
            'error': str(error),
            'error_type': type(error).__name__,
            'timestamp': datetime.now().isoformat(),
            'retry_count': 0
        }
        self.queue.append(dead_letter)
        self._save_queue()
    
    def get_next(self) -> Dict[str, Any]:
        """Retrieves the next task from the queue"""
        if not self.queue:
            return None
        
        return self.queue.pop(0)
    
    def retry(self, max_retries: int = 3):
        """Retries failed tasks"""
        remaining_tasks = []
        
        for dead_letter in self.queue:
            if dead_letter['retry_count'] < max_retries:
                try:
                    # Re-execute the task
                    execute_task(dead_letter['task'])
                    print(f"Task {dead_letter['task']['id']} retried successfully")
                except Exception as e:
                    dead_letter['retry_count'] += 1
                    dead_letter['last_error'] = str(e)
                    remaining_tasks.append(dead_letter)
            else:
                print(f"Task {dead_letter['task']['id']} reached max retries")
        
        self.queue = remaining_tasks
        self._save_queue()

# Usage
dlq = DeadLetterQueue()

try:
    execute_task(task)
except Exception as e:
    dlq.add(task, e)

5. Graceful Degradation

Fallback Strategies:

# graceful_degradation.py
from typing import Optional, Callable

class FallbackChain:
    def __init__(self):
        self.fallbacks = []
    
    def add_fallback(self, func: Callable, priority: int = 0):
        """Registers a fallback function"""
        self.fallbacks.append((priority, func))
        self.fallbacks.sort(key=lambda x: x[0], reverse=True)
    
    def execute(self, *args, **kwargs) -> Optional[Any]:
        """Executes functions with fallback strategy"""
        for priority, func in self.fallbacks:
            try:
                result = func(*args, **kwargs)
                if result is not None:
                    return result
            except Exception as e:
                print(f"Fallback {func.__name__} failed: {e}")
                continue
        
        return None

# Usage
fallback_chain = FallbackChain()

# Primary method
def primary_backup(source: str, target: str):
    """Primary backup method (Cloud)"""
    return upload_to_cloud(source, target)

# First fallback
def secondary_backup(source: str, target: str):
    """Secondary backup method (FTP)"""
    return upload_to_ftp(source, target)

# Second fallback
def tertiary_backup(source: str, target: str):
    """Tertiary backup method (Local)"""
    return copy_locally(source, target)

fallback_chain.add_fallback(primary_backup, priority=3)
fallback_chain.add_fallback(secondary_backup, priority=2)
fallback_chain.add_fallback(tertiary_backup, priority=1)

result = fallback_chain.execute(source, target)

6. Comprehensive Error Logging

Detailed Error Logging:

# error_logger.py
import logging
import traceback
import sys
from datetime import datetime
from typing import Optional

class ErrorLogger:
    def __init__(self, log_file: str = "automation_errors.log"):
        self.logger = logging.getLogger('error_logger')
        self.logger.setLevel(logging.ERROR)
        
        # File handler
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.ERROR)
        
        # Formatter
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        file_handler.setFormatter(formatter)
        
        self.logger.addHandler(file_handler)
    
    def log_error(
        self,
        error: Exception,
        context: Optional[dict] = None,
        critical: bool = False
    ):
        """Logs an error with contextual information"""
        error_info = {
            'error_type': type(error).__name__,
            'error_message': str(error),
            'traceback': traceback.format_exc(),
            'context': context or {},
            'timestamp': datetime.now().isoformat()
        }
        
        log_message = f"Error: {error_info}"
        if critical:
            self.logger.critical(log_message)
        else:
            self.logger.error(log_message)
        
        return error_info
    
    def log_exception(self, func: Callable):
        """Decorator for automatic exception logging"""
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                self.log_error(e, context={'function': func.__name__})
                raise
        return wrapper

# Usage
error_logger = ErrorLogger()

@error_logger.log_exception
def risky_operation():
    """Risky operation with automatic logging"""
    pass

7. Error Recovery Strategies

Automatic Recovery:

# recovery.py
from typing import Callable, Any

class RecoveryManager:
    def __init__(self):
        self.recovery_strategies = {}
    
    def register_recovery(self, error_type: type, recovery_func: Callable):
        """Registers a recovery strategy for an error type"""
        self.recovery_strategies[error_type] = recovery_func
    
    def execute_with_recovery(self, func: Callable, *args, **kwargs) -> Any:
        """Executes a function with automatic recovery"""
        try:
            return func(*args, **kwargs)
        except Exception as e:
            error_type = type(e)
            
            if error_type in self.recovery_strategies:
                print(f"Attempting recovery for {error_type.__name__}")
                recovery_func = self.recovery_strategies[error_type]
                return recovery_func(e, *args, **kwargs)
            else:
                raise

# Usage
recovery_manager = RecoveryManager()

def recover_from_connection_error(error: Exception, *args, **kwargs):
    """Recovery from connection errors"""
    print("Reconnecting...")
    time.sleep(5)
    return func(*args, **kwargs)

def recover_from_disk_full(error: Exception, *args, **kwargs):
    """Recovery from disk full errors"""
    print("Cleaning up old files...")
    cleanup_old_files()
    return func(*args, **kwargs)

recovery_manager.register_recovery(ConnectionError, recover_from_connection_error)
recovery_manager.register_recovery(OSError, recover_from_disk_full)

8. Timeout Handling

Handling timeouts for long-running operations:

# timeout_handler.py
import signal
from contextlib import contextmanager
import time

class TimeoutError(Exception):
    """Custom Timeout Exception"""
    pass

@contextmanager
def timeout_handler(seconds: int):
    """Context Manager for Timeout"""
    def signal_handler(signum, frame):
        raise TimeoutError(f"Operation timed out after {seconds} seconds")
    
    # Register Signal Handler
    old_handler = signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    
    try:
        yield
    finally:
        # Reset Signal Handler
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

# Usage
try:
    with timeout_handler(30):
        long_running_operation()
except TimeoutError:
    print("Operation timed out, using fallback")
    fallback_operation()

9. Error Aggregation

Collecting multiple errors:

# error_aggregator.py
from typing import List, Dict, Any

class ErrorAggregator:
    def __init__(self):
        self.errors: List[Dict[str, Any]] = []
    
    def add_error(self, error: Exception, context: Dict[str, Any] = None):
        """Adds an error to the collection"""
        self.errors.append({
            'error': str(error),
            'type': type(error).__name__,
            'context': context or {},
            'timestamp': time.time()
        })
    
    def has_errors(self) -> bool:
        """Checks if errors are present"""
        return len(self.errors) > 0
    
    def get_summary(self) -> Dict[str, Any]:
        """Returns a summary of errors"""
        error_types = {}
        for error in self.errors:
            error_type = error['type']
            error_types[error_type] = error_types.get(error_type, 0) + 1
        
        return {
            'total_errors': len(self.errors),
            'error_types': error_types,
            'errors': self.errors
        }
    
    def clear(self):
        """Clears all errors"""
        self.errors = []

# Usage
aggregator = ErrorAggregator()

for item in items:
    try:
        process_item(item)
    except Exception as e:
        aggregator.add_error(e, context={'item': item})

if aggregator.has_errors():
    print(f"Processing completed with {len(aggregator.errors)} errors")
    print(aggregator.get_summary())

10. Error Notification System

Alerting on critical errors:

# error_notifier.py
from typing import List, Callable

class ErrorNotifier:
    def __init__(self):
        self.notifiers: List[Callable] = []
    
    def add_notifier(self, notifier: Callable):
        """Adds a notification function"""
        self.notifiers.append(notifier)
    
    def notify(self, error: Exception, context: dict = None):
        """Sends notification to all notifiers"""
        error_info = {
            'error': str(error),
            'type': type(error).__name__,
            'context': context or {},
            'timestamp': time.time()
        }
        
        for notifier in self.notifiers:
            try:
                notifier(error_info)
            except Exception as e:
                print(f"Notifier failed: {e}")

# Example notifiers
def slack_notifier(error_info: dict):
    """Slack notification"""
    send_slack_message(f"Error: {error_info['error']}")

def email_notifier(error_info: dict):
    """Email notification"""
    send_email(f"Error: {error_info['error']}", error_info)

# Usage
notifier = ErrorNotifier()
notifier.add_notifier(slack_notifier)
notifier.add_notifier(email_notifier)

try:
    critical_operation()
except Exception as e:
    notifier.notify(e, context={'operation': 'critical_operation'})