| 1 | """Protocol and bounded recovery regressions retained from PRs 1065 and 1064.""" |
| 2 | |
| 3 | from unittest.mock import patch |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from lib import bluesky, http |
| 8 | |
| 9 | |
| 10 | @pytest.fixture(autouse=True) |
| 11 | def reset_session(): |
| 12 | bluesky._reset_session_cache() |
| 13 | yield |
| 14 | bluesky._reset_session_cache() |
| 15 | |
| 16 | |
| 17 | def unauthorized(): |
| 18 | return http.HTTPError("Unauthorized", 401, "") |
| 19 | |
| 20 | |
| 21 | def search(): |
| 22 | return bluesky.search_bluesky( |
| 23 | "test", "2026-08-01", "2026-09-01", |
| 24 | config={"BSKY_HANDLE": "fake.bsky.social", "BSKY_APP_PASSWORD": "fake-password"}, |
| 25 | ) |
| 26 | |
| 27 | |
| 28 | @pytest.mark.parametrize("second_unauthorized", [False, True]) |
| 29 | def test_missing_refresh_token_uses_one_fresh_login(second_unauthorized): |
| 30 | with patch.object(bluesky.http, "request", side_effect=[ |
| 31 | {"accessJwt": "old"}, unauthorized(), {"accessJwt": "new"}, |
| 32 | unauthorized() if second_unauthorized else {"posts": []}, |
| 33 | ]) as request: |
| 34 | result = search() |
| 35 | assert request.call_count == 4 |
| 36 | assert request.call_args_list[2].args[1] == bluesky.BSKY_SESSION_URL |
| 37 | assert request.call_args_list[3].kwargs["headers"] == {"Authorization": "Bearer new"} |
| 38 | if second_unauthorized: |
| 39 | assert "remained unauthorized" in result["error"] |
| 40 | assert result["error"] != "refresh" |
| 41 | else: |
| 42 | assert "error" not in result |
| 43 | |
| 44 | |
| 45 | @pytest.mark.parametrize("status", [400, 401]) |
| 46 | def test_rejected_refresh_uses_one_fresh_login(status): |
| 47 | with patch.object(bluesky.http, "request", side_effect=[ |
| 48 | {"accessJwt": "old", "refreshJwt": "refresh-old"}, unauthorized(), |
| 49 | http.HTTPError("Rejected", status, ""), |
| 50 | {"accessJwt": "new", "refreshJwt": "refresh-new"}, {"posts": []}, |
| 51 | ]) as request: |
| 52 | result = search() |
| 53 | assert "error" not in result |
| 54 | assert request.call_count == 5 |
| 55 | assert request.call_args_list[3].args[1] == bluesky.BSKY_SESSION_URL |
| 56 | |
| 57 | |
| 58 | @pytest.mark.parametrize("status", [429, 500]) |
| 59 | def test_transient_refresh_never_falls_back_to_login_or_leaks_tokens(status, capsys): |
| 60 | with patch.object(bluesky.http, "request", side_effect=[ |
| 61 | {"accessJwt": "old", "refreshJwt": "SENTINEL-REFRESH"}, unauthorized(), |
| 62 | http.HTTPError("Echoed SENTINEL-REFRESH", status, "SENTINEL-REFRESH"), |
| 63 | ]) as request: |
| 64 | result = search() |
| 65 | assert request.call_count == 3 |
| 66 | assert "refresh failed" in result["error"] |
| 67 | assert "SENTINEL-REFRESH" not in result["error"] + capsys.readouterr().err |
| 68 | |
| 69 | |
| 70 | def test_missing_access_token_does_not_replace_cached_tokens(): |
| 71 | with patch.object(bluesky.http, "request", side_effect=[ |
| 72 | {"accessJwt": "old", "refreshJwt": "refresh-old"}, unauthorized(), |
| 73 | {"refreshJwt": "malformed-rotation"}, |
| 74 | ]) as request: |
| 75 | result = search() |
| 76 | assert request.call_count == 3 |
| 77 | assert "No accessJwt" in result["error"] |
| 78 | assert bluesky._cached_token == "old" |
| 79 | assert bluesky._cached_refresh_token == "refresh-old" |
| 80 | |
| 81 | |
| 82 | def test_refreshed_search_second_401_is_actionable_not_sentinel(): |
| 83 | with patch.object(bluesky.http, "request", side_effect=[ |
| 84 | {"accessJwt": "old", "refreshJwt": "refresh-old"}, unauthorized(), |
| 85 | {"accessJwt": "new", "refreshJwt": "refresh-new"}, unauthorized(), |
| 86 | ]) as request: |
| 87 | result = search() |
| 88 | assert request.call_count == 4 |
| 89 | assert "remained unauthorized" in result["error"] |
| 90 | assert result["error"] != "refresh" |
| 91 |