Add test for upload progress update

Author: YuriiMotov

tests/test_cli_deploy.py

@@ -1,4 +1,5 @@
 import random
+import re
 import string
 from datetime import timedelta
 from pathlib import Path
@@ -8,6 +9,7 @@
 import httpx
 import pytest
 import respx
+import rich_toolkit
 from click.testing import Result
 from httpx import Response
 from time_machine import TimeMachineFixture
@@ -1578,6 +1580,101 @@ def test_deploy_with_token_fails(
         )
 
 
+@pytest.mark.parametrize(
+    ("size", "expected_msgs"),
+    [
+        (
+            100,
+            [
+                r"\(\d+ bytes\)",  # e.g. "(123 bytes)"
+                r"\(\d+ bytes of \d+ bytes\)",  # e.g. "(123 bytes of 456 bytes)"
+            ],
+        ),
+        (
+            10 * 1024,
+            [
+                r"\(\d+\.\d+ KB\)",  # e.g. "(1.23 KB)"
+                r"\(\d+\.\d+ KB of \d+\.\d+ KB\)",  # e.g. "(1.23 KB of 4.56 KB)"
+            ],
+        ),
+        (
+            10 * 1024 * 1024,
+            [
+                r"\(\d+\.\d+ MB\)",  # e.g. "(1.23 MB)"
+                r"\(\d+\.\d+ KB of \d+\.\d+ MB\)",  # e.g. "(1.23 KB of 4.56 MB)"
+                r"\(\d+\.\d+ MB of \d+\.\d+ MB\)",  # e.g. "(1.23 MB of 4.56 MB)"
+            ],
+        ),
+    ],
+)
+@pytest.mark.respx
+def test_upload_deployment_progress(
+    logged_in_cli: None,
+    tmp_path: Path,
+    respx_mock: respx.MockRouter,
+    size: int,
+    expected_msgs: list[str],
+) -> None:
+    app_data = _get_random_app()
+    team_data = _get_random_team()
+    app_id = app_data["id"]
+    team_id = team_data["id"]
+    deployment_data = _get_random_deployment(app_id=app_id)
+    deployment_id = deployment_data["id"]
+
+    config_path = tmp_path / ".fastapicloud" / "cloud.json"
+    config_path.parent.mkdir(parents=True, exist_ok=True)
+    config_path.write_text(f'{{"app_id": "{app_id}", "team_id": "{team_id}"}}')
+
+    (tmp_path / "file.bin").write_bytes(random.randbytes(size))
+
+    respx_mock.get(f"/apps/{app_id}").mock(return_value=Response(200, json=app_data))
+    respx_mock.post(f"/apps/{app_id}/deployments/").mock(
+        return_value=Response(201, json=deployment_data)
+    )
+    respx_mock.post(f"/deployments/{deployment_id}/upload").mock(
+        return_value=Response(
+            200,
+            json={"url": "http://test.com", "fields": {"key": "value"}},
+        )
+    )
+    respx_mock.post("http://test.com", data={"key": "value"}).mock(
+        return_value=Response(200)
+    )
+    respx_mock.post(f"/deployments/{deployment_id}/upload-complete").mock(
+        return_value=Response(200)
+    )
+    respx_mock.get(f"/deployments/{deployment_id}/build-logs").mock(
+        return_value=Response(
+            200,
+            content=build_logs_response(
+                {"type": "message", "message": "Building...", "id": "1"},
+                {"type": "complete"},
+            ),
+        )
+    )
+    respx_mock.get(f"/apps/{app_id}/deployments/{deployment_id}").mock(
+        return_value=Response(200, json={**deployment_data, "status": "success"})
+    )
+
+    with (
+        changing_dir(tmp_path),
+        patch.object(rich_toolkit.toolkit.Progress, "log") as mock_progress,
+    ):
+        result = runner.invoke(app, ["deploy"])
+        assert result.exit_code == 0
+
+    call_args = [
+        c.args[0] for c in mock_progress.call_args_list if isinstance(c.args[0], str)
+    ]
+
+    for expected_msg in expected_msgs:
+        pattern = re.compile(f"Uploading deployment {expected_msg}\\.\\.\\.")
+        assert any(pattern.match(arg) for arg in call_args), (
+            f"Expected message '{pattern.pattern}' not found in {call_args}"
+        )
+
+
 @pytest.mark.respx
 def test_deploy_with_app_id_arg(
     logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter