Update for new zigpy topology API (#211)

* Use the new zigpy topology API

* Use zigpy types to avoid duplication

* Add 3.11 to CI

* Migrate flake8 pre-commit hook to GitHub URL

* Ensure `AsyncMock` is used instead of `MagicMock` on older Pythons

* Use exact Python versions for CI

.github/workflows/ci.yml

@@ -18,7 +18,7 @@ jobs:
  runs-on: ubuntu-latest
  strategy:
  matrix:
- python-version: [3.8, 3.9, "3.10"]
+ python-version: ['3.8.14', '3.9.15', '3.10.8', '3.11.0']
  steps:
  - name: Check out code from GitHub
  uses: actions/checkout@v2
@@ -271,7 +271,7 @@ jobs:
  needs: prepare-base
  strategy:
  matrix:
- python-version: [3.8, 3.9, "3.10"]
+ python-version: ['3.8.14', '3.9.15', '3.10.8', '3.11.0']
  name: >-
  Run tests Python ${{ matrix.python-version }}
  steps:

.pre-commit-config.yaml

@@ -7,7 +7,7 @@ repos:
  - --safe
  - --quiet
 
- - repo: https://gitlab.com/pycqa/flake8
+ - repo: https://github.com/pycqa/flake8
  rev: 3.8.4
  hooks:
  - id: flake8

setup.py

@@ -21,6 +21,6 @@
  author_email="[email protected]",
  license="GPL-3.0",
  packages=find_packages(exclude=["tests"]),
- install_requires=["zigpy>=0.51.0"],
+ install_requires=["zigpy>=0.52.1"],
  tests_require=["pytest", "asynctest"],
 )

tests/test_application.py

@@ -6,7 +6,6 @@
 import pytest
 import zigpy.config
 import zigpy.device
-import zigpy.neighbor
 from zigpy.types import EUI64
 import zigpy.zdo.types as zdo_t
 
@@ -357,52 +356,37 @@ async def test_restore_neighbours(app):
  """Test neighbour restoration."""
 
  # FFD, Rx on when idle
- desc_1 = zdo_t.NodeDescriptor(1, 64, 142, 0xBEEF, 82, 82, 0, 82, 0)
- device_1 = MagicMock()
- device_1.node_desc = desc_1
- device_1.ieee = sentinel.ieee_1
- device_1.nwk = 0x1111
- nei_1 = zigpy.neighbor.Neighbor(sentinel.nei_1, device_1)
+ device_1 = app.add_device(nwk=0x0001, ieee=EUI64.convert("00:00:00:00:00:00:00:01"))
+ device_1.node_desc = zdo_t.NodeDescriptor(1, 64, 142, 0xBEEF, 82, 82, 0, 82, 0)
 
  # RFD, Rx on when idle
- desc_2 = zdo_t.NodeDescriptor(1, 64, 142, 0xBEEF, 82, 82, 0, 82, 0)
- device_2 = MagicMock()
- device_2.node_desc = desc_2
- device_2.ieee = sentinel.ieee_2
- device_2.nwk = 0x2222
- nei_2 = zigpy.neighbor.Neighbor(sentinel.nei_2, device_2)
-
- # Missing node descriptor
- device_3 = MagicMock()
- device_3.node_desc = None
- device_3.ieee = sentinel.ieee_3
- device_3.nwk = 0x3333
- nei_3 = zigpy.neighbor.Neighbor(sentinel.nei_3, device_3)
+ device_2 = app.add_device(nwk=0x0002, ieee=EUI64.convert("00:00:00:00:00:00:00:02"))
+ device_2.node_desc = zdo_t.NodeDescriptor(1, 64, 142, 0xBEEF, 82, 82, 0, 82, 0)
 
- # no device
- nei_4 = zigpy.neighbor.Neighbor(sentinel.nei_4, None)
+ device_3 = app.add_device(nwk=0x0003, ieee=EUI64.convert("00:00:00:00:00:00:00:03"))
+ device_3.node_desc = None
 
  # RFD, Rx off when idle
- desc_5 = zdo_t.NodeDescriptor(2, 64, 128, 0xBEEF, 82, 82, 0, 82, 0)
- device_5 = MagicMock()
- device_5.node_desc = desc_5
- device_5.ieee = sentinel.ieee_5
- device_5.nwk = 0x5555
- nei_5 = zigpy.neighbor.Neighbor(sentinel.nei_5, device_5)
+ device_5 = app.add_device(nwk=0x0005, ieee=EUI64.convert("00:00:00:00:00:00:00:05"))
+ device_5.node_desc = zdo_t.NodeDescriptor(2, 64, 128, 0xBEEF, 82, 82, 0, 82, 0)
 
  coord = MagicMock()
- coord.ieee = sentinel.coord_ieee
- coord.nwk = 0x0000
- neighbours = zigpy.neighbor.Neighbors(coord)
- neighbours.neighbors.append(nei_1)
- neighbours.neighbors.append(nei_2)
- neighbours.neighbors.append(nei_3)
- neighbours.neighbors.append(nei_4)
- neighbours.neighbors.append(nei_5)
- coord.neighbors = neighbours
-
- p2 = patch.object(app, "_api", spec_set=zigpy_deconz.api.Deconz(None, None))
- with patch.object(app, "get_device", return_value=coord), p2 as api_mock:
+ coord.ieee = EUI64.convert("aa:aa:aa:aa:aa:aa:aa:aa")
+
+ app.devices[coord.ieee] = coord
+ app.state.node_info.ieee = coord.ieee
+
+ app.topology.neighbors[coord.ieee] = [
+ zdo_t.Neighbor(ieee=device_1.ieee),
+ zdo_t.Neighbor(ieee=device_2.ieee),
+ zdo_t.Neighbor(ieee=device_3.ieee),
+ zdo_t.Neighbor(ieee=EUI64.convert("00:00:00:00:00:00:00:04")),
+ zdo_t.Neighbor(ieee=device_5.ieee),
+ ]
+
+ p = patch.object(app, "_api", spec_set=zigpy_deconz.api.Deconz(None, None))
+
+ with p as api_mock:
  api_mock.add_neighbour = AsyncMock()
  await app.restore_neighbours()
 
@@ -415,7 +399,6 @@ async def test_delayed_scan():
  """Delayed scan."""
 
  coord = MagicMock()
- coord.neighbors.scan = AsyncMock()
  config = application.ControllerApplication.SCHEMA(
  {
  zigpy.config.CONF_DEVICE: {zigpy.config.CONF_DEVICE_PATH: "usb0"},
@@ -425,8 +408,9 @@ async def test_delayed_scan():
 
  app = application.ControllerApplication(config)
  with patch.object(app, "get_device", return_value=coord):
- await app._delayed_neighbour_scan()
- assert coord.neighbors.scan.await_count == 1
+ with patch.object(app, "topology", AsyncMock()):
+ await app._delayed_neighbour_scan()
+ app.topology.scan.assert_called_once_with(devices=[coord])
 
 
 @patch("zigpy_deconz.zigbee.application.CHANGE_NETWORK_WAIT", 0.001)

zigpy_deconz/types.py

@@ -3,6 +3,7 @@
 import enum
 
 import zigpy.types as zigpy_t
+from zigpy.types import bitmap8, bitmap16 # noqa: F401
 
 
 def deserialize(data, schema):
@@ -128,33 +129,6 @@ class AddressMode(uint8_t, enum.Enum):
  NWK_AND_IEEE = 0x04
 
 
-def bitmap_factory(int_type: uint_t) -> enum.Flag:
- class _NewEnum(int_type, enum.Flag):
- # Rebind classmethods to our own class
- _missing_ = classmethod(enum.IntFlag._missing_.__func__)
- _create_pseudo_member_ = classmethod(
- enum.IntFlag._create_pseudo_member_.__func__
- )
-
- __or__ = enum.IntFlag.__or__
- __and__ = enum.IntFlag.__and__
- __xor__ = enum.IntFlag.__xor__
- __ror__ = enum.IntFlag.__ror__
- __rand__ = enum.IntFlag.__rand__
- __rxor__ = enum.IntFlag.__rxor__
- __invert__ = enum.IntFlag.__invert__
-
- return _NewEnum
-
-
-class bitmap8(bitmap_factory(uint8_t)):
- pass
-
-
-class bitmap16(bitmap_factory(uint16_t)):
- pass
-
-
 class DeconzSendDataFlags(bitmap8):
  NONE = 0x00
  NODE_ID = 0x01

zigpy_deconz/zigbee/application.py

@@ -13,7 +13,6 @@
 import zigpy.endpoint
 import zigpy.exceptions
 from zigpy.exceptions import FormationFailure, NetworkNotFormed
-import zigpy.neighbor
 import zigpy.state
 import zigpy.types
 import zigpy.util
@@ -126,7 +125,6 @@ async def start_network(self):
  self._config[zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH],
  )
 
- coordinator.neighbors.add_context_listener(self._dblistener)
  self.devices[self.state.node_info.ieee] = coordinator
  if self._api.protocol_version >= PROTO_VER_NEIGBOURS:
  await self.restore_neighbours()
@@ -466,10 +464,13 @@ def handle_tx_confirm(self, req_id, status):
  async def restore_neighbours(self) -> None:
  """Restore children."""
  coord = self.get_device(ieee=self.state.node_info.ieee)
- devices = (nei.device for nei in coord.neighbors)
- for device in devices:
- if device is None:
+
+ for neighbor in self.topology.neighbors[coord.ieee]:
+ try:
+ device = self.get_device(ieee=neighbor.ieee)
+ except KeyError:
  continue
+
  descr = device.node_desc
  LOGGER.debug(
  "device: 0x%04x - %s %s, FFD=%s, Rx_on_when_idle=%s",
@@ -498,7 +499,7 @@ async def _delayed_neighbour_scan(self) -> None:
  """Scan coordinator's neighbours."""
  await asyncio.sleep(DELAY_NEIGHBOUR_SCAN_S)
  coord = self.get_device(ieee=self.state.node_info.ieee)
- await coord.neighbors.scan()
+ await self.topology.scan(devices=[coord])
 
  def connection_lost(self, exc: Exception) -> None:
  """Lost connection."""
@@ -584,9 +585,6 @@ async def new(cls, application, ieee, nwk, version: int, device_path: str):
  from_dev = application.get_device(ieee=ieee)
  dev.status = from_dev.status
  dev.node_desc = from_dev.node_desc
- dev.neighbors = zigpy.neighbor.Neighbors(dev)
- for nei in from_dev.neighbors.neighbors:
- dev.neighbors.add_neighbor(nei.neighbor)
  for ep_id, from_ep in from_dev.endpoints.items():
  if not ep_id:
  continue # Skip ZDO