Function test_channel_state_changed_unilateral

tests/test_plugin.py:896–1004

We open, disconnect, force-close a channel and check for notifications. The misc_notifications.py plugin logs `channel_state_changed` events.

(node_factory, bitcoind)

@pytest.mark.openchannel('v1')
@pytest.mark.openchannel('v2')
def test_channel_state_changed_unilateral(node_factory, bitcoind):
    """ We open, disconnect, force-close a channel and check for notifications.

    The misc_notifications.py plugin logs `channel_state_changed` events.
    """
    opts = {"plugin": os.path.join(os.getcwd(), "tests/plugins/misc_notifications.py"),
            "allow_warning": True,
            'may_reconnect': True}

    l1, l2 = node_factory.line_graph(2, opts=opts)

    l1_id = l1.rpc.getinfo()["id"]
    cid = l1.get_channel_id(l2)
    scid = l1.get_channel_scid(l2)

    # a helper that gives us the next channel_state_changed log entry
    def wait_for_event(node):
        msg = node.daemon.wait_for_log("channel_state_changed.*new_state.*")
        event = ast.literal_eval(re.findall(".*({.*}).*", msg)[0])
        return event

    event2 = wait_for_event(l2)
    assert(event2['peer_id'] == l1_id)
    assert(event2['channel_id'] == cid)
    assert(event2['short_channel_id'] is None)
    assert(event2['cause'] == "remote")

    if l2.config('experimental-dual-fund'):
        assert(event2['old_state'] == "DUALOPEND_OPEN_INIT")
        assert(event2['new_state'] == "DUALOPEND_AWAITING_LOCKIN")
        assert(event2['message'] == "Sigs exchanged, waiting for lock-in")
    else:
        assert(event2['old_state'] == "unknown")
        assert(event2['new_state'] == "CHANNELD_AWAITING_LOCKIN")
        assert(event2['message'] == "new channel opened")

    event2 = wait_for_event(l2)
    assert(event2['short_channel_id'] == scid)
    if l2.config('experimental-dual-fund'):
        assert(event2['old_state'] == "DUALOPEND_AWAITING_LOCKIN")
    else:
        assert(event2['old_state'] == "CHANNELD_AWAITING_LOCKIN")
    assert(event2['new_state'] == "CHANNELD_NORMAL")
    assert(event2['cause'] == "remote")
    assert(event2['message'] == "Lockin complete")

    # close channel unilaterally and look for stateful events
    l1.rpc.stop()
    wait_for(lambda: not only_one(l2.rpc.listpeers()['peers'])['connected'])
    l2.rpc.close(scid, 1)  # force close after 1sec timeout

    event2 = wait_for_event(l2)
    assert(event2['old_state'] == "CHANNELD_NORMAL")
    assert(event2['new_state'] == "CHANNELD_SHUTTING_DOWN")
    assert(event2['cause'] == "user")
    assert(event2['message'] == "User or plugin invoked close command")
    event2 = wait_for_event(l2)
    assert(event2['old_state'] == "CHANNELD_SHUTTING_DOWN")