test_h5_auth.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139
  1. from fastapi.testclient import TestClient
  2. from zbt.core.config import Settings
  3. from zbt.domains.identity.repository import InMemoryIdentityRepository
  4. from zbt.main import create_app
  5. def test_h5_user_can_login_with_development_otp_and_read_profile() -> None:
  6. settings = Settings(
  7. app_env="test",
  8. jwt_access_secret="a" * 32,
  9. jwt_refresh_secret="b" * 32,
  10. field_encryption_key="c" * 32,
  11. dev_fixed_otp="147258",
  12. )
  13. app = create_app(
  14. settings=settings,
  15. identity_repository=InMemoryIdentityRepository(),
  16. )
  17. with TestClient(app) as client:
  18. code_response = client.post(
  19. "/api/v1/h5/auth/request-code",
  20. json={"mobile": "18800000001"},
  21. )
  22. assert code_response.status_code == 200
  23. assert code_response.json()["data"] == {"expires_in": 300}
  24. login_response = client.post(
  25. "/api/v1/h5/auth/login",
  26. json={"mobile": "18800000001", "code": "147258"},
  27. )
  28. assert login_response.status_code == 200
  29. assert login_response.json()["data"]["user"]["mobile"] == "18800000001"
  30. tokens = login_response.json()["data"]["tokens"]
  31. profile_response = client.get(
  32. "/api/v1/h5/me",
  33. headers={"Authorization": f"Bearer {tokens['access_token']}"},
  34. )
  35. assert profile_response.status_code == 200
  36. assert profile_response.json()["data"]["mobile_masked"] == "188****0001"
  37. def test_refresh_token_is_rotated_and_cannot_be_reused() -> None:
  38. settings = Settings(
  39. app_env="test",
  40. jwt_access_secret="a" * 32,
  41. jwt_refresh_secret="b" * 32,
  42. field_encryption_key="c" * 32,
  43. dev_fixed_otp="147258",
  44. )
  45. app = create_app(
  46. settings=settings,
  47. identity_repository=InMemoryIdentityRepository(),
  48. )
  49. with TestClient(app) as client:
  50. login = client.post(
  51. "/api/v1/h5/auth/login",
  52. json={"mobile": "18800000001", "code": "147258"},
  53. ).json()["data"]["tokens"]
  54. refreshed = client.post(
  55. "/api/v1/h5/auth/refresh",
  56. json={"refresh_token": login["refresh_token"]},
  57. )
  58. assert refreshed.status_code == 200
  59. next_tokens = refreshed.json()["data"]["tokens"]
  60. assert next_tokens["refresh_token"] != login["refresh_token"]
  61. reused = client.post(
  62. "/api/v1/h5/auth/refresh",
  63. json={"refresh_token": login["refresh_token"]},
  64. )
  65. assert reused.status_code == 401
  66. assert reused.json()["error"]["code"] == "AUTH_REQUIRED"
  67. def test_logout_revokes_current_h5_session() -> None:
  68. settings = Settings(
  69. app_env="test",
  70. jwt_access_secret="a" * 32,
  71. jwt_refresh_secret="b" * 32,
  72. field_encryption_key="c" * 32,
  73. dev_fixed_otp="147258",
  74. )
  75. app = create_app(
  76. settings=settings,
  77. identity_repository=InMemoryIdentityRepository(),
  78. )
  79. with TestClient(app) as client:
  80. tokens = client.post(
  81. "/api/v1/h5/auth/login",
  82. json={"mobile": "18800000001", "code": "147258"},
  83. ).json()["data"]["tokens"]
  84. headers = {"Authorization": f"Bearer {tokens['access_token']}"}
  85. logout = client.post("/api/v1/h5/auth/logout", headers=headers)
  86. assert logout.status_code == 200
  87. profile = client.get("/api/v1/h5/me", headers=headers)
  88. assert profile.status_code == 401
  89. assert profile.json()["error"]["code"] == "AUTH_REQUIRED"
  90. def test_h5_refresh_cookie_restores_session_without_request_body() -> None:
  91. settings = Settings(
  92. app_env="test",
  93. jwt_access_secret="a" * 32,
  94. jwt_refresh_secret="b" * 32,
  95. field_encryption_key="c" * 32,
  96. dev_fixed_otp="147258",
  97. )
  98. app = create_app(
  99. settings=settings,
  100. identity_repository=InMemoryIdentityRepository(),
  101. )
  102. with TestClient(app) as client:
  103. login = client.post(
  104. "/api/v1/h5/auth/login",
  105. json={"mobile": "18800000001", "code": "147258"},
  106. )
  107. assert "zbt_h5_refresh=" in login.headers["set-cookie"]
  108. assert "HttpOnly" in login.headers["set-cookie"]
  109. refreshed = client.post("/api/v1/h5/auth/refresh")
  110. assert refreshed.status_code == 200
  111. access_token = refreshed.json()["data"]["tokens"]["access_token"]
  112. logout = client.post(
  113. "/api/v1/h5/auth/logout",
  114. headers={"Authorization": f"Bearer {access_token}"},
  115. )
  116. assert logout.status_code == 200
  117. assert "zbt_h5_refresh=" in logout.headers["set-cookie"]
  118. assert "Max-Age=0" in logout.headers["set-cookie"]