diff --git a/python/hdfs_native/cli.py b/python/hdfs_native/cli.py index f3478fe..d7158ef 100644 --- a/python/hdfs_native/cli.py +++ b/python/hdfs_native/cli.py @@ -1,5 +1,6 @@ import functools import os +import re import sys from argparse import ArgumentParser, Namespace from typing import List, Optional, Sequence @@ -60,6 +61,22 @@ def cat(args: Namespace): sys.stdout.buffer.flush() +def chmod(args: Namespace): + if not re.fullmatch(r"1?[0-7]{3}", args.octalmode): + raise ValueError(f"Invalid mode supplied: {args.octalmode}") + + permission = int(args.octalmode, base=8) + + for url in args.path: + client = _client_for_url(url) + for path in _glob_path(client, _path_for_url(url)): + if args.recursive: + for status in client.list_status(path, True): + client.set_permission(status.path, permission) + else: + client.set_permission(path, permission) + + def chown(args: Namespace): split = args.owner.split(":") if len(split) > 2: @@ -139,6 +156,24 @@ def main(in_args: Optional[Sequence[str]] = None): cat_parser.add_argument("src", nargs="+", help="File pattern to print") cat_parser.set_defaults(func=cat) + chmod_parser = subparsers.add_parser( + "chmod", + help="Changes permissions of a file", + description="Changes permissions of a file. Only octal permissions are supported.", + ) + chmod_parser.add_argument( + "-R", + "--recursive", + action="store_true", + help="Modify files recursively", + ) + chmod_parser.add_argument( + "octalmode", + help="The mode to set the permission to in octal format", + ) + chmod_parser.add_argument("path", nargs="+", help="File pattern to update") + chmod_parser.set_defaults(func=chmod) + chown_parser = subparsers.add_parser( "chown", help="Changes owner and group of a file", diff --git a/python/tests/test_cli.py b/python/tests/test_cli.py index 14a48fc..6de04a9 100644 --- a/python/tests/test_cli.py +++ b/python/tests/test_cli.py @@ -31,6 +31,44 @@ def test_cat(client: Client): client.delete("/testfile2") +def test_chmod(client: Client): + with pytest.raises(FileNotFoundError): + cli_main(["chmod", "755", "/testfile"]) + + client.create("/testfile").close() + + cli_main(["chmod", "700", "/testfile"]) + assert client.get_file_info("/testfile").permission == 0o700 + + cli_main(["chmod", "007", "/testfile"]) + assert client.get_file_info("/testfile").permission == 0o007 + + cli_main(["chmod", "1777", "/testfile"]) + assert client.get_file_info("/testfile").permission == 0o1777 + + with pytest.raises(ValueError): + cli_main(["chmod", "2777", "/testfile"]) + + with pytest.raises(ValueError): + cli_main(["chmod", "2778", "/testfile"]) + + client.delete("/testfile") + + client.mkdirs("/testdir") + client.create("/testdir/testfile").close() + original_permission = client.get_file_info("/testdir/testfile").permission + + cli_main(["chmod", "700", "/testdir"]) + assert client.get_file_info("/testdir").permission == 0o700 + assert client.get_file_info("/testdir/testfile").permission == original_permission + + cli_main(["chmod", "-R", "700", "/testdir"]) + assert client.get_file_info("/testdir").permission == 0o700 + assert client.get_file_info("/testdir/testfile").permission == 0o700 + + client.delete("/testdir", True) + + def test_chown(client: Client): with pytest.raises(FileNotFoundError): cli_main(["chown", "testuser", "/testfile"])