bootstrap-python-fastapi/tests/common/test_asyncapi.py at 844fceee0b0704e1e09a82f15094475a6fcf21d5 · febus982/bootstrap-python-fastapi · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import pytest
from pydantic import BaseModel

from common.asyncapi import (
    get_schema,
    init_asyncapi_info,
    register_channel,
    register_channel_operation,
    register_server,
)


# Test fixtures
@pytest.fixture
def reset_asyncapi_state():
    """Reset all global state between tests"""
    from common.asyncapi import _channels, _components_schemas, _operations, _servers

    _servers.clear()
    _channels.clear()
    _operations.clear()
    _components_schemas.clear()
    yield
    _servers.clear()
    _channels.clear()
    _operations.clear()
    _components_schemas.clear()


# Test message models
class SomeTestMessage(BaseModel):
    content: str
    timestamp: int


class AnotherTestMessage(BaseModel):
    status: bool
    code: int


# Test cases
def test_init_asyncapi_info():
    """Test initialization of AsyncAPI info"""
    title = "Test API"
    version = "2.0.0"

    init_asyncapi_info(title=title, version=version)
    schema = get_schema()

    assert schema.info.title == title
    assert schema.info.version == version


def test_register_server(reset_asyncapi_state):
    """Test server registration"""
    server_id = "test-server"
    host = "localhost"
    protocol = "ws"
    pathname = "/ws"

    register_server(id=server_id, host=host, protocol=protocol, pathname=pathname)

    schema = get_schema()
    assert server_id in schema.servers
    assert schema.servers[server_id].host == host
    assert schema.servers[server_id].protocol == protocol
    assert schema.servers[server_id].pathname == pathname


def test_register_channel(reset_asyncapi_state):
    """Test channel registration"""
    channel_id = "test-channel"
    address = "test/topic"
    description = "Test channel"
    title = "Test Channel"

    register_channel(address=address, id=channel_id, description=description, title=title)

    schema = get_schema()
    assert channel_id in schema.channels
    assert schema.channels[channel_id].address == address
    assert schema.channels[channel_id].description == description
    assert schema.channels[channel_id].title == title


def test_register_channel_with_server(reset_asyncapi_state):
    """Test channel registration with server reference"""
    server_id = "test-server"
    channel_id = "test-channel"

    register_server(id=server_id, host="localhost", protocol="ws")
    register_channel(address="test/topic", id=channel_id, server_id=server_id)

    schema = get_schema()
    assert len(schema.channels[channel_id].servers) == 1
    assert schema.channels[channel_id].servers[0].ref == f"#/servers/{server_id}"


def test_register_channel_operation(reset_asyncapi_state):
    """Test channel operation registration"""
    channel_id = "test-channel"
    operation_type = "receive"

    register_channel(address="test/topic", id=channel_id)
    register_channel_operation(
        channel_id=channel_id,
        operation_type=operation_type,
        messages=[SomeTestMessage],
        operation_name="test-operation",
    )

    schema = get_schema()
    assert "test-operation" in schema.operations
    assert schema.operations["test-operation"].action == operation_type
    assert schema.operations["test-operation"].channel.ref == f"#/channels/{channel_id}"
    assert SomeTestMessage.__name__ in schema.components.schemas


def test_register_channel_operation_invalid_channel(reset_asyncapi_state):
    """Test channel operation registration with invalid channel"""
    with pytest.raises(ValueError, match="Channel non-existent does not exist"):
        register_channel_operation(channel_id="non-existent", operation_type="receive", messages=[SomeTestMessage])


def test_multiple_messages_registration(reset_asyncapi_state):
    """Test registration of multiple messages for an operation"""
    channel_id = "test-channel"

    register_channel(address="test/topic", id=channel_id)
    register_channel_operation(
        channel_id=channel_id, operation_type="send", messages=[SomeTestMessage, AnotherTestMessage]
    )

    schema = get_schema()
    assert SomeTestMessage.__name__ in schema.components.schemas
    assert AnotherTestMessage.__name__ in schema.components.schemas
    assert SomeTestMessage.__name__ in schema.channels[channel_id].messages
    assert AnotherTestMessage.__name__ in schema.channels[channel_id].messages