python selenium2 动态调试

Posted 2020-10-06

#coding=utf-8
‘‘‘
Created on 2017-9-9

@author: ceshi 转自https://testerhome.com/topics/9897
‘‘‘

# rpcserver.py
import pickle
from selenium.webdriver.chrome.webdriver import WebDriver
from multiprocessing.connection import Listener
from threading import Thread

dr = WebDriver(executable_path="D:\\python27\\chromedriver")

def rpc_server(handler, address, authkey):
sock = Listener(address, authkey=authkey)
while True:
client = sock.accept()
t = Thread(target=handler.handle_connection, args=(client,))
t.daemon = True
t.start()

class RPCHandler(object):
def __init__(self):
# rpc functions map
self._functions = {}

def register_function(self, func):
self._functions[func.__name__] = func

def handle_connection(self, connection):
try:
while True:
#接收到一条消息，使用pickle协议编码
func_name, args, kwargs = pickle.loads(connection.recv())
# rpc调用函数，并返回结果
try:
r = self._functions[func_name](*args, **kwargs)
print(type(r))
connection.send(pickle.dumps(r))
except Exception as e:
connection.send(pickle.dumps(e))
except EOFError:
pass

if __name__ == ‘__main__‘:
# 写几个测试方法
def add():
reload(rpcclient)
rpcclient.get(dr)
# 新建一个handler类实例, 并将add方法注册到handler里面
import rpcclient
from imp import reload
rpc_handler = RPCHandler()

rpc_handler.register_function(add)

rpc_server(rpc_handler, (‘192.168.3.19‘, 17001), authkey=b‘tab_space‘)

==========================

#coding=utf-8
‘‘‘
Created on 2017-9-9

@author: ceshi
‘‘‘

#rpcclient.py
import pickle

class RPCProxy(object):
def __init__(self, connection):
self._connection = connection

def __getattr__(self, name):
# 通过name，得到一个函数
def do_rpc(*args, **kwargs):
self._connection.send(pickle.dumps((name, args, kwargs)))
result = pickle.loads(self._connection.recv())
if isinstance(result, Exception):
raise result
return result

return do_rpc

def get(dr):
#dr.get("https://www.baidu.com")
dr.get(‘https://testerhome.com/topics/9897‘)
# dr.get("https://www.baidu.com")
# dr.find_element_by_id("kw").send_keys("selenium")
# dr.find_element_by_id("su").click()

# 远程连接并且调用
if __name__ == ‘__main__‘:
from multiprocessing.connection import Client

rpc_client = Client((‘192.168.3.19‘, 17001), authkey=b‘tab_space‘)
proxy = RPCProxy(rpc_client)
b = proxy.add()