实现连接策略的插件化

pyproject.toml

@@ -1,15 +1,21 @@
 [tool.poetry]
 name = "robotframework-airtest"
-version = "0.1.1"
+version = "0.1.2"
 description = "网易Airtest/pocoui的Robotframework测试库"
 authors = ["kaluluosi <[email protected]>"]
 license = "MIT"
 readme = "README.md"
 packages = [{ include = "robotframework_airtest" }]
 repository = "https://github.com/kaluluosi/robotframework-airtest"
 
-[tool.poetry.plugins."vmg.generator"]
-unity = "robotframework_airtest.vmg.generators.unity"
+[tool.poetry.plugins."robotframework_airtest.vmg.generator"]
+unity = "robotframework_airtest.vmg.generators.unity:generate"
+
+[tool.poetry.plugins."robotframework_airtest.device.connect_strategy"]
+android = "robotframework_airtest.device.connects.impl.android:AndroidConnectStrategy"
+windows = "robotframework_airtest.device.connects.impl.windows:WindowsConnectStrategy"
+unity = "robotframework_airtest.device.connects.impl.unity:UnityConnectStrategy"
+
 
 [tool.poetry.scripts]
 ra = "robotframework_airtest.cli:main"

robotframework_airtest/device/__init__.py

@@ -2,7 +2,7 @@
 from airtest.core.settings import Settings
 from robot.api import logger, deco
 from robot.libraries.BuiltIn import BuiltIn
-from .connects import ConnectStrategy, factory
+from .connects import ConnectStrategyBase, factory
 
 # 坐标点类型声明
 Point = Tuple[float, float]
@@ -26,7 +26,7 @@ def __init__(
         self.auto_start_app = auto_start_app or self.var("${auto_start_app}") == "True"
         self.device_uri = device_uri or self.var("${device_uri}")
         self.pkg_name = pkg_name or self.var("${pkg_name}")
-        self.conn: ConnectStrategy = None
+        self.conn: ConnectStrategyBase = None
         logger.console(
             "DeviceLibrary初始化 device_uri:{} pkg_name:{} auto_start_app:{}".format(
                 self.device_uri,

robotframework_airtest/device/connects/__init__.py

@@ -1,4 +1,4 @@
-from .connect_strategy import ConnectStrategy
+from .connect_strategy import ConnectStrategyBase
 from .connect import factory
 
-__all__ = ["ConnectStrategy", "factory"]
+__all__ = ["ConnectStrategyBase", "factory"]

robotframework_airtest/device/connects/connect.py

@@ -29,29 +29,40 @@
     找到匹配platform的入口，加载这个入获取ConnectStrategy类，然后用这个类来实例化具体的ConnectStrategy。
 """
 
-import importlib
+from importlib import metadata
+from typing import Dict
 from urllib.parse import urlparse
-from .connect_strategy import ConnectStrategy
-from . import impl
+from .connect_strategy import ConnectStrategyBase
 
 
-def factory(device_uri: str, pkg_name: str) -> ConnectStrategy:
+def get_valid_connect_strategy() -> Dict[str, ConnectStrategyBase]:
+    """获取所有支持的连接策略
+
+    Returns:
+        Dict[str,ConnectStrategy]: 支持的连接策略
+    """
+    entry_points = metadata.entry_points().get(
+        "robotframework_airtest.device.connect_strategy", []
+    )
+    all_connect_strategies = {}
+    for ep in entry_points:
+        connect_strategy = ep.load()
+        all_connect_strategies[ep.name] = connect_strategy
+
+    return all_connect_strategies
+
+
+def factory(device_uri: str, pkg_name: str) -> ConnectStrategyBase:
     """动态加载实现的工厂，用来创建连接策略实例
 
     Returns:
         ConnectStrategy: 连接策略实例
     """
     res = urlparse(device_uri)
     platform = res.scheme
-    try:
-        module = importlib.import_module("." + platform.lower(), impl.__package__)
-        cls_connect_strategy = getattr(
-            module, "{}ConnectStrategy".format(platform.title())
-        )
-    except Exception as e:
-        # 预设的连接策略里没有找到该平台的连接策略，尝试去插件里找
-        # TODO: 要重新实现插件，这里先直接抛错
-
-        raise RuntimeError(f"没有合适的连接策略，不支持该平台连接 {e}")
+    all_connect_strategies = get_valid_connect_strategy()
+    connect_strategy = all_connect_strategies.get(platform, None)
+    if connect_strategy is None:
+        raise RuntimeError(f"没有合适的连接策略，不支持 {platform} 平台连接")
 
-    return cls_connect_strategy(device_uri, pkg_name)
+    return connect_strategy(device_uri, pkg_name)

robotframework_airtest/device/connects/connect_strategy.py

@@ -3,7 +3,7 @@
 from airtest.core.helper import G
 
 
-class ConnectStrategy(ABC):
+class ConnectStrategyBase(ABC):
     def __init__(self, device_uri: str, pkg_name: str):
         """初始化连接策略
 

robotframework_airtest/device/connects/impl/android.py

@@ -2,10 +2,10 @@
 from robot.api import logger
 from airtest.core.api import connect_device
 from airtest.core.helper import G
-from ..connect_strategy import ConnectStrategy
+from ..connect_strategy import ConnectStrategyBase
 
 
-class AndroidConnectStrategy(ConnectStrategy):
+class AndroidConnectStrategy(ConnectStrategyBase):
     def connect(self, auto_start_app=False) -> Device:
         self.device = connect_device(self.device_uri)
         logger.console("设备<{device}>：连接".format(device=self.device_uri))

robotframework_airtest/device/connects/impl/windows.py

@@ -19,7 +19,7 @@
 
 from robot.api import logger
 
-from ..connect_strategy import ConnectStrategy
+from ..connect_strategy import ConnectStrategyBase
 
 
 DLL_PATH = pkg_resources.resource_filename(robotframework_airtest.__package__, "dll")
@@ -130,7 +130,7 @@ def _snapshot(self, *args, **kwargs):
     return win_device
 
 
-class WindowsConnectStrategy(ConnectStrategy):
+class WindowsConnectStrategy(ConnectStrategyBase):
     def connect(self, auto_start_app: bool = False) -> Device:
         if auto_start_app:
             # 尝试连接一下看是不是正有一个客户端开着

robotframework_airtest/reporter/__init__.py

@@ -15,12 +15,14 @@
 
 
 class AirtestReporter:
-    """这个模块是Robotframework监听器，通过再robot命令后面追加
-    --listener robotframework_airtest.reporter.AirtestReporter，可以在测试执行的时候
-    生成airtest的测试报告。
+    """
+    Airtest日志导出报告监听器
+    NOTE: 如果robot没有指明这个监听器，那么将不会导出airtest的报告也不会插入到robot的报告里
+
+    通过再robot命令后面追加
+    `--listener robotframework_airtest.reporter.AirtestReporter` 可以在测试执行的时候生成airtest的测试报告。
+
 
-    Returns:
-        [type]: [description]
     """
 
     ROBOT_LISTENER_API_VERSION = 2

robotframework_airtest/vmg/__init__.py

@@ -9,14 +9,16 @@
 
 
 def get_valid_generators() -> dict:
-    entrypoints = metadata.entry_points().get("vmg.generator", [])
+    entrypoints = metadata.entry_points().get(
+        "robotframework_airtest.vmg.generator", []
+    )
     all_generators = {}
     for ep in entrypoints:
         try:
-            module: ModuleType = ep.load()
-            all_generators[ep.name] = getattr(module, "generate")
+            generator = ep.load()
+            all_generators[ep.name] = generator
         except Exception as e:
-            logger.error(f"{module.__name__} {e}")
+            logger.error(f"{ep.__name__} {e}")
 
     return all_generators