Add AssertChannelSettledEvent

scenario_player/runner.py

@@ -228,7 +228,7 @@ def wait_for_token_network_discovery(
 def maybe_create_token_network(
     token_network_proxy: TokenNetworkRegistry, token_proxy: CustomToken
 ) -> TokenNetworkAddress:
-    """ Make sure the token is registered with the node's network registry. """
+    """Make sure the token is registered with the node's network registry."""
     block_identifier = token_network_proxy.rpc_client.get_confirmed_blockhash()
     token_address = token_proxy.address
 
@@ -509,7 +509,7 @@ def setup_environment_and_run_main_task(self, node_addresses: Set[ChecksumAddres
     def setup_raiden_nodes_ether_balances(
         self, pool: Pool, node_addresses: Set[ChecksumAddress]
     ) -> Set[Greenlet]:
-        """ Makes sure every Raiden node has at least `NODE_ACCOUNT_BALANCE_MIN`. """
+        """Makes sure every Raiden node has at least `NODE_ACCOUNT_BALANCE_MIN`."""
 
         greenlets: Set[Greenlet] = set()
         for address in node_addresses:

scenario_player/tasks/blockchain.py

@@ -24,6 +24,9 @@
 log = structlog.get_logger(__name__)
 
 
+CHANNEL_INFO_KEY = "channel_info_key"
+
+
 def decode_event(abi_codec: ABICodec, abi: ABI, log_: LogReceipt) -> Dict:
     """Helper function to unpack event data using a provided ABI
 
@@ -86,6 +89,14 @@ def query_blockchain_events(
     ]
 
 
+def verify_config(config, required_keys):
+    all_required_options_provided = all(key in config.keys() for key in required_keys)
+    if not all_required_options_provided:
+        raise ScenarioError(
+            "Not all required keys provided. Required: " + ", ".join(required_keys)
+        )
+
+
 class AssertBlockchainEventsTask(Task):
     """Assert on blockchain events.
 
@@ -122,16 +133,10 @@ def __init__(
     ) -> None:
         super().__init__(runner, config, parent)
 
-        required_keys = ["contract_name", "event_name", "num_events"]
-        all_required_options_provided = all(key in config.keys() for key in required_keys)
-        if not all_required_options_provided:
-            raise ScenarioError(
-                "Not all required keys provided. Required: " + ", ".join(required_keys)
-            )
-
+        verify_config(config, required_keys=["contract_name", "event_name"])
         self.contract_name = config["contract_name"]
         self.event_name = config["event_name"]
-        self.num_events = config["num_events"]
+        self.num_events = config.get("num_events")
         self.event_args: Dict[str, Any] = config.get("event_args", {}).copy()
 
         self.web3 = self._runner.client.web3
@@ -171,23 +176,94 @@ def _run(self, *args, **kwargs) -> Dict[str, Any]:  # pylint: disable=unused-arg
         if self.event_args:
             for key, value in self.event_args.items():
                 if "participant" in key:
-                    if isinstance(value, int) or (isinstance(value, str) and value.isnumeric()):
-                        # Replace node index with eth address
-                        self.event_args[key] = self._runner.get_node_address(int(value))
+                    self.event_args[key] = self._get_node_address(value)
 
             event_args_items = self.event_args.items()
             # Filter the events by the given event args.
             # `.items()` produces a set like object which supports intersection (`&`)
             events = [e for e in events if e["args"] and event_args_items & e["args"].items()]
 
-        # Raise exception when events do not match
-        if not self.num_events == len(events):
+        if self.num_events is not None:
+            # Raise exception when events do not match
+            if self.num_events != len(events):
+                raise ScenarioAssertionError(
+                    f"Expected number of events ({self.num_events}) did not match the number "
+                    f"of events found ({len(events)})"
+                )
+        return {"events": events}
+
+    def _get_node_address(self, value):
+        if isinstance(value, int) or (isinstance(value, str) and value.isnumeric()):
+            # Replace node index with eth address
+            return self._runner.get_node_address(int(value))
+        return value
+
+
+class AssertChannelSettledEvent(AssertBlockchainEventsTask):
+    _name = "assert_channel_settled_event"
+    SYNCHRONIZATION_TIME_SECONDS = 0
+    DEFAULT_TIMEOUT = 5 * 60  # 5 minutes
+
+    def __init__(
+        self, runner: scenario_runner.ScenarioRunner, config: Any, parent: "Task" = None
+    ) -> None:
+        config.update({"contract_name": "TokenNetwork", "event_name": "ChannelSettled"})
+        super().__init__(runner, config, parent)
+
+        verify_config(config, required_keys=["initiator", "partner", CHANNEL_INFO_KEY])
+
+        self.initiator = self._get_node_address(config["initiator"])
+        self.initiator_amount = config.get("initiator_amount")
+        self.partner = self._get_node_address(config["partner"])
+        self.partner_amount = config.get("partner_amount")
+        self.channel_close_expected = config.get("channel_close_expected", False)
+
+    def _run(self, *args, **kwargs) -> Dict[str, Any]:  # pylint: disable=unused-argument
+        channel_infos = self._runner.task_storage[STORAGE_KEY_CHANNEL_INFO].get(
+            self._config[CHANNEL_INFO_KEY]
+        )
+        if channel_infos is None:
+            raise ScenarioError(
+                f"No stored channel info found for key '{self._config[CHANNEL_INFO_KEY]}'."
+            )
+
+        channel_identifier = channel_infos["channel_identifier"]
+        self.event_args["channel_identifier"] = int(channel_identifier)
+        event_dict = super()._run(*args, **kwargs)
+        events = event_dict["events"]
+
+        # query in "both directions"
+        events = self._filter_for_channel_settled(
+            events, self.initiator, self.initiator_amount, self.partner, self.partner_amount
+        )
+        if not events:
+            events = self._filter_for_channel_settled(
+                events, self.partner, self.partner_amount, self.initiator, self.initiator_amount
+            )
+        if len(events) != 1:
+            raise ScenarioAssertionError("Did not find expected ChannelSettled event!")
+        coop_settle_event = events[0]
+        transaction_hash = coop_settle_event["transactionHash"]
+        transaction = self.web3.eth.get_transaction(transaction_hash)
+        transaction_sender = transaction["from"]
+        if transaction_sender != self.initiator:
             raise ScenarioAssertionError(
-                f"Expected number of events ({self.num_events}) did not match the number "
-                f"of events found ({len(events)})"
+                f"The ChannelSettled event was emitted from a tx by {transaction_sender}, but was "
+                f"expected to be emitted from a tx by {self.initiator}"
             )
+        return event_dict
 
-        return {"events": events}
+    @staticmethod
+    def _filter_for_channel_settled(
+        events, participant1, participant1_amount, participant2, participant2_amount
+    ):
+        event_args = {"participant1": participant1, "participant2": participant2}
+        if participant1_amount is not None:
+            event_args["participant1_amount"] = int(participant1_amount)
+        if participant2_amount is not None:
+            event_args["participant2_amount"] = int(participant2_amount)
+        event_args_items = event_args.items()
+        return tuple(e for e in events if e["args"] and event_args_items & e["args"].items())
 
 
 class AssertMSClaimTask(Task):
@@ -200,7 +276,7 @@ def __init__(
     ) -> None:
         super().__init__(runner, config, parent)
 
-        required_keys = {"channel_info_key"}
+        required_keys = {CHANNEL_INFO_KEY}
         if not required_keys.issubset(config.keys()):
             raise ScenarioError(
                 f'Not all required keys provided. Required: {", ".join(required_keys)}'
@@ -224,12 +300,12 @@ def __init__(
 
     def _run(self, *args, **kwargs) -> Dict[str, Any]:  # pylint: disable=unused-argument
         channel_infos = self._runner.task_storage[STORAGE_KEY_CHANNEL_INFO].get(
-            self._config["channel_info_key"]
+            self._config[CHANNEL_INFO_KEY]
         )
 
         if channel_infos is None:
             raise ScenarioError(
-                f"No stored channel info found for key '{self._config['channel_info_key']}'."
+                f"No stored channel info found for key '{self._config[CHANNEL_INFO_KEY]}'."
             )
 
         # calculate reward_id
@@ -260,7 +336,7 @@ def _run(self, *args, **kwargs) -> Dict[str, Any]:  # pylint: disable=unused-arg
 
         # Filter matching events
         def match_event(event: Dict):
-            if not event["event"] == MonitoringServiceEvent.REWARD_CLAIMED:
+            if event["event"] != MonitoringServiceEvent.REWARD_CLAIMED:
                 return False
 
             event_reward_id = bytes(event["args"]["reward_identifier"])

scenario_player/utils/configuration/nodes.py

@@ -47,7 +47,7 @@ def count(self):
 
     @property
     def reuse_accounts(self) -> bool:
-        """ Should node accounts be re-used across scenario runs. """
+        """Should node accounts be re-used across scenario runs."""
         return self.dict.get("reuse_accounts", False)
 
     @property

scenario_player/utils/configuration/token.py

@@ -94,12 +94,12 @@ def address(self):
 
     @property
     def can_reuse_token(self):
-        """ Return a boolean indicating if previous token reuse is enabled and available. """
+        """Return a boolean indicating if previous token reuse is enabled and available."""
         return self._config.get("reuse", False) and self.token_file.exists()
 
     @property
     def should_reuse_token(self):
-        """ Return a boolean indicating if token reuse is enabled.  """
+        """Return a boolean indicating if token reuse is enabled."""
         return self._config.get("reuse", False)
 
     @property

scenario_player/utils/reclaim.py

@@ -328,7 +328,7 @@ def _get_all_token_network_events(
     start_block: BlockNumber,
     target_block: BlockNumber,
 ) -> Iterable[Dict]:
-    """ Read all TokenNetwork events up to the current confirmed head. """
+    """Read all TokenNetwork events up to the current confirmed head."""
 
     chain_id = ChainID(web3.eth.chainId)
     blockchain_events = BlockchainEvents(
@@ -380,7 +380,7 @@ def _withdraw_participant_left_capacity_from_channel(
     token_network: TokenNetwork,
     current_confirmed_head: BlockIdentifier,
 ) -> None:
-    """ Withdraw all tokens in channel to channel["participant1"] """
+    """Withdraw all tokens in channel to channel["participant1"]"""
     assert token_network.client.address == channel["participant1"]
 
     # Check if channel still has deposits