21935: Add a docstring to arvados.commands

[arvados.git] / sdk / python / tests / test_util.py

diff --git a/sdk/python/tests/test_util.py b/sdk/python/tests/test_util.py
index 75d4a89e30ea77ee061908f601377d24ba74201a..874db86e35fc43be50aa33cad666763f46fe86ac 100644 (file)
--- a/sdk/python/tests/test_util.py
+++ b/sdk/python/tests/test_util.py
@@ -4,57 +4,27 @@
 
 import itertools
 import os
-import parameterized
 import subprocess
 import unittest
 
+import parameterized
 from unittest import mock
 
 import arvados
 import arvados.util
 
-class MkdirDashPTest(unittest.TestCase):
-    def setUp(self):
-        try:
-            os.path.mkdir('./tmp')
-        except:
-            pass
-    def tearDown(self):
-        try:
-            os.unlink('./tmp/bar')
-            os.rmdir('./tmp/foo')
-            os.rmdir('./tmp')
-        except:
-            pass
-    def runTest(self):
-        arvados.util.mkdir_dash_p('./tmp/foo')
-        with open('./tmp/bar', 'wb') as f:
-            f.write(b'bar')
-        self.assertRaises(OSError, arvados.util.mkdir_dash_p, './tmp/bar')
-
-
-class RunCommandTestCase(unittest.TestCase):
-    def test_success(self):
-        stdout, stderr = arvados.util.run_command(['echo', 'test'],
-                                                  stderr=subprocess.PIPE)
-        self.assertEqual("test\n".encode(), stdout)
-        self.assertEqual("".encode(), stderr)
-
-    def test_failure(self):
-        with self.assertRaises(arvados.errors.CommandFailedError):
-            arvados.util.run_command(['false'])
-
 class KeysetTestHelper:
-    def __init__(self, expect):
+    def __init__(self, expect, expect_num_retries=0):
         self.n = 0
         self.expect = expect
+        self.expect_num_retries = expect_num_retries
 
     def fn(self, **kwargs):
-        if self.expect[self.n][0] != kwargs:
-            raise Exception("Didn't match %s != %s" % (self.expect[self.n][0], kwargs))
+        assert kwargs == self.expect[self.n][0]
         return self
 
     def execute(self, num_retries):
+        assert num_retries == self.expect_num_retries
         self.n += 1
         return self.expect[self.n-1][1]
 
@@ -64,6 +34,12 @@ _SELECT_FAKE_ITEM = {
     'created_at': '2023-08-28T12:34:56.123456Z',
 }
 
+_FAKE_COMPUTED_PERMISSIONS_ITEM = {
+    'user_uuid': 'zzzzz-zyyyz-zzzzzyyyyywwwww',
+    'target_uuid': 'zzzzz-ttttt-xxxxxyyyyyzzzzz',
+    'perm_level': 'can_write',
+}
+
 class KeysetListAllTestCase(unittest.TestCase):
     def test_empty(self):
         ks = KeysetTestHelper([[
@@ -185,20 +161,24 @@ class KeysetListAllTestCase(unittest.TestCase):
         ls = list(arvados.util.keyset_list_all(ks.fn, ascending=False))
         self.assertEqual(ls, [{"created_at": "2", "uuid": "2"}, {"created_at": "1", "uuid": "1"}])
 
-    @parameterized.parameterized.expand(zip(
-        itertools.cycle(_SELECT_FAKE_ITEM),
-        itertools.chain.from_iterable(
-            itertools.combinations(_SELECT_FAKE_ITEM, count)
-            for count in range(len(_SELECT_FAKE_ITEM) + 1)
-        ),
-    ))
-    def test_select(self, order_key, select):
+    @parameterized.parameterized.expand(
+        (fake_item, key_fields, order_key, select)
+        for (fake_item, key_fields) in [
+            (_SELECT_FAKE_ITEM, ('uuid',)),
+            (_FAKE_COMPUTED_PERMISSIONS_ITEM, ('user_uuid', 'target_uuid')),
+        ]
+        for order_key in fake_item
+        if order_key != 'perm_level'
+        for count in range(len(fake_item) + 1)
+        for select in itertools.combinations(fake_item, count)
+    )
+    def test_select(self, fake_item, key_fields, order_key, select):
         # keyset_list_all must have both uuid and order_key to function.
         # Test that it selects those fields along with user-specified ones.
-        expect_select = {'uuid', order_key, *select}
+        expect_select = {*key_fields, order_key, *select}
         item = {
             key: value
-            for key, value in _SELECT_FAKE_ITEM.items()
+            for key, value in fake_item.items()
             if key in expect_select
         }
         list_func = mock.Mock()
@@ -210,7 +190,7 @@ class KeysetListAllTestCase(unittest.TestCase):
             ],
         )
         list_func.reset_mock()
-        actual = list(arvados.util.keyset_list_all(list_func, order_key, select=list(select)))
+        actual = list(arvados.util.keyset_list_all(list_func, order_key, select=list(select), key_fields=key_fields))
         self.assertEqual(actual, [item])
         calls = list_func.call_args_list
         self.assertTrue(len(calls) >= 2, "list_func() not called enough to exhaust items")