Skip to content

Commit

Permalink
feat(cli): add -r and -v flags for safer usage
Browse files Browse the repository at this point in the history
 - rm will fail if given a directory without the -r flag (similar to unix rm)
 - rm will print the removed files/folders when given the -v flag
  • Loading branch information
justindujardin committed Apr 24, 2020
1 parent 5a02f63 commit a87e36f
Show file tree
Hide file tree
Showing 2 changed files with 55 additions and 18 deletions.
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

0 comments on commit a87e36f

Please sign in to comment.