Add chmod with octal mode support (#194)

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",

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"])