path: root/pyagl/tests/test_can.py

# Copyright (C) 2020 Konsulko Group
# Author: Edi Feschiyan
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import os
import pytest
import logging
import asyncssh
import subprocess
from pathlib import Path

from pyagl.services.base import AFBResponse, AFBT

from pyagl.services.can import CANService as cs
pytestmark = [pytest.mark.asyncio, pytest.mark.can_low_level]


@pytest.fixture(scope='module')
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop


@pytest.fixture(scope='module')
async def service():
    address = os.environ.get('AGL_TGT_IP', 'localhost')
    port = os.environ.get('AGL_TGT_PORT', None)
    svc = await cs(ip=address, port=port)
    yield svc
    await svc.websocket.close()


@pytest.fixture()
async def prepare_replay_file(service: cs, request):
    path = Path(__file__).parent
    data = path.joinpath('data', 'can', request.param)
    if service.ip != 'localhost':
        async with asyncssh.connect(service.ip, username='root', known_hosts=None) as ssh:
            await asyncssh.scp(str(data), (ssh, '/tmp/'))


# Helper class to run canplayer
class canplayer:
    service: cs
    filename: str
    remote: bool
    args = None
    task = None
    ssh = None

    def __init__(self, service : cs):
        self.service = service
        self.remote = service.ip != 'localhost'
        self.interface = os.environ.get('AGL_CAN_INTERFACE', 'can0')
        self.args = [ '/usr/bin/canplayer', '-I' ]

    async def play(self, filename : str):
        if self.remote:
            filename = '/tmp/' + filename
            args = self.args
            args.append(filename)
            if self.interface != "can0":
                args.append(self.interface + "=can0")
            self.ssh = await asyncssh.connect(self.service.ip, username='root', known_hosts=None)
            self.task = await self.ssh.create_process(' '.join(args))
        else:
            path = Path(__file__).parent
            data = path.joinpath('data', 'can', filename)
            args = self.args
            args.append(str(data))
            if self.interface != "can0":
                args.append(self.interface + "=can0")
            self.task = subprocess.Popen(args)

    async def stop(self):
        if self.remote:
            # Ideally we'd just call self.task.kill() here, but that does not work
            # with OpenSSH's sshd as the remote side, as it does not implement
            # signalling.  Just running a killall over the open connection is the
            # seemingly simplest solution.  Other more complicated options would
            # be to hook up the stdin and set the term type so that a break
            # character can be sent, or tinkering with the command to get the pid
            # over stdout so the specific process can be killed...
            # Just closing the connection leaves the canplayer running in the
            # background, which can throw off subsequent tests.
            await self.ssh.run('killall -9 canplayer')
            self.ssh.close()
        else:
            self.task.kill()
        self.task = None


# Common test CAN message contents
hsmessage = {'bus_name': 'hs',
             'frame': {
                 'can_id': 1568,
                 'can_dlc': 8,
                 'can_data': [255, 255, 255, 255, 255, 255, 255, 255]
             }}


# Basic tests

@pytest.mark.dependency()
async def test_list(event_loop, service: cs):
    msgid = await service.list()
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


async def test_get(event_loop, service: cs):
    msgid = await service.get("engine.speed")
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


@pytest.mark.dependency(depends=['test_list'])
async def test_get_all_messages(event_loop, service: cs):
    msgid = await service.list()
    resp = await service.afbresponse()
    messagelist = [m for m in resp.data if m.startswith('messages.')]
    for m in messagelist:
        msgid = await service.get(m)
        resp = await service.afbresponse()
        assert resp.status == 'success', f'.get() failed with message {m}'

@pytest.mark.xfail(reason='This test can fail on the 2nd invocation. See SPEC-3648. XFAIL until root-cause if found.')
async def test_write_wo_auth(event_loop, service: cs):
    msgid = await service.write({'signal_name': 'engine.speed', 'signal_value': 12})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


@pytest.mark.dependency()
async def test_auth(event_loop, service: cs):
    msgid = await service.auth()
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


@pytest.mark.dependency(depends=['test_auth'])
async def test_write(event_loop, service: cs):
    msgid = await service.write({"signal_name": "hvac.temperature.left", "signal_value": 21})
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


async def test_write_raw(event_loop, service: cs):
    msgid = await service.write(hsmessage)
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


async def test_write_unwritable_signal(event_loop, service: cs):
    msgid = await service.write({'signal_name': 'vehicle.average.speed', 'signal_value': 1234})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_wrong_value_key(event_loop, service: cs):
    msgid = await service.write({'name': 'vehicle.average.speed', 'signal_value': 21})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_bus_key(event_loop, service: cs):
    message = dict(hsmessage)
    message['bus'] = message.pop('bus_name')
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_frame_key(event_loop, service: cs):
    message = dict(hsmessage)
    message['fram'] = message.pop('frame')
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_can_id_key(event_loop, service: cs):
    message = dict(hsmessage)
    message['frame']['id'] = message['frame'].pop('can_id')
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_can_id_args(event_loop, service: cs):
    message = dict(hsmessage)
    message['frame']['can_id'] = "1568"
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_can_dlc_key(event_loop, service: cs):
    message = dict(hsmessage)
    message['frame']['dlc'] = message['frame'].pop('can_dlc')
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_can_dlc_args(event_loop, service: cs):
    message = dict(hsmessage)
    message['frame']['can_dlc'] = "8"
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_can_data_key(event_loop, service: cs):
    message = dict(hsmessage)
    message['frame']['data'] = message['frame'].pop('can_data')
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_write_raw_invalid_can_data_value(event_loop, service: cs):
    message = dict(hsmessage)
    message['frame']['can_data'] = ["255", 255, 255, 255, 255, 255, 255, 255]
    msgid = await service.write(message)
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR


async def test_get_written_message(event_loop, service: cs):
    msgid = await service.get("hvac.temperature.left")
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


@pytest.mark.dependency()
async def test_subscribe(event_loop, service: cs):
    msgid = await service.subscribe('*')
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


@pytest.mark.dependency(depends=['test_subscribe'])
async def test_unsubscribe(event_loop, service: cs):
    msgid = await service.unsubscribe('*')
    resp = await service.afbresponse()
    assert resp.status == 'success', resp.info


@pytest.mark.parametrize('prepare_replay_file', ['test1.canreplay'], indirect=True)
async def test_diagnostic_engine_speed_simulation(event_loop, service: cs, prepare_replay_file):
    eventname = 'diagnostic_messages.engine.speed'
    msgid = await service.subscribe(eventname)
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname}; info: {resp.info}'

    async def listen(service: cs):
        async for resp in service.listener():
            assert 'name' in resp.data
            assert resp.data['name'] == eventname
            break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('test1.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe(eventname)
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type != AFBT.RESPONSE: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break


@pytest.mark.parametrize('prepare_replay_file', ['test2-3.canreplay'], indirect=True)
async def test_Subscribe_all(event_loop, service: cs, prepare_replay_file):
    eventname = 'messages.vehicle.average.speed'
    msgid = await service.subscribe('*')
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for all events; info: {resp.info}'

    async def listen(service: cs):
        async for resp in service.listener():
            assert 'name' in resp.data
            logging.debug(f"{resp.data['name']} = {resp.data['value']} ({resp.data['timestamp']})")
            if resp.data['name'] == 'diagnostic_messages.engine.speed':
                continue
            assert resp.data['name'] == eventname
            break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('test2-3.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe('*')
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type != AFBT.RESPONSE: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break

# End of basic tests

# Filter tests

@pytest.mark.parametrize('prepare_replay_file', ['testFilter01filteredOut.canreplay'], indirect=True)
async def test_Filter_Test_01_Step_1(event_loop, service: cs, prepare_replay_file):
    minspeed = 30
    maxspeed = 100
    eventname = 'messages.engine.speed'
    msgid = await service.subscribe({'event': eventname, 'filter': {'min': minspeed, 'max': maxspeed}})
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname} events; info: {resp.info}'

    async def listen(service: cs):
        async for resp in service.listener():
            assert resp is None, "Test failed, was supposed to timeout and return None on this fixture, but returned data"
            break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('testFilter01filteredOut.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe({'event': eventname, 'filter': {'min': minspeed, 'max': maxspeed}})
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type == AFBT.EVENT: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break


@pytest.mark.parametrize('prepare_replay_file', ['testFilter01pass.canreplay'], indirect=True)
async def test_Filter_Test_01_Step_2(event_loop, service: cs, prepare_replay_file):
    minspeed = 30
    maxspeed = 100
    eventname = 'messages.engine.speed'
    msgid = await service.subscribe({'event': eventname, 'filter': {'min': minspeed, 'max': maxspeed}})
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname} events; info: {resp.info}'

    async def listen(service: cs):
        expected = [ 30.0, 100.0, 36.0, 32.0 ]
        i = 0
        async for resp in service.listener():
            assert 'name' in resp.data
            logging.debug(f"{resp.data['name']} = {resp.data['value']} ({resp.data['timestamp']})")
            assert resp.data['name'] == eventname
            assert minspeed <= resp.data['value'] <= maxspeed
            assert resp.data['value'] == expected[i]
            i += 1
            if i == len(expected):
                break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('testFilter01pass.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe({'event': eventname, 'filter': {'min': minspeed, 'max': maxspeed}})
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type == AFBT.EVENT: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break


@pytest.mark.parametrize('prepare_replay_file', ['test2-3.canreplay'], indirect=True)
async def test_Filter_Test_01_Step_3(event_loop, service: cs, prepare_replay_file):
    # this testcase is supposed to test event filter frequency
    minspeed = 30
    maxspeed = 100
    tolerance = 100000  # usec
    eventname = 'messages.vehicle.average.speed'
    msgid = await service.subscribe({'event': eventname, 'filter': {'frequency': 1, 'min': minspeed, 'max': maxspeed}})
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname} events; info: {resp.info}'

    resplist = []
    async def listen(service: cs):
        async for resp in service.listener():
            logging.debug(f"{resp.data['name']} = {resp.data['value']} ({resp.data['timestamp']})")
            resplist.append(resp)
            if len(resplist) == 2:
                break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('test2-3.canreplay')
    await listener
    await player.stop()

    assert len(resplist) == 2
    for r in resplist:
        assert 'name' in r.data
        assert r.data['name'] == eventname
        assert 'value' in r.data
        assert minspeed < r.data['value'] < maxspeed

    # check whether the time delta between the two events is more than 1 second ('frequency' in filter above, in usec)
    delta = resplist[-1].data['timestamp'] - resplist[0].data['timestamp']
    assert 1000000 - tolerance < delta < 1000000 + tolerance

    msgid = await service.unsubscribe({'event': eventname, 'filter': {'frequency': 1, 'min': minspeed, 'max': maxspeed}})
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type == AFBT.EVENT: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break

# end of Filter tests

#
# J1939 tests
#

# Un/Subscription tests

@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding widget built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_event(event_loop, service: cs):
    msgid = await service.subscribe('Eng.Momentary.Overspeed.Enable')
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_unsubscribe_j1939_event(event_loop, service: cs):
    msgid = await service.unsubscribe('Eng.Momentary.Overspeed.Enable')
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_events(event_loop, service: cs):
    msgid = await service.subscribe('Eng.*')
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_unsubscribe_j1939_events(event_loop, service: cs):
    msgid = await service.unsubscribe('Eng.*')
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_pgn(event_loop, service: cs):
    msgid = await service.subscribe({'pgn': 61442})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_unsubscribe_j1939_pgn(event_loop, service: cs):
    msgid = await service.unsubscribe({'pgn': 61442})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_all_pgn(event_loop, service: cs):
    msgid = await service.subscribe({'pgn': '*'})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_unsubscribe_j1939_all_pgn(event_loop, service: cs):
    msgid = await service.unsubscribe({'pgn': '*'})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_id(event_loop, service: cs):
    msgid = await service.subscribe({'id': 61442})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_unsubscribe_j1939_id(event_loop, service: cs):
    msgid = await service.unsubscribe({'id': 61442})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_all_id(event_loop, service: cs):
    msgid = await service.subscribe({'id': '*'})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_unsubscribe_j1939_all_id(event_loop, service: cs):
    msgid = await service.unsubscribe({'id': '*'})
    resp = await service.afbresponse()
    assert resp.type == AFBT.RESPONSE
    assert resp.status == 'success', resp.info


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_no_event(event_loop, service: cs):
    msgid = await service.subscribe({'event': ''})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR, f'Expected ERROR type messsage, got {resp.type}'


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_no_id(event_loop, service: cs):
    msgid = await service.subscribe({'id': ''})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR, f'Expected ERROR type messsage, got {resp.type}'


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding built with SDK, see SPEC-3765')
async def test_low_can_subscribe_j1939_no_pgn(event_loop, service: cs):
    msgid = await service.subscribe({'pgn': ''})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR, f'Expected ERROR type messsage, got {resp.type}'

# End of Un/Subscription tests

# Write tests

@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding widget built with SDK, see SPEC-3765')
@pytest.mark.dependency(depends=['test_auth'])
async def test_low_can_write_j1939_wo_auth(event_loop, service: cs):
    msgid = await service.write({'signal_name': 'Eng.Momentary.Overspeed.Enable', 'signal_value': 1})
    async for resp in service.listener():  # using a listener because sometimes there are events in the queue from
        # previous tests or subscriptions even they are properly unsubscribed and awaited for confirmation
        if resp is AFBResponse and resp.type is not AFBT.ERROR:
            logging.warning(f'Expected Error response, got {resp}')
            continue
        assert resp.type == AFBT.ERROR, resp
        assert resp.msgid == msgid
        break


@pytest.mark.can_j1939
@pytest.mark.xfail(reason='May fail when run against binding widget built with SDK, see SPEC-3765')
async def test_low_can_write_j1939_signal(event_loop, service: cs):
    msgid = await service.write({'signal_name': 'Eng.Momentary.Overspeed.Enable', 'signal_value': 1})
    resp = await service.afbresponse()
    assert resp.type == AFBT.ERROR, resp


# Value tests
@pytest.mark.parametrize('prepare_replay_file', ['testValueCAN_1.canreplay'], indirect=True)
async def test_value_test_1(event_loop, service: cs, prepare_replay_file):
    eventname = 'messages.vehicle.average.speed'
    msgid = await service.subscribe(eventname)
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname} events; info: {resp.info}'

    async def listen(service: cs):
        async for resp in service.listener():
            assert 'name' in resp.data
            assert resp.data['name'] == eventname
            break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('testValueCAN_1.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe(eventname)
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type != AFBT.RESPONSE: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break


@pytest.mark.parametrize('prepare_replay_file', ['testValueCAN_2.canreplay'], indirect=True)
async def test_value_test_2(event_loop, service: cs, prepare_replay_file):
    eventname = 'diagnostic_messages.engine.speed'
    msgid = await service.subscribe(eventname)
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname} events; info: {resp.info}'

    async def listen(service: cs):
        async for resp in service.listener():
            assert 'name' in resp.data
            assert resp.data['name'] == eventname
            break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('testValueCAN_2.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe(eventname)
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type != AFBT.RESPONSE: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break


@pytest.mark.parametrize('prepare_replay_file', ['testValueCAN_3.canreplay'], indirect=True)
async def test_value_test_3(event_loop, service: cs, prepare_replay_file):
    eventname = 'diagnostic_messages.engine.load'
    msgid = await service.subscribe(eventname)
    resp = await service.afbresponse()
    assert resp.status == 'success', f'Could not subscribe for {eventname} events; info: {resp.info}'

    async def listen(service: cs):
        async for resp in service.listener():
            assert 'name' in resp.data
            assert resp.data['name'] == eventname
            break

    listener = asyncio.create_task(listen(service))
    player = canplayer(service)
    assert player is not None
    await player.play('testValueCAN_3.canreplay')
    await listener
    await player.stop()

    msgid = await service.unsubscribe(eventname)
    # wait until the event queue flushes out and we get unsubscribe confirmation
    async for resp in service.listener():
        if resp.type != AFBT.RESPONSE: continue
        if resp.msgid != msgid: continue
        assert resp.status == 'success', f'Could not unsubscribe from {eventname}; info: {resp.info}'
        break