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
|
import pytest
import functools
from unittest.mock import Mock
from azure.eventhub._pyamqp.sasl import SASLAnonymousCredential, SASLPlainCredential
from azure.eventhub._pyamqp.authentication import SASLPlainAuth, JWTTokenAuth, SASTokenAuth
def test_sasl_plain_auth():
auth = SASLPlainAuth(
authcid="authcid",
passwd="passwd",
authzid="Some Authzid"
)
assert auth.auth_type=="AUTH_SASL_PLAIN"
assert auth.sasl.mechanism==b"PLAIN"
assert auth.sasl.start() == b'Some Authzid\x00authcid\x00passwd'
def test_jwt_token_auth():
credential = Mock()
attr = {"get_token.return_value": "my_token"}
credential.configure_mock(**attr)
auth = JWTTokenAuth(
uri="my_uri",
audience="my_audience_field",
get_token=functools.partial(credential.get_token, "my_auth_uri")
)
assert auth.uri == "my_uri"
assert auth.audience == "my_audience_field"
def test_sas_token_auth():
auth = SASTokenAuth(
uri="my_uri",
audience="my_audience",
username="username",
password="password"
)
assert auth.uri == "my_uri"
assert auth.audience == "my_audience"
assert auth.username == "username"
assert auth.password == "password"
|
