review: preserve staging and OAuth polling

Author: sumitake

bin/youtube-autoencoder

@@ -691,7 +691,7 @@ def reconcile_command_args() -> list[str]:
         "reconcile-broadcast",
         "--allow-create",
         "--privacy",
-        env("YTA_YOUTUBE_STAGING_PRIVACY", env("YTA_YOUTUBE_PRIVACY", "unlisted")),
+        env("YTA_YOUTUBE_STAGING_PRIVACY", "unlisted"),
         "--title-prefix",
         env("YTA_YOUTUBE_TITLE_PREFIX", "AutoEncoder Live"),
     ]

bin/youtube-autoencoder-api

@@ -237,11 +237,18 @@ def http_json(
             parsed = json.loads(raw)
         except json.JSONDecodeError:
             parsed = {"error": raw}
-        details = parsed.get("error") if isinstance(parsed, dict) else None
-        details = details if isinstance(details, dict) else {}
-        errors = details.get("errors") or []
-        reasons = tuple(str(item.get("reason")) for item in errors if isinstance(item, dict) and item.get("reason"))
-        message = str(details.get("message") or exc.reason or "YouTube API request failed")
+        parsed_object = parsed if isinstance(parsed, dict) else {}
+        error_value = parsed_object.get("error")
+        if isinstance(error_value, dict):
+            errors = error_value.get("errors") or []
+            reasons = tuple(str(item.get("reason")) for item in errors if isinstance(item, dict) and item.get("reason"))
+            message = str(error_value.get("message") or exc.reason or "YouTube API request failed")
+        elif isinstance(error_value, str):
+            reasons = (error_value,)
+            message = str(parsed_object.get("error_description") or error_value)
+        else:
+            reasons = ()
+            message = str(exc.reason or "YouTube API request failed")
         retry_after = None
         retry_after_value = exc.headers.get("Retry-After") if exc.headers else None
         if retry_after_value:
@@ -295,11 +302,10 @@ def authorize(_args: argparse.Namespace) -> int:
                     "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                 },
             )
-        except RuntimeError as exc:
-            text = str(exc)
-            if "authorization_pending" in text:
+        except YouTubeApiError as exc:
+            if "authorization_pending" in exc.reasons:
                 continue
-            if "slow_down" in text:
+            if "slow_down" in exc.reasons:
                 interval += 5
                 continue
             raise

tests/test_api.py

@@ -149,6 +149,72 @@ def fail_request(_request, timeout):
     }
 
 
+def test_http_json_preserves_oauth_device_error_reason(load_script, monkeypatch):
+    api = load_script("youtube-autoencoder-api", "yta_api_oauth_error")
+    body = json.dumps(
+        {
+            "error": "authorization_pending",
+            "error_description": "The user has not completed authorization.",
+        }
+    ).encode()
+
+    def fail_request(_request, timeout):
+        assert timeout == 30
+        raise urllib.error.HTTPError(
+            "https://oauth2.googleapis.com/token",
+            400,
+            "Bad Request",
+            {},
+            io.BytesIO(body),
+        )
+
+    monkeypatch.setattr(api.urllib.request, "urlopen", fail_request)
+
+    with pytest.raises(api.YouTubeApiError) as raised:
+        api.http_json("POST", "https://oauth2.googleapis.com/token", form={"device_code": "code"})
+
+    assert raised.value.reasons == ("authorization_pending",)
+    assert str(raised.value) == "The user has not completed authorization."
+
+
+def test_authorize_retries_pending_device_flow(load_script, monkeypatch, tmp_path):
+    api = load_script("youtube-autoencoder-api", "yta_api_authorize_pending")
+    calls = []
+    responses = iter(
+        [
+            {
+                "verification_url": "https://example.test/device",
+                "user_code": "ABCD-EFGH",
+                "device_code": "device-code",
+                "interval": 1,
+                "expires_in": 600,
+            },
+            api.YouTubeApiError(
+                status=400,
+                reasons=("authorization_pending",),
+                message="authorization_pending",
+            ),
+            {"access_token": "access", "refresh_token": "refresh", "expires_in": 3600},
+        ]
+    )
+
+    def fake_http(*_args, **_kwargs):
+        calls.append(True)
+        response = next(responses)
+        if isinstance(response, Exception):
+            raise response
+        return response
+
+    monkeypatch.setattr(api, "TOKEN_FILE", tmp_path / "youtube-token.json")
+    monkeypatch.setattr(api, "client_config", lambda: {"client_id": "id", "client_secret": "secret"})
+    monkeypatch.setattr(api, "http_json", fake_http)
+    monkeypatch.setattr(api.time, "sleep", lambda _seconds: None)
+
+    assert api.authorize(argparse.Namespace()) == 0
+    assert len(calls) == 3
+    assert json.loads(api.TOKEN_FILE.read_text(encoding="utf-8"))["refresh_token"] == "refresh"
+
+
 def test_corrupt_state_is_quarantined(load_script, monkeypatch, tmp_path):
     api = load_script("youtube-autoencoder-api", "yta_api_corrupt_state")
     state = tmp_path / "youtube-live-state.json"

tests/test_supervisor.py

@@ -616,6 +616,16 @@ def publish(_runtime, state):
     assert state["privacy"] == "public"
 
 
+def test_legacy_public_privacy_never_changes_staging_default(load_script, monkeypatch):
+    supervisor = load_script("youtube-autoencoder", "yta_staging_legacy_privacy")
+    monkeypatch.delenv("YTA_YOUTUBE_STAGING_PRIVACY", raising=False)
+    monkeypatch.setenv("YTA_YOUTUBE_PRIVACY", "public")
+
+    args = supervisor.reconcile_command_args()
+
+    assert args[args.index("--privacy") + 1] == "unlisted"
+
+
 def test_unattended_supervisor_has_no_completion_helper(load_script):
     supervisor = load_script("youtube-autoencoder", "yta_no_auto_complete")