diff --git a/tests/asyncapi/rabbit/v3_0_0/test_security.py b/tests/asyncapi/rabbit/v3_0_0/test_security.py new file mode 100644 index 0000000000..babe62d5ea --- /dev/null +++ b/tests/asyncapi/rabbit/v3_0_0/test_security.py @@ -0,0 +1,126 @@ +import ssl + +from faststream.app import FastStream +from faststream.asyncapi.generate import get_app_schema +from faststream.asyncapi.version import AsyncAPIVersion +from faststream.rabbit import RabbitBroker +from faststream.security import ( + BaseSecurity, + SASLPlaintext, +) + + +def test_base_security_schema(): + ssl_context = ssl.create_default_context() + security = BaseSecurity(ssl_context=ssl_context) + + broker = RabbitBroker("amqp://guest:guest@localhost:5672/", security=security) + + assert ( + broker.url == "amqps://guest:guest@localhost:5672/" # pragma: allowlist secret + ) # pragma: allowlist secret + assert broker._connection_kwargs.get("ssl_context") is ssl_context + + schema = get_app_schema(FastStream(broker, asyncapi_version=AsyncAPIVersion.v3_0)).to_jsonable() + + assert schema == { + "asyncapi": "3.0.0", + "channels": {}, + "operations": {}, + "components": {"messages": {}, "schemas": {}, "securitySchemes": {}}, + "defaultContentType": "application/json", + "info": {"description": "", "title": "FastStream", "version": "0.1.0"}, + "servers": { + "development": { + "protocol": "amqps", + "protocolVersion": "0.9.1", + "security": [], + "host": "guest:guest@localhost:5672", # pragma: allowlist secret + "pathname": "/", + } + }, + } + + +def test_plaintext_security_schema(): + ssl_context = ssl.create_default_context() + + security = SASLPlaintext( + ssl_context=ssl_context, + username="admin", + password="password", # pragma: allowlist secret + ) + + broker = RabbitBroker("amqp://guest:guest@localhost/", security=security) + + assert ( + broker.url + == "amqps://admin:password@localhost:5671/" # pragma: allowlist secret + ) # pragma: allowlist secret + assert broker._connection_kwargs.get("ssl_context") is ssl_context + + schema = get_app_schema(FastStream(broker, asyncapi_version=AsyncAPIVersion.v3_0)).to_jsonable() + assert ( + schema + == { + "asyncapi": "3.0.0", + "channels": {}, + "operations": {}, + "components": { + "messages": {}, + "schemas": {}, + "securitySchemes": {"user-password": {"type": "userPassword"}}, + }, + "defaultContentType": "application/json", + "info": {"description": "", "title": "FastStream", "version": "0.1.0"}, + "servers": { + "development": { + "protocol": "amqps", + "protocolVersion": "0.9.1", + "security": [{"user-password": []}], + "host": "admin:password@localhost:5671", # pragma: allowlist secret + "pathname": "/", + } + }, + } + ) + + +def test_plaintext_security_schema_without_ssl(): + security = SASLPlaintext( + username="admin", + password="password", # pragma: allowlist secret + ) + + broker = RabbitBroker("amqp://guest:guest@localhost:5672/", security=security) + + assert ( + broker.url + == "amqp://admin:password@localhost:5672/" # pragma: allowlist secret + ) # pragma: allowlist secret + + schema = get_app_schema(FastStream(broker, asyncapi_version=AsyncAPIVersion.v3_0)).to_jsonable() + assert ( + schema + == { + "asyncapi": "3.0.0", + "channels": {}, + "operations": {}, + "components": { + "messages": {}, + "schemas": {}, + "securitySchemes": {"user-password": {"type": "userPassword"}}, + }, + "defaultContentType": "application/json", + "info": {"description": "", "title": "FastStream", "version": "0.1.0"}, + "servers": { + "development": { + "protocol": "amqp", + "protocolVersion": "0.9.1", + "security": [{"user-password": []}], + "host": "admin:password@localhost:5672", # pragma: allowlist secret + "pathname": "/", # pragma: allowlist secret + } + }, + } + )