Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 57 additions & 1 deletion src/pragma_cli/commands/providers.py
Original file line number Diff line number Diff line change
Expand Up @@ -766,7 +766,7 @@ def upgrade(
console.print()

if not yes:
confirm = typer.confirm("Proceed with upgrade?")
confirm = typer.confirm(f"Upgrade {name} to v{target}?")

if not confirm:
console.print("[dim]Cancelled.[/dim]")
Expand All @@ -792,6 +792,62 @@ def upgrade(
console.print(f"[green]Upgraded:[/green] {name} -> v{result.installed_version}")


@app.command()
def downgrade(
name: Annotated[str, typer.Argument(help="Provider name (org/name format)")],
version: Annotated[str, typer.Option("--version", "-v", help="Target version to downgrade to")],
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
) -> None:
"""Downgrade an installed provider to a previous version.

Requires an explicit target version. Migrations run sequentially
through each intermediate version in reverse order.

Examples:
pragma providers downgrade pragmatiks/qdrant --version 1.0.0
pragma providers downgrade pragmatiks/postgres -v 1.2.0 -y
""" # noqa: DOC501
client = get_client()
_require_auth(client)

console.print(f"[bold]Downgrading:[/bold] {name} -> v{version}")
console.print()

if not yes:
confirm = typer.confirm(f"Downgrade {name} to v{version}?")

if not confirm:
console.print("[dim]Cancelled.[/dim]")
raise typer.Exit(0)

try:
result = _fetch_with_spinner(
"Downgrading provider...",
lambda: client.downgrade_provider(name, target_version=version),
)
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
console.print(f"[red]Error:[/red] Provider '{name}' is not installed.")
raise typer.Exit(1) from e

if e.response.status_code == 409:
console.print(f"[yellow]Warning:[/yellow] Provider '{name}' is already on v{version}.")
raise typer.Exit(1) from e

if e.response.status_code == 422:
console.print(f"[red]Error:[/red] {_format_api_error(e)}")
console.print("[dim]The version chain between current and target may be broken.[/dim]")
raise typer.Exit(1) from e

console.print(f"[red]Error:[/red] {_format_api_error(e)}")
raise typer.Exit(1) from e

console.print(f"[green]Downgraded:[/green] {name} -> v{result.installed_version}")


@app.command("list")
def list_providers(
trust_tier: Annotated[
Expand Down
58 changes: 58 additions & 0 deletions tests/test_providers.py
Original file line number Diff line number Diff line change
Expand Up @@ -729,6 +729,64 @@ def test_upgrade_already_on_version(cli_runner, mock_pragma_client):
assert "already on the requested version" in result.output


# ---------------------------------------------------------------------------
# Downgrade command tests
# ---------------------------------------------------------------------------


def test_downgrade_success(cli_runner, mock_pragma_client, mocker):
"""Downgrade command downgrades provider to specified version."""
downgraded = mocker.Mock()
downgraded.installed_version = "1.0.0"
mock_pragma_client.downgrade_provider.return_value = downgraded

result = cli_runner.invoke(app, ["providers", "downgrade", "qdrant", "--version", "1.0.0", "-y"])

assert result.exit_code == 0
assert "Downgraded" in result.output
assert "1.0.0" in result.output
mock_pragma_client.downgrade_provider.assert_called_once_with("qdrant", target_version="1.0.0")


def test_downgrade_already_on_version(cli_runner, mock_pragma_client):
"""Downgrade command handles 409 (already on version)."""
mock_response = httpx.Response(409, json={"detail": "Already on version"})
mock_pragma_client.downgrade_provider.side_effect = httpx.HTTPStatusError(
"Conflict", request=httpx.Request("POST", "http://test"), response=mock_response
)

result = cli_runner.invoke(app, ["providers", "downgrade", "qdrant", "--version", "1.0.0", "-y"])

assert result.exit_code == 1
assert "already on v1.0.0" in result.output


def test_downgrade_not_installed(cli_runner, mock_pragma_client):
"""Downgrade command handles 404 (provider not installed)."""
mock_response = httpx.Response(404, json={"detail": "Not found"})
mock_pragma_client.downgrade_provider.side_effect = httpx.HTTPStatusError(
"Not Found", request=httpx.Request("POST", "http://test"), response=mock_response
)

result = cli_runner.invoke(app, ["providers", "downgrade", "qdrant", "--version", "1.0.0", "-y"])

assert result.exit_code == 1
assert "not installed" in result.output


def test_downgrade_version_chain_broken(cli_runner, mock_pragma_client):
"""Downgrade command handles 422 (version chain broken)."""
mock_response = httpx.Response(422, json={"detail": "Missing intermediate version 1.5.0"})
mock_pragma_client.downgrade_provider.side_effect = httpx.HTTPStatusError(
"Unprocessable Entity", request=httpx.Request("POST", "http://test"), response=mock_response
)

result = cli_runner.invoke(app, ["providers", "downgrade", "qdrant", "--version", "1.0.0", "-y"])

assert result.exit_code == 1
assert "version chain" in result.output.lower()


# ---------------------------------------------------------------------------
# List command tests (store browsing)
# ---------------------------------------------------------------------------
Expand Down