HES-Selenium/update_dcu.py

import time
import openpyxl
import pexpect
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException
from pathlib import Path

def create_ssh_tunnel(dcu_ip, ssh_user="callysta_icon", ssh_host="10.232.4.113", ssh_port=21112, ssh_password="Callysta_icon2024!"):
    ssh_command = f"ssh -L 8888:{dcu_ip}:80 {ssh_user}@{ssh_host} -p {ssh_port}"
    child = pexpect.spawn(ssh_command)
    try:
        # Expect the password prompt and send the password
        child.expect("callysta_icon@10.232.4.113's password: ")
        child.sendline(ssh_password)
        child.expect(pexpect.EOF)
        print("SSH tunnel established.")
    except pexpect.exceptions.EOF:
        print("Error: SSH connection failed.")
    except pexpect.exceptions.TIMEOUT:
        print("Error: SSH connection timed out.")
    return child  # Return the child process to keep the tunnel open

def verify_dcu_update(driver, expected_ip, retries=10):
    """Verify if the DCU update is successful by checking IP and Connect State."""
    for attempt in range(retries):
        try:
            print(f"Attempt {attempt + 1} to verify DCU update...")

            # Click on the "Running State" tab to ensure correct content is displayed
            driver.find_element(By.ID, "running-state").click()
            time.sleep(2)  # Small delay to let the page render

            # Wait until the main_host_info container is present, indicating page has loaded
            WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.ID, "main_host_info"))
            )

            # Locate info items within main_host_info container
            info_items = driver.find_elements(By.XPATH, "//div[@id='main_host_info']/div[@class='info_item']")
            print(f"Found {len(info_items)} info items.")  # Debug output for verification

            updated_ip = None
            update_status = None

            # Extract the information based on description (desc)
            for item in info_items:
                desc = item.find_element(By.CLASS_NAME, "desc").text.strip()
                data = item.find_element(By.CLASS_NAME, "data").text.strip()
                print(f"Description: {desc}, Data: {data}")  # Debugging output for each item

                if desc == "IP And Port":
                    updated_ip = data
                elif desc == "Connect State":
                    update_status = data

            print(f"Found Updated IP: {updated_ip}, Update Status: {update_status}")

            # Check if the updated IP matches and the state is "Success"
            if updated_ip == expected_ip and update_status == "Success":
                print("DCU update verified successfully!")
                return updated_ip, update_status

            # Retry only if the expected data is not yet as desired
            time.sleep(5 * (attempt + 1))

        except TimeoutException:
            print("Timeout while waiting for the Running State or information container.")
        except NoSuchElementException as e:
            print(f"Error locating elements: {e}")

    # If the loop exits without success
    raise Exception(f"Failed to verify DCU update after {retries} retries.")


def update_dcu_ip(excel_file, log_file):
    # Load the Excel file
    wb = openpyxl.load_workbook(excel_file)
    sheet = wb.active

    # Set up Chrome WebDriver
    chrome_options = Options()
    chrome_options.binary_location = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"  # Path to Chrome binary
    chrome_options.add_argument("--disable-gpu")
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-dev-shm-usage")

    driver_path = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/chromedriver/chromedriver"  # Path to ChromeDriver binary

    try:
        driver = webdriver.Chrome(service=Service(driver_path), options=chrome_options)
    except Exception as e:
        print(f"Error initializing WebDriver: {e}")
        return

    # Open log file for writing
    with open(log_file, "w") as log:
        log.write("DCU IP,Updated IP,Update Status\n")

        # Loop through each DCU IP in the Excel file
        for row in range(2, sheet.max_row + 1):  # Start from row 2
            dcu_ip = sheet.cell(row=row, column=5).value.strip('="')  # Column E contains DCU IPs
            print(f"Processing DCU IP: {dcu_ip}")

            # Create the SSH tunnel
            ssh_password = "Callysta_icon2024!"  # Replace with actual password
            ssh_process = create_ssh_tunnel(dcu_ip, ssh_password=ssh_password)

            try:
                # Access the DCU web application
                driver.get("http://localhost:8888")
                WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))

                # Login section
                driver.find_element(By.ID, "username").send_keys("admin")
                driver.find_element(By.ID, "password").send_keys("jh1296")
                driver.find_element(By.ID, "submit_btn").click()
                WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "parameter-setting")))

                # Navigate to Parameter Setting
                driver.find_element(By.ID, "parameter-setting").click()

                # Handle iframe if necessary
                iframes = driver.find_elements(By.TAG_NAME, "iframe")
                if iframes:
                    driver.switch_to.frame(iframes[0])  # Adjust index based on iframe structure

                # Update the MDC IP
                new_ip = "10.232.107.250:9032"  # Replace this with the new IP and port you want to set

                # Wait for the input field to become visible and interactable
                WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "main_ip")))
                ip_field = driver.find_element(By.ID, "main_ip")

                # Retry clearing the field
                for _ in range(3):
                    ip_field.clear()
                    if not ip_field.get_attribute("value").strip():
                        break
                    time.sleep(1)
                else:
                    raise Exception("Failed to clear the IP field!")

                # Enter the new IP and submit
                ip_field.send_keys(new_ip)
                driver.find_element(By.ID, "main_host_submit").click()

                time.sleep(3)  # Wait for changes to take effect

                # Verify the changes with retry logic
                updated_ip, update_status = verify_dcu_update(driver, expected_ip=new_ip)

                # Log the result
                log.write(f"{dcu_ip},{updated_ip},{update_status}\n")

            except TimeoutException as e:
                print(f"Timeout error for DCU IP {dcu_ip}: {e}")
                log.write(f"{dcu_ip},ERROR,Timeout\n")
            except Exception as e:
                print(f"Error processing DCU IP {dcu_ip}: {e}")
                log.write(f"{dcu_ip},ERROR,{str(e)}\n")

            finally:
                # Close the SSH tunnel
                if ssh_process.isalive():
                    ssh_process.terminate()

    # Close the driver
    driver.quit()

# Main function
if __name__ == "__main__":
    excel_file = Path("/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/DCU-LIST.xlsx")  # Replace with the path to your Excel file
    log_file = Path("/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/update_log.csv")  # Replace with the desired path for your log file
    update_dcu_ip(excel_file, log_file)