Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(cli): add -r and -v flags for safer usage #21

Merged
merged 1 commit into from
Apr 24, 2020
Merged
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
31 changes: 23 additions & 8 deletions pathy/cli.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,5 @@
#!/usr/bin/env python3.7
from pathlib import Path

import typer
from pathy import Pathy, FluidPath
from .api import Pathy, FluidPath

app = typer.Typer()

Expand Down Expand Up @@ -68,20 +65,38 @@ def mv(from_location: str, to_location: str):


@app.command()
def rm(location: str, strict: bool = False):
def rm(
location: str,
recursive: bool = typer.Option(
False, "--recursive", "-r", help="Recursively remove files and folders."
),
verbose: bool = typer.Option(
False, "--verbose", "-v", help="Print removed files and folders."
),
):
"""
Remove a blob or folder of blobs from a given location.
"""
path: FluidPath = Pathy.fluid(location)
if not path.exists() and strict:
raise ValueError(f"from_path is not an existing Path or Pathy: {path}")
if not path.exists():
raise typer.Exit(f"rm: {path}: No such file or directory")

if path.is_dir():
to_unlink = [b for b in path.rglob("*") if b.is_file()]
if not recursive:
raise typer.Exit(f"rm: {path}: is a directory")
selector = path.rglob("*") if recursive else path.glob("*")
to_unlink = [b for b in selector if b.is_file()]
for blob in to_unlink:
if verbose:
typer.echo(str(blob))
blob.unlink()
if path.exists():
if verbose:
typer.echo(str(path))
path.rmdir()
elif path.is_file():
if verbose:
typer.echo(str(path))
path.unlink()


Expand Down
42 changes: 32 additions & 10 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,30 +98,52 @@ def test_cli_mv_folder_across_buckets(with_adapter, bucket: str, other_bucket: s
assert (destination / f"{i}" / f"{j}").is_file()


@pytest.mark.parametrize("adapter", TEST_ADAPTERS)
def test_cli_rm_file(with_adapter, bucket: str):
source = f"gs://{bucket}/cli_rm_file/file.txt"
Pathy(source).write_text("---")
assert Pathy(source).exists()
assert runner.invoke(app, ["rm", source]).exit_code == 0
assert not Pathy(source).exists()


@pytest.mark.parametrize("adapter", TEST_ADAPTERS)
def test_cli_rm_verbose(with_adapter, bucket: str):
root = Pathy.from_bucket(bucket) / "cli_rm_folder"
source = str(root / "file.txt")
other = str(root / "folder/other")
Pathy(source).write_text("---")
Pathy(other).write_text("---")
result = runner.invoke(app, ["rm", "-v", source])
assert result.exit_code == 0
assert source in result.output
assert other not in result.output

Pathy(source).write_text("---")
result = runner.invoke(app, ["rm", "-rv", str(root)])
assert result.exit_code == 0
assert source in result.output
assert other in result.output


@pytest.mark.parametrize("adapter", TEST_ADAPTERS)
def test_cli_rm_folder(with_adapter, bucket: str):
root = Pathy.from_bucket(bucket)
source = root / "cli_rm_folder"
for i in range(2):
for j in range(2):
(source / f"{i}" / f"{j}").write_text("---")
assert runner.invoke(app, ["rm", str(source)]).exit_code == 0

# Returns exit code 1 without recursive flag when given a folder
assert runner.invoke(app, ["rm", str(source)]).exit_code == 1
assert runner.invoke(app, ["rm", "-r", str(source)]).exit_code == 0
assert not Pathy(source).exists()
# Ensure source files are gone
for i in range(2):
for j in range(2):
assert not (source / f"{i}" / f"{j}").is_file()


@pytest.mark.parametrize("adapter", TEST_ADAPTERS)
def test_cli_rm_file(with_adapter, bucket: str):
source = f"gs://{bucket}/cli_rm_file/file.txt"
Pathy(source).write_text("---")
assert Pathy(source).exists()
assert runner.invoke(app, ["rm", source]).exit_code == 0
assert not Pathy(source).exists()


@pytest.mark.parametrize("adapter", TEST_ADAPTERS)
def test_cli_ls(with_adapter, bucket: str):
root = Pathy.from_bucket(bucket) / "cli_ls"
Expand Down