81 lines
1.5 KiB
Python
81 lines
1.5 KiB
Python
|
import uuid
|
||
|
|
import redis
|
||
|
|
|
||
|
|
from utils.http_client import http_request
|
||
|
|
from utils.yaml_loader import load_config
|
||
|
|
from utils.template import render
|
||
|
|
|
||
|
|
|
||
|
|
config = load_config()
|
||
|
|
|
||
|
|
BASE_URL = config["base_url"]
|
||
|
|
flow = config["flows"]["register_login"]
|
||
|
|
|
||
|
|
redis_client = redis.Redis(
|
||
|
|
host="redis",
|
||
|
|
port=6379,
|
||
|
|
decode_responses=True
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_auth_flow():
|
||
|
|
|
||
|
|
context = {}
|
||
|
|
|
||
|
|
context["random_username"] = f"user_{uuid.uuid4().hex[:6]}"
|
||
|
|
|
||
|
|
# -------- register --------
|
||
|
|
|
||
|
|
register = flow["register"]
|
||
|
|
|
||
|
|
body = render(register["body"], context)
|
||
|
|
|
||
|
|
res = http_request(
|
||
|
|
register["method"],
|
||
|
|
BASE_URL + register["path"],
|
||
|
|
json=body
|
||
|
|
)
|
||
|
|
|
||
|
|
assert res["status_code"] == register["expected_status"]
|
||
|
|
|
||
|
|
assert res["json"]["msg"] == register["expected_json"]["msg"]
|
||
|
|
|
||
|
|
# -------- login --------
|
||
|
|
|
||
|
|
login = flow["login"]
|
||
|
|
|
||
|
|
body = render(login["body"], context)
|
||
|
|
|
||
|
|
res = http_request(
|
||
|
|
login["method"],
|
||
|
|
BASE_URL + login["path"],
|
||
|
|
json=body
|
||
|
|
)
|
||
|
|
|
||
|
|
assert res["status_code"] == login["expected_status"]
|
||
|
|
|
||
|
|
token_field = login["extract"]["token"]
|
||
|
|
|
||
|
|
token = res["json"][token_field]
|
||
|
|
|
||
|
|
assert token is not None
|
||
|
|
|
||
|
|
context["token"] = token
|
||
|
|
|
||
|
|
# -------- redis store --------
|
||
|
|
|
||
|
|
redis_cfg = login["store_redis"]
|
||
|
|
|
||
|
|
key = render(redis_cfg["key"], context)
|
||
|
|
|
||
|
|
redis_client.set(
|
||
|
|
key,
|
||
|
|
token,
|
||
|
|
ex=redis_cfg["ttl"]
|
||
|
|
)
|
||
|
|
|
||
|
|
saved_token = redis_client.get(key)
|
||
|
|
|
||
|
|
assert saved_token == token
|
||
|
|
|