sdk/python/tests/test_cmd_util.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 import contextlib
   6 import copy
   7 import itertools
   8 import json
   9 import os
  10 import tempfile
  11 import unittest
  12
  13 from pathlib import Path
  14
  15 import pytest
  16 from parameterized import parameterized
  17
  18 import arvados.commands._util as cmd_util
  19
  20 FILE_PATH = Path(__file__)
  21
  22 class ValidateFiltersTestCase(unittest.TestCase):
  23     NON_FIELD_TYPES = [
  24         None,
  25         123,
  26         ('name', '=', 'tuple'),
  27         {'filters': ['name', '=', 'object']},
  28     ]
  29     NON_FILTER_TYPES = NON_FIELD_TYPES + ['string']
  30     VALID_FILTERS = [
  31         ['owner_uuid', '=', 'zzzzz-tpzed-12345abcde67890'],
  32         ['name', 'in', ['foo', 'bar']],
  33         '(replication_desired > replication_cofirmed)',
  34         '(replication_confirmed>=replication_desired)',
  35     ]
  36
  37     @parameterized.expand(itertools.combinations(VALID_FILTERS, 2))
  38     def test_valid_filters(self, f1, f2):
  39         expected = [f1, f2]
  40         actual = cmd_util.validate_filters(copy.deepcopy(expected))
  41         self.assertEqual(actual, expected)
  42
  43     @parameterized.expand([(t,) for t in NON_FILTER_TYPES])
  44     def test_filters_wrong_type(self, value):
  45         with self.assertRaisesRegex(ValueError, r'^filters are not a list\b'):
  46             cmd_util.validate_filters(value)
  47
  48     @parameterized.expand([(t,) for t in NON_FIELD_TYPES])
  49     def test_single_filter_wrong_type(self, value):
  50         with self.assertRaisesRegex(ValueError, r'^filter at index 0 is not a string or list\b'):
  51             cmd_util.validate_filters([value])
  52
  53     @parameterized.expand([
  54         ([],),
  55         (['owner_uuid'],),
  56         (['owner_uuid', 'zzzzz-tpzed-12345abcde67890'],),
  57         (['name', 'not in', 'foo', 'bar'],),
  58         (['name', 'in', 'foo', 'bar', 'baz'],),
  59     ])
  60     def test_filters_wrong_arity(self, value):
  61         with self.assertRaisesRegex(ValueError, r'^filter at index 0 does not have three items\b'):
  62             cmd_util.validate_filters([value])
  63
  64     @parameterized.expand(itertools.product(
  65         [0, 1],
  66         NON_FIELD_TYPES,
  67     ))
  68     def test_filter_definition_wrong_type(self, index, bad_value):
  69         value = ['owner_uuid', '=', 'zzzzz-tpzed-12345abcde67890']
  70         value[index] = bad_value
  71         name = ('field name', 'operator')[index]
  72         with self.assertRaisesRegex(ValueError, rf'^filter at index 0 {name} is not a string\b'):
  73             cmd_util.validate_filters([value])
  74
  75     @parameterized.expand([
  76         # Not enclosed in parentheses
  77         'foo = bar',
  78         '(foo) < bar',
  79         'foo > (bar)',
  80         # Not exactly one operator
  81         '(a >= b >= c)',
  82         '(foo)',
  83         '(file_count version)',
  84         # Invalid field identifiers
  85         '(version = 1)',
  86         '(2 = file_count)',
  87         '(replication.desired <= replication.confirmed)',
  88         # Invalid whitespace
  89         '(file_count\t=\tversion)',
  90         '(file_count >= version\n)',
  91     ])
  92     def test_invalid_string_filter(self, value):
  93         with self.assertRaisesRegex(ValueError, r'^filter at index 0 has invalid syntax\b'):
  94             cmd_util.validate_filters([value])
  95
  96
  97 class JSONArgumentTestCase(unittest.TestCase):
  98     JSON_OBJECTS = [
  99         None,
 100         123,
 101         456.789,
 102         'string',
 103         ['list', 1],
 104         {'object': True, 'yaml': False},
 105     ]
 106
 107     @classmethod
 108     def setUpClass(cls):
 109         cls.json_file = tempfile.NamedTemporaryFile(
 110             'w+',
 111             encoding='utf-8',
 112             prefix='argtest',
 113             suffix='.json',
 114         )
 115         cls.parser = cmd_util.JSONArgument()
 116
 117     @classmethod
 118     def tearDownClass(cls):
 119         cls.json_file.close()
 120
 121     def setUp(self):
 122         self.json_file.seek(0)
 123         self.json_file.truncate()
 124
 125     @parameterized.expand((obj,) for obj in JSON_OBJECTS)
 126     def test_valid_argument_string(self, obj):
 127         actual = self.parser(json.dumps(obj))
 128         self.assertEqual(actual, obj)
 129
 130     @parameterized.expand((obj,) for obj in JSON_OBJECTS)
 131     def test_valid_argument_path(self, obj):
 132         json.dump(obj, self.json_file)
 133         self.json_file.flush()
 134         actual = self.parser(self.json_file.name)
 135         self.assertEqual(actual, obj)
 136
 137     @parameterized.expand([
 138         '',
 139         '\0',
 140         None,
 141     ])
 142     def test_argument_not_json_or_path(self, value):
 143         if value is None:
 144             with tempfile.NamedTemporaryFile() as gone_file:
 145                 value = gone_file.name
 146         with self.assertRaisesRegex(ValueError, r'\bnot a valid JSON string or file path\b'):
 147             self.parser(value)
 148
 149     @parameterized.expand([
 150         FILE_PATH.parent,
 151         FILE_PATH / 'nonexistent.json',
 152         None,
 153     ])
 154     def test_argument_path_unreadable(self, path):
 155         if path is None:
 156             bad_file = tempfile.NamedTemporaryFile()
 157             os.chmod(bad_file.fileno(), 0o000)
 158             path = bad_file.name
 159             @contextlib.contextmanager
 160             def ctx():
 161                 try:
 162                     yield
 163                 finally:
 164                     os.chmod(bad_file.fileno(), 0o600)
 165         else:
 166             ctx = contextlib.nullcontext
 167         with self.assertRaisesRegex(ValueError, rf'^error reading JSON file path {str(path)!r}: '), ctx():
 168             self.parser(str(path))
 169
 170     @parameterized.expand([
 171         FILE_PATH,
 172         None,
 173     ])
 174     def test_argument_path_not_json(self, path):
 175         if path is None:
 176             path = self.json_file.name
 177         with self.assertRaisesRegex(ValueError, rf'^error decoding JSON from file {str(path)!r}'):
 178             self.parser(str(path))
 179
 180
 181 class JSONArgumentValidationTestCase(unittest.TestCase):
 182     @parameterized.expand((obj,) for obj in JSONArgumentTestCase.JSON_OBJECTS)
 183     def test_object_returned_from_validator(self, value):
 184         parser = cmd_util.JSONArgument(lambda _: copy.deepcopy(value))
 185         self.assertEqual(parser('{}'), value)
 186
 187     @parameterized.expand((obj,) for obj in JSONArgumentTestCase.JSON_OBJECTS)
 188     def test_exception_raised_from_validator(self, value):
 189         json_value = json.dumps(value)
 190         def raise_func(_):
 191             raise ValueError(json_value)
 192         parser = cmd_util.JSONArgument(raise_func)
 193         with self.assertRaises(ValueError) as exc_check:
 194             parser(json_value)
 195         self.assertEqual(exc_check.exception.args, (json_value,))
 196
 197
 198 class TestRangedValue:
 199     @pytest.fixture(scope='class')
 200     def cmpint(self):
 201         return cmd_util.RangedValue(int, range(-1, 2))
 202
 203     @pytest.mark.parametrize('s', ['-1', '0', '1'])
 204     def test_valid_values(self, cmpint, s):
 205         assert cmpint(s) == int(s)
 206
 207     @pytest.mark.parametrize('s', ['foo', '-2', '2', '0.2', '', ' '])
 208     def test_invalid_values(self, cmpint, s):
 209         with pytest.raises(ValueError):
 210             cmpint(s)
 211
 212
 213 class TestUniqueSplit:
 214     @pytest.fixture(scope='class')
 215     def argtype(self):
 216         return cmd_util.UniqueSplit()
 217
 218     @pytest.mark.parametrize('arg', [
 219         'foo',
 220         'foo,bar',
 221         'foo, bar, baz',
 222         'foo , bar , baz , quux',
 223     ])
 224     def test_basic_parse(self, arg, argtype):
 225         expected = ['foo', 'bar', 'baz', 'quux'][:arg.count(',') + 1]
 226         assert argtype(arg) == expected
 227
 228     @pytest.mark.parametrize('arg', [
 229         'foo, foo, bar',
 230         'foo, bar, foo',
 231         'foo, bar, bar',
 232     ])
 233     def test_uniqueness(self, arg, argtype):
 234         assert argtype(arg) == ['foo', 'bar']