test_rabotai.py 1.54 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#!/usr/bin/env python3

# Copyright (C) 2020-2021:
#   Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.

import asyncio

import pytest
import argparse
Felix Delattre's avatar
Felix Delattre committed
23
from rabotai.rabotai import RABOTAI
24
25
26
27
28
29
30
31
32
33
34
35
36


class MockMessageBus:
    def __init__(self):
        self.calls = []
        self.connected = asyncio.Event()
        self.connected.set()

    async def send(self, *args, **kwargs):
        self.calls.append((args, kwargs))


@pytest.mark.asyncio
Felix Delattre's avatar
Felix Delattre committed
37
async def test_rabotai():
38
39
40
    message_bus = MockMessageBus()

    parser = argparse.ArgumentParser()
Felix Delattre's avatar
Felix Delattre committed
41
42
    rabotai = RABOTAI(parser=parser)
    rabotai.set_message_bus(message_bus)
43

Felix Delattre's avatar
Felix Delattre committed
44
    rabotai.setup_parser()
45
46
47
48
49

    help_formatted = parser.format_help()
    assert help_formatted.startswith("usage:")

    arguments = {"building_id": "1"}
Felix Delattre's avatar
Felix Delattre committed
50
    await rabotai.execute(arguments)
51
52

    assert message_bus.calls == [((), {"message": "building", "payload": {"building_id": "1"}})]