pyats

import os import logging from pyats import aetest from genie.testbed import load from unicon.core.errors import ConnectionError # Set up basic logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Create a dummy testbed file (testbed.yaml) for demonstration # In a real scenario, this would describe your network devices. # For a live environment, set environment variables for credentials: # export PYATS_TESTBED_USERNAME='your_username' # export PYATS_TESTBED_PASSWORD='your_password' # Example content for testbed.yaml: # devices: # CSR1: # os: iosxe # connections: # cli: # protocol: ssh # ip: 10.1.1.1 # Replace with your device IP # credentials: # default: # username: '%ENV{PYATS_TESTBED_USERNAME}' # password: '%ENV{PYATS_TESTBED_PASSWORD}' class MyFirstTest(aetest.Testcase): @aetest.setup def setup(self, testbed): # Load the testbed from a YAML file try: self_testbed = load(os.path.join(os.path.dirname(__file__), 'testbed.yaml')) self.parent.parameters.update(testbed=self_testbed) logger.info(f"Testbed loaded successfully: {list(self_testbed.devices.keys())}") except Exception as e: self.failed(f"Failed to load testbed: {e}") @aetest.test def connect_and_execute(self, testbed): if not hasattr(self.parent.parameters, 'testbed'): self.skipped('Testbed not loaded in setup.') return for device_name, device in self.parent.parameters['testbed'].devices.items(): logger.info(f"Attempting to connect to device: {device_name}") try: device.connect(init_exec_params={'timeout': 60}) logger.info(f"Successfully connected to {device_name}") # Example: Execute a 'show version' command output = device.execute('show version') logger.info(f"Output from {device_name}:\n{output[:200]}...") # Print first 200 chars # Example: Parse 'show ip interface brief' parsed_output = device.parse('show ip interface brief') logger.info(f"Parsed output from {device_name} (first entry): {list(parsed_output.values())[0] if parsed_output else 'No interfaces found'}") self.passed(f"Successfully connected, executed, and parsed on {device_name}") except ConnectionError as e: self.failed(f"Connection failed for {device_name}: {e}") except Exception as e: self.failed(f"An error occurred with {device_name}: {e}") finally: if device.is_connected(): device.disconnect() logger.info(f"Disconnected from {device_name}") @aetest.cleanup def cleanup(self): logger.info("Testcase cleanup completed.") if __name__ == '__main__': import sys # Create a dummy testbed.yaml for local execution if it doesn't exist testbed_path = os.path.join(os.path.dirname(__file__), 'testbed.yaml') if not os.path.exists(testbed_path): with open(testbed_path, 'w') as f: f.write(""" --- devices: CSR1: os: iosxe connections: cli: protocol: ssh ip: 10.1.1.1 # REPLACE WITH YOUR DEVICE IP OR A MOCK IP FOR TESTING credentials: default: username: '%ENV{PYATS_TESTBED_USERNAME}' password: '%ENV{PYATS_TESTBED_PASSWORD}' """) logger.warning(f"Created a dummy '{testbed_path}'. Please replace 10.1.1.1 with a real device IP and set environment variables PYATS_TESTBED_USERNAME and PYATS_TESTBED_PASSWORD.") # Set dummy env vars if not present for local execution to avoid crashes os.environ.setdefault('PYATS_TESTBED_USERNAME', 'admin') os.environ.setdefault('PYATS_TESTBED_PASSWORD', 'password') # Run the AETest aetest.main(argv=sys.argv)