иним чиним чиним чиним а так же новая система друзей и бутстраппера
This commit is contained in:
@@ -72,10 +72,66 @@ class TestPassMyStatus:
|
||||
"""Test /auth/pass/my endpoint."""
|
||||
|
||||
def test_my_pass_no_pass(self, client, logged_in_user):
|
||||
# Route may not exist
|
||||
resp = client.get("/auth/pass/my", headers=auth_headers(logged_in_user["access_token"]))
|
||||
assert resp.status_code in (200, 404)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
assert "has_active" in data
|
||||
assert data["has_active"] is False
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data == {"has_active": False}
|
||||
|
||||
def test_my_pass_with_pass(self, client, logged_in_user_with_pass):
|
||||
conn = sqlite3.connect(str(auth.AUTH_DB))
|
||||
pass_code = f"PASS-{secrets.token_hex(4)}"
|
||||
conn.execute("INSERT INTO passes (code, is_active, max_uses, uses) VALUES (?, 1, 1, 0)", (pass_code,))
|
||||
conn.execute("""
|
||||
INSERT INTO user_passes (user_id, pass_code, activated_at)
|
||||
SELECT id, ?, ? FROM users WHERE username = ?
|
||||
""", (pass_code, time.time(), logged_in_user_with_pass["username"]))
|
||||
conn.execute("UPDATE passes SET uses = 1 WHERE code = ?", (pass_code,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
resp = client.get("/auth/pass/my", headers=auth_headers(logged_in_user_with_pass["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data == {"has_active": True}
|
||||
|
||||
def test_my_pass_after_activation(self, client, logged_in_user):
|
||||
pass_code = f"AFTER-{secrets.token_hex(4)}"
|
||||
conn = sqlite3.connect(str(auth.AUTH_DB))
|
||||
conn.execute("INSERT INTO passes (code, is_active, max_uses, uses) VALUES (?, 1, 1, 0)", (pass_code,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
resp = client.post("/auth/pass/activate", json={"pass_code": pass_code},
|
||||
headers=auth_headers(logged_in_user["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
|
||||
resp = client.get("/auth/pass/my", headers=auth_headers(logged_in_user["access_token"]))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data == {"has_active": True}
|
||||
|
||||
def test_my_pass_stale_jwt_role(self, client, registered_user):
|
||||
"""Test that /auth/pass/my works even if JWT has stale role.
|
||||
|
||||
Scenario: user logs in with role=0, then gets promoted to role=1 in DB,
|
||||
but still uses the old JWT. The endpoint should check DB directly."""
|
||||
resp = client.post("/auth/login", json=registered_user)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
old_token = data["access_token"]
|
||||
assert data["role"] == 0
|
||||
|
||||
conn = sqlite3.connect(str(auth.AUTH_DB))
|
||||
conn.execute("UPDATE users SET role = 1 WHERE username = ?", (registered_user["username"],))
|
||||
pass_code = f"STALE-{secrets.token_hex(4)}"
|
||||
conn.execute("INSERT INTO passes (code, is_active, max_uses, uses) VALUES (?, 1, 1, 0)", (pass_code,))
|
||||
conn.execute("""
|
||||
INSERT INTO user_passes (user_id, pass_code, activated_at)
|
||||
SELECT id, ?, ? FROM users WHERE username = ?
|
||||
""", (pass_code, time.time(), registered_user["username"]))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
resp = client.get("/auth/pass/my", headers=auth_headers(old_token))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data == {"has_active": True}, "Should detect active pass despite stale JWT role"
|
||||
|
||||
Reference in New Issue
Block a user