Welche Methode ist die beste? Multiple IP´s in einem Unittest ausführen

Sockets, TCP/IP, (XML-)RPC und ähnliche Themen gehören in dieses Forum
Antworten
frentmeister
User
Beiträge: 13
Registriert: Freitag 3. August 2018, 12:46

In meinem Testaufbau habe ich aktuell nur ein Gerät, und andere kann ich zwar hinzufügen, aber da ich mit dem Skript hier mehrere komplexe Aufbauten mit diversen Switches testen muss wäre natürlich eine eigene passende Methode sehr hilfreich.

Nun ich habe einige Ideen, die ich hier für richtig halte:

- Über eine CSV einbinden
- Eine eigene Methode entwickeln

Wie kann ich aber einen entsprechenden Aufbau realisieren?

Setup

Genutzte Pakete:

Paramiko
PySerial
Unittest

Code: Alles auswählen

##########################################################################
#  Setup
##########################################################################

    def setUp(self):

        self.s = testcore.control.ssh.SSH(host='xxx.xxx.xxx.xxx',
                                      username='admin', password='admin', type_of_dut='ecos')
        logger.info('self.s = testcore.control.ssh.SSH')
        self.s.query_interactive=True


        if self.s.login():
            logger.info('login')
            q = self.s.query('account')
            logger.info('account')
            self.assertIsNotNone(q, 'missing answer')
            self.assertEqual('\r\n', q, 'unexpected result')
            logger.info('missing answer')
            logger.info('unexpected result')

            # switch to prompt account

            q=self.s.query('enforce-Password-Rules yes')
            logger.info('enforce-Password-Rules yes')
            q=self.s.query('exit')
            logger.info('exit')

    def tearDown(self):
        self.s.close()


        #########################################################################
        # Test 1 create external NTP Server http://www.pool.ntp.org/zone/de IP v4
        #########################################################################

    def test_create_ntp_external1(self):
        logger_true.info('test_create_ntp_external')
        if self.s.loggedin:
            logger.info('self.s.loggedin')
            q = self.s.query('time')
            logger.info('time')
            self.assertIsNotNone(q, 'missing answer')
            self.assertEqual('\r\n', q, 'unexpected result')
            logger.info('missing answer')
            logger.info('unexpected result')

            # switch to prompt account

            q = self.s.query('ntp 1 94.16.116.137')
            logger.info('ntp 1 94.16.116.137')
            q = self.s.query('exit')
            self.assertIsNotNone(q, 'missing answer')
            self.assertEqual('\r\n', q, 'unexpected result')
            logger.info('missing answer')
            logger.info('unexpected result')

            q = self.s.query('logout')
            logger.info('logout')

            import time
            print('Wait')
            time.sleep(2)
            print('True')
            logger_true.info('True')
            self.s.close()
Benutzeravatar
sls
User
Beiträge: 480
Registriert: Mittwoch 13. Mai 2015, 23:52
Wohnort: Country country = new Zealand();

Willst du mit einem Unittest die Funktionalität deines Programmes testen, oder prüfen ob deine Netzwerkgeräte laufen? Für letzteres ist das IMO nicht gemacht.

Zur eigentlichen Frage: wenn deine Daten (und so sollten sie idealerweise vorliegen) in einem JSON, YAML, oder auch CSV bereitstehen, kannst du diese über eine entsprechende Funktion einlesen und an die Klasse übergeben.

Sofern es sinnvoll ist für jedes Netzwerkgerät / Test ein eigenes Objekt zu haben, kannst du eine Fabrikfunktion dafür schreiben.

Noch ein paar Anmerkungen zum Code:

Methoden, Funktionen und Variablen werden in Python klein_mit_unterstrich geschrieben, kein CamelCase. `s`, `q` sind wenig aussagekräftige Attributnamen. Variablennamen sollten eindeutig gewählt werden und beschreiben, wozu diese gut sind. Dein Programm enthält keine Fehlerbehandlung, was passiert denn, wenn eine SSH-Verbindung fehlschlägt (timeout usw.)? Spätestens würde eine Exception fliegen. Diese sollte man abfangen und sauber rausloggen. Dein logging ist auch nicht aussagekräftig, "login" -> ja, aber *wo* wird sich denn eingeloggt? Warum loggst du eine Instantiierung raus?

Hier würde ich eher `Connecting to <hostname> <ip>...` o.ä. loggen. Hierfür eignet sich Try-Except.

Warum `test_create_ntp_external1` ? Gibt's noch eine Methode mit dem Namen 2 am Ende?

NTP-Public-IP würde ich konfigurierbar machen und nicht hardcoden. Warum time.sleep()? Dein ganzes Programm steht an dieser Stelle, spätestens jetzt sich dein Test in die Länge wenn du mal ein paar hundert Geräte testen möchtest.
When we say computer, we mean the electronic computer.
frentmeister
User
Beiträge: 13
Registriert: Freitag 3. August 2018, 12:46

Willst du mit einem Unittest die Funktionalität deines Programmes testen, oder prüfen ob deine Netzwerkgeräte laufen? Für letzteres ist das IMO nicht gemacht.
Vollkommen richtig, aktuell ist das nur ein allerster Test.Dieser (der gesamte Test) testet gerade aber schon 100 einzelne Test nutzt dabei sowohl Pyserial, Paramiko Funktionen etc. aus Gründen der Übersichtlichkeit, weil mehrere tausend Zeilen Code, nur ein Abriss dessen, was mir da vorliegt.
Zur eigentlichen Frage: wenn deine Daten (und so sollten sie idealerweise vorliegen) in einem JSON, YAML, oder auch CSV bereitstehen, kannst du diese über eine entsprechende Funktion einlesen und an die Klasse übergeben.
Wäre das aber die beste Lösung, ein einlesen oder besser die Funktion als Python Methode schreiben? Vielleicht als vorlaufendes Setup?
Methoden, Funktionen und Variablen werden in Python klein_mit_unterstrich geschrieben, kein CamelCase. `s`, `q` sind wenig aussagekräftige Attributnamen. Variablennamen sollten eindeutig gewählt werden und beschreiben, wozu diese gut sind. Dein Programm enthält keine Fehlerbehandlung, was passiert denn, wenn eine SSH-Verbindung fehlschlägt (timeout usw.)? Spätestens würde eine Exception fliegen. angen und sauber rausloggen. Dein logging ist auch nicht aussagekräftig, "login" -> ja, aber *wo* wird sich denn eingeloggt? Warum loggst du eine Instantiierung raus?

self.s.close()
self.s = testcore.control.ssh.SSH(host='172.23.56.250', username='testuser_P2', password='testuser_P2')
self.s.query_interactive = True
self.s.login()
self.assertEqual(self.s.loggedin, False)

Dies ist doch über den Assert geregelt. Sehe aber gerade das ich das in dem Beispiel nicht drin habe. Ich habe da entsprechend in anderen Beispielen für einen nicht Login (Timeout)
logger_true.info('test_try_login2')
self.s.close()
self.s = testcore.control.ssh.SSH(host='172.23.56.xxx', username='testuser_P2', password='testuser_P2')
self.s.query_interactive = True
self.s.login()
self.assertEqual(self.s.loggedin, False)
Was passiert hier wenn der Login nicht funktioniert. Test schlägt fehl, und der nächste Test beginnt.

Warum `test_create_ntp_external1` ? Gibt's noch eine Methode mit dem Namen 2 am Ende?
100 weitere Funktionen ;)
NTP-Public-IP würde ich konfigurierbar machen und nicht hardcoden. Warum time.sleep()? Dein ganzes Programm steht an dieser Stelle, spätestens jetzt sich dein Test in die Länge wenn du mal ein paar hundert Geräte testen möchtest.
Nur wir hatten Probleme auf den Geräten, Befehl kam zu schnell und das Propmt hat sie mehr oder weniger verschluckt. Du hast aber recht, aktuell nicht mehr nötig.

Vielen Dank für die vielen Infos.
frentmeister
User
Beiträge: 13
Registriert: Freitag 3. August 2018, 12:46

Da ich ja prinzipiell eben Paramiko für meine Aufbaueten nutze, würde folgendes Beispiel ausreichend sein?

Code: Alles auswählen

import sys, os, string, threading
import paramiko

cmd = "grep -h 'king' /opt/data/horror_20100810*"

outlock = threading.Lock()

def workon(host):

    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host, username='xy', password='xy')
    stdin, stdout, stderr = ssh.exec_command(cmd)
    stdin.write('xy\n')
    stdin.flush()

    with outlock:
        print stdout.readlines()

def main():
    hosts = ['10.10.3.10', '10.10.4.12', '10.10.2.15', ] # etc
    threads = []
    for h in hosts:
        t = threading.Thread(target=workon, args=(h,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

main()
Wie gesagt bei 100 Geräten die ich damit abtesten will, ist dass schon sehr schwierig meiner Ansicht nach. Dies Beispiel habe ich erstmal genutzt um zu schauen ob es generell funktioniert.
Benutzeravatar
sls
User
Beiträge: 480
Registriert: Mittwoch 13. Mai 2015, 23:52
Wohnort: Country country = new Zealand();

Mal ein ganz triviales Beispiel:

Code: Alles auswählen

import yaml

def send_ssh_cmd(host, username, password, commands):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host, username=username, password=password)
    for command in commands:
    	stdin, stdout, stderr = ssh.exec_command(command)
        ....
    


def send_telnet_cmd():
    pass

....

def main(devices):
    for host in devices['switch']:
        if devices['switch'][host]['connection'] == 'ssh':
            send_ssh_cmd(devices['switch'][host]['hostname'],
                        devices['switch'][host]['ip'],
                        devices['switch'][host]['commands'],
                            )
        elif ...


if __name__ == '__main__':
    with open('devices.yml', 'r') as input_file:
        devices = yaml.load(input_file)
    main(devices)

Das File mit den hinterlegten Netzwerkgeräten:

Code: Alles auswählen

switch:
  01:
    hostname: 'core01.inside.example'
    ip: '10.0.0.1'
    connection: 'ssh'
    vendor: 'cisco xyz'
    commands:
      - 'do something'
  02:
    hostname: 'edge01.inside.example'
    ip: '172.16.0.2'
    connection: 'ssh'
    vendor: 'juniper xyz'
    commands:
      - 'set ntp server 172.20.X.Y\r\n'
      - 'set ntp server 172.22.X.Y\r\n'

Struktur und Aufbau dienen hier einfach mal um zu zeigen, was ich meine:

Du hast ein Config-File in yaml-Format in welchem alle Netzwerkgeräte aufgelistet sind. Die main()-Methode liest dieses File ein, und ermittelt die jeweiligen Commandos. Wenn man das so machen möchte, muss man noch die Redundanzen entfernen (damit nicht für bspw. 60 Geräte einer Sorte alle Kommandos im Config-File kopiert werden müssen usw., dafür könnte man den NTP-Server an einer Stelle im Yaml-File konfigurieren und Platzhalter für die Kommandos definieren bzw. eine Liste mit flags angeben, z.B. commands: ['ntp', 'logging' usw. usf.])

Wichtig ist, alle Parameter wie IP-Adressen usw. nicht hardcoden.

Natürlich fehlt bei obiger Bearbeitung noch die Nebenläufigkeit, ich würde das evtl. mit asyncio umsetzen und auf gar keinen Fall ein einziges threading.Lock() verwenden. Genau das blockiert dir das ganze Programm wenn ein Thread das Lock() akquiriert und dann z.B. in ein quälend langes SSH-Timeout läuft.

Der Vorteil ist, dass man klare Unterscheidungen zwischen den Gerätetypen treffen kann und nur noch die Art der Verbindung definiert wird. Der Verbindungsaufbau von telnet, ssh oder seriell unterscheidet sich, weshalb hier eine jeweils eigene Methode sinnvoll ist, die gerätespezifischen Kommandos hingegen lassen sich wunderbar als Parameter an diese Funktionen übergeben.

Am Ende kannst du das ganze dann mit Unittests versehen um (und zwar nur um!) die eigentliche Funktionalität dieses Konstruktes zu prüfen.
When we say computer, we mean the electronic computer.
frentmeister
User
Beiträge: 13
Registriert: Freitag 3. August 2018, 12:46

Das hat geholfen, ich baue gerade ein entsprechendes File und importiere das dann entsprechend. Danke!
Sirius3
User
Beiträge: 17711
Registriert: Sonntag 21. Oktober 2012, 17:20

Unittest hat eine sehr unpythonische API, besser ist z.B. pytest. Dann ist es auch nicht ganz so schlimm, dass Unittests eigentlich nicht für das Testen von Servern gedacht ist.
frentmeister
User
Beiträge: 13
Registriert: Freitag 3. August 2018, 12:46

Sirius3 hat geschrieben: Samstag 29. September 2018, 19:26 Unittest hat eine sehr unpythonische API, besser ist z.B. pytest. Dann ist es auch nicht ganz so schlimm, dass Unittests eigentlich nicht für das Testen von Servern gedacht ist.
Im ganzen Test der diverse Switche Testet, ist auch noch Tftp, Serial, und weitere Möglichkeiten implementiert. Pytest wäre besser, aber aktuell aufgrund von ca. 40K Test (richtig gelesen!) ein riesen Moloch das alles umzusetzen. Steht auf meiner Liste! :)
Antworten