Добавил тестер метролога

selenium/metrolog.py

@@ -1,41 +1,99 @@
+import time
+import threading
+from transport import *
 from selenium import webdriver
 from selenium.webdriver.common.keys import Keys
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support.ui import WebDriverWait
 from selenium.webdriver.support import expected_conditions as ES
-import time
 
 
-class MetrologWebTest():
+class MetrologTester():
     USER = 'user'
     PASWORD = 'uchetmo'
+    PGW_PORT_1 = 1001
+    PGW_PORT_2 = 1002
+    PGW_PORT_3 = 1003
+    PGW_TEST_DATA = b'abcdefghijklmnopqrstuvwxyz'
+
+
+    def __init__(self, ip):
 
-    def __init__(self):
+        self.pgw_1 = PortgwTcp(ip, self.PGW_PORT_1)
+        self.pgw_2 = PortgwTcp(ip, self.PGW_PORT_2)
+        self.pgw_3 = PortgwTcp(ip, self.PGW_PORT_3)
+
+        self.pgw_1_thread = threading.Thread(target=self.portgw_sender, args=(self.pgw_1, ))
+        self.pgw_2_thread = threading.Thread(target=self.portgw_sender, args=(self.pgw_2, ))
+        self.pgw_3_thread = threading.Thread(target=self.portgw_sender, args=(self.pgw_3, ))
+
+        self.ip = ip
         self.driver = webdriver.Firefox()
+        self.web_thread = threading.Thread(target=self.test_page_click)
+
+    def connect(self):
+        self.driver.get('http://' + self.ip)
 
-    def login(self, ip: str) -> bool:  
-        self.driver.get('http://' + ip)
-        time.sleep(1)
+    def login(self) -> bool:  
         self.driver.find_element(By.ID, 'login').send_keys(self.USER)
         self.driver.find_element(By.ID, 'pass').send_keys(self.PASWORD)
         self.driver.find_element(By.CLASS_NAME, 'btn.btn-primary').click()
-        time.sleep(1)
+        time.sleep(2)
         return True if self.driver.title == "Состояние модема" else False
 
     def test_page_click(self):
+        self.connect()
         while True:
+            if self.login() == False:
+                return
+            for i in range(1):
+                time.sleep(1)
+                nav = self.driver.find_element(By.XPATH, "//ul[@id='nav']/li[2]/a").click()
+                time.sleep(1)
+                nav = self.driver.find_element(By.XPATH, "//ul[@id='nav']/li[3]/a").click()
+                time.sleep(1)
+                nav = self.driver.find_element(By.XPATH, "//ul[@id='nav']/li[1]/a").click()
+                time.sleep(1)
+            nav = self.driver.find_element(By.ID, 'logout').click()
             time.sleep(1)
-            nav = self.driver.find_element(By.XPATH, "//ul[@id='nav']/li[2]/a").click()
-            time.sleep(1)
-            nav = self.driver.find_element(By.XPATH, "//ul[@id='nav']/li[3]/a").click()
-            time.sleep(1)
-            nav = self.driver.find_element(By.XPATH, "//ul[@id='nav']/li[1]/a").click()
+    
+    def start_web_clicker(self):
+        self.web_thread.start()
+        
+    def start_pgw_sender(self, args):
+        if len(args) != 0:
+            if args[0] == 1:
+                self.pgw_1_thread.start()
+            if args[1] == 1:
+                self.pgw_2_thread.start()
+            if args[2] == 1:
+                self.pgw_3_thread.start()
+        else:
+            self.pgw_1_thread.start()
+            self.pgw_2_thread.start()
+            self.pgw_3_thread.start()
+
+    def portgw_sender(self, pgw):
+        while True:
+            pgw.open()
+            pgw.write(self.PGW_TEST_DATA)
+            pgw.read(len(self.PGW_TEST_DATA))
+            pgw.close()
+            time.sleep(0.1)
 
-            # element = WebDriverWait(self.driver, 10).until(ES.element_to_be_clickable(('link text', 'settings.html')))
-            # element.click()
         
 
+    
+
+def metrolog():
+    robot = MetrologTester('192.168.31.230') 
+    robot.start_pgw_sender([0, 1, 1])
+    robot.start_web_clicker()
+    
+    
 
-my_test = MetrologWebTest() 
-if my_test.login('192.168.31.230') == True:
-    my_test.test_page_click()
+if __name__ == '__main__':
+    metrolog()
+    
+    
+    

selenium/transport.py

@@ -0,0 +1,74 @@
+import socket
+
+
+class PortgwTcp():
+    def __init__(self, ip, port):
+        self.ip = ip
+        self.port = port
+
+    def open(self):
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            self.sock.connect((self.ip, self.port))
+        except socket.error:
+            print(f'Connection to {self.ip} failed. Exit')
+            return False
+        return True
+    
+    def close(self):
+        self.sock.close()
+
+    def read(self, nbytes, timeout=1):
+        data = ''
+        self.sock.settimeout(timeout)
+        while len(data) < nbytes:
+            try:
+                ret = self.sock.recv(nbytes)
+            except socket.error:
+                print("Read socker error!")
+                break
+            data += str(ret)
+        return data
+
+    def write(self, data):
+        send = self.sock.send(data)
+        if send == 0:
+            raise RuntimeError('Socker write failed')
+        return send
+
+
+class PortgwBase():
+    def __init__(self, transport):
+        self.open = transport.open
+        self.close = transport.close
+
+        self.write = transport.write
+        self.read = transport.read
+
+class PortgwBaseTcp(PortgwBase):
+    def __init__(self, ip, port):
+        transport = PortgwTcp(ip, port)
+        super(PortgwBaseTcp, self).__init__(transport)        
+
+
+def pgw_test():
+    pgw_1 = PortgwTcp('192.168.31.230', 1002)
+    pgw_1.open()
+    test_data = b"test data string"
+    print('Socker writed data:', pgw_1.write(test_data))
+    pgw_1.read(len(test_data))
+    pgw_1.close()
+
+    pgw_2 = PortgwTcp('192.168.31.230', 1003)
+    pgw_2.open()
+    test_data = b"test data string"
+    print('Socker writed data:', pgw_2.write(test_data))
+    pgw_2.read(len(test_data))
+    pgw_2.close()
+
+
+if __name__ == "__main__":
+    pgw_test()
+
+
+