sdk/python/tests/test_arv_keepdocker.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 from __future__ import absolute_import
   6 import arvados
   7 import collections
   8 import copy
   9 import hashlib
  10 import mock
  11 import os
  12 import subprocess
  13 import sys
  14 import tempfile
  15 import unittest
  16 import logging
  17
  18 import arvados.commands.keepdocker as arv_keepdocker
  19 from . import arvados_testutil as tutil
  20 from . import run_test_server
  21
  22
  23 class StopTest(Exception):
  24     pass
  25
  26
  27 class ArvKeepdockerTestCase(unittest.TestCase, tutil.VersionChecker):
  28     def run_arv_keepdocker(self, args, err):
  29         sys.argv = ['arv-keepdocker'] + args
  30         log_handler = logging.StreamHandler(err)
  31         arv_keepdocker.logger.addHandler(log_handler)
  32         try:
  33             return arv_keepdocker.main()
  34         finally:
  35             arv_keepdocker.logger.removeHandler(log_handler)
  36
  37     def test_unsupported_arg(self):
  38         out = tutil.StringIO()
  39         with tutil.redirected_streams(stdout=out, stderr=out), \
  40              self.assertRaises(SystemExit):
  41             self.run_arv_keepdocker(['-x=unknown'], sys.stderr)
  42         self.assertRegex(out.getvalue(), 'unrecognized arguments')
  43
  44     def test_version_argument(self):
  45         with tutil.redirected_streams(
  46                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  47             with self.assertRaises(SystemExit):
  48                 self.run_arv_keepdocker(['--version'], sys.stderr)
  49         self.assertVersionOutput(out, err)
  50
  51     @mock.patch('arvados.commands.keepdocker.list_images_in_arv',
  52                 return_value=[])
  53     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
  54                 return_value=['abc123'])
  55     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
  56                 return_value='abc123')
  57     def test_image_format_compatibility(self, _1, _2, _3):
  58         old_id = hashlib.sha256(b'old').hexdigest()
  59         new_id = 'sha256:'+hashlib.sha256(b'new').hexdigest()
  60         for supported, img_id, expect_ok in [
  61                 (['v1'], old_id, True),
  62                 (['v1'], new_id, False),
  63                 (None, old_id, False),
  64                 ([], old_id, False),
  65                 ([], new_id, False),
  66                 (['v1', 'v2'], new_id, True),
  67                 (['v1'], new_id, False),
  68                 (['v2'], new_id, True)]:
  69
  70             fakeDD = arvados.api('v1')._rootDesc
  71             if supported is None:
  72                 del fakeDD['dockerImageFormats']
  73             else:
  74                 fakeDD['dockerImageFormats'] = supported
  75
  76             err = tutil.StringIO()
  77             out = tutil.StringIO()
  78
  79             with tutil.redirected_streams(stdout=out), \
  80                  mock.patch('arvados.api') as api, \
  81                  mock.patch('arvados.commands.keepdocker.popen_docker',
  82                             return_value=subprocess.Popen(
  83                                 ['echo', img_id],
  84                                 stdout=subprocess.PIPE)), \
  85                  mock.patch('arvados.commands.keepdocker.prep_image_file',
  86                             side_effect=StopTest), \
  87                  self.assertRaises(StopTest if expect_ok else SystemExit):
  88
  89                 api()._rootDesc = fakeDD
  90                 self.run_arv_keepdocker(['--force', 'testimage'], err)
  91
  92             self.assertEqual(out.getvalue(), '')
  93             if expect_ok:
  94                 self.assertNotRegex(
  95                     err.getvalue(), "refusing to store",
  96                     msg=repr((supported, img_id)))
  97             else:
  98                 self.assertRegex(
  99                     err.getvalue(), "refusing to store",
 100                     msg=repr((supported, img_id)))
 101             if not supported:
 102                 self.assertRegex(
 103                     err.getvalue(),
 104                     "server does not specify supported image formats",
 105                     msg=repr((supported, img_id)))
 106
 107         fakeDD = arvados.api('v1')._rootDesc
 108         fakeDD['dockerImageFormats'] = ['v1']
 109         err = tutil.StringIO()
 110         out = tutil.StringIO()
 111         with tutil.redirected_streams(stdout=out), \
 112              mock.patch('arvados.api') as api, \
 113              mock.patch('arvados.commands.keepdocker.popen_docker',
 114                         return_value=subprocess.Popen(
 115                             ['echo', new_id],
 116                             stdout=subprocess.PIPE)), \
 117              mock.patch('arvados.commands.keepdocker.prep_image_file',
 118                         side_effect=StopTest), \
 119              self.assertRaises(StopTest):
 120             api()._rootDesc = fakeDD
 121             self.run_arv_keepdocker(
 122                 ['--force', '--force-image-format', 'testimage'], err)
 123         self.assertRegex(err.getvalue(), "forcing incompatible image")
 124
 125     def test_tag_given_twice(self):
 126         with tutil.redirected_streams(stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
 127             with self.assertRaises(SystemExit):
 128                 self.run_arv_keepdocker(['myrepo:mytag', 'extratag'], sys.stderr)
 129             self.assertRegex(err.getvalue(), "cannot add tag argument 'extratag'")
 130
 131     def test_image_given_as_repo_colon_tag(self):
 132         with self.assertRaises(StopTest), \
 133              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 134                         side_effect=StopTest) as find_image_mock:
 135             self.run_arv_keepdocker(['repo:tag'], sys.stderr)
 136         find_image_mock.assert_called_with('repo', 'tag')
 137
 138         with self.assertRaises(StopTest), \
 139              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 140                         side_effect=StopTest) as find_image_mock:
 141             self.run_arv_keepdocker(['myreg.example:8888/repo/img:tag'], sys.stderr)
 142         find_image_mock.assert_called_with('myreg.example:8888/repo/img', 'tag')
 143
 144     def test_image_has_colons(self):
 145         with self.assertRaises(StopTest), \
 146              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 147                         side_effect=StopTest) as find_image_mock:
 148             self.run_arv_keepdocker(['[::1]:8888/repo/img'], sys.stderr)
 149         find_image_mock.assert_called_with('[::1]:8888/repo/img', 'latest')
 150
 151         with self.assertRaises(StopTest), \
 152              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 153                         side_effect=StopTest) as find_image_mock:
 154             self.run_arv_keepdocker(['[::1]/repo/img'], sys.stderr)
 155         find_image_mock.assert_called_with('[::1]/repo/img', 'latest')
 156
 157     @mock.patch('arvados.commands.keepdocker.list_images_in_arv',
 158                 return_value=[])
 159     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
 160                 return_value=['abc123'])
 161     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 162                 return_value='abc123')
 163     def test_collection_property_update(self, _1, _2, _3):
 164         image_id = 'sha256:'+hashlib.sha256(b'image').hexdigest()
 165         fakeDD = arvados.api('v1')._rootDesc
 166         fakeDD['dockerImageFormats'] = ['v2']
 167
 168         err = tutil.StringIO()
 169         out = tutil.StringIO()
 170         File = collections.namedtuple('File', ['name'])
 171         mocked_file = File(name='docker_image')
 172         mocked_collection = {
 173             'uuid': 'new-collection-uuid',
 174             'properties': {
 175                 'responsible_person_uuid': 'person_uuid',
 176             }
 177         }
 178
 179         with tutil.redirected_streams(stdout=out), \
 180              mock.patch('arvados.api') as api, \
 181              mock.patch('arvados.commands.keepdocker.popen_docker',
 182                         return_value=subprocess.Popen(
 183                             ['echo', image_id],
 184                             stdout=subprocess.PIPE)), \
 185              mock.patch('arvados.commands.keepdocker.prep_image_file',
 186                         return_value=(mocked_file, False)), \
 187              mock.patch('arvados.commands.put.main',
 188                         return_value='new-collection-uuid'), \
 189              self.assertRaises(StopTest):
 190
 191             api()._rootDesc = fakeDD
 192             api().collections().get().execute.return_value = copy.deepcopy(mocked_collection)
 193             api().collections().update().execute.side_effect = StopTest
 194             self.run_arv_keepdocker(['--force', 'testimage'], err)
 195
 196         updated_properties = mocked_collection['properties']
 197         updated_properties.update({'docker-image-repo-tag': 'testimage:latest'})
 198         api().collections().update.assert_called_with(
 199             uuid=mocked_collection['uuid'],
 200             body={'properties': updated_properties})