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.find_image_hashes',
  52                 return_value=['abc123'])
  53     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
  54                 return_value='abc123')
  55     def test_image_format_compatibility(self, _1, _2):
  56         old_id = hashlib.sha256(b'old').hexdigest()
  57         new_id = 'sha256:'+hashlib.sha256(b'new').hexdigest()
  58         for supported, img_id, expect_ok in [
  59                 (['v1'], old_id, True),
  60                 (['v1'], new_id, False),
  61                 (None, old_id, False),
  62                 ([], old_id, False),
  63                 ([], new_id, False),
  64                 (['v1', 'v2'], new_id, True),
  65                 (['v1'], new_id, False),
  66                 (['v2'], new_id, True)]:
  67
  68             fakeDD = arvados.api('v1')._rootDesc
  69             if supported is None:
  70                 del fakeDD['dockerImageFormats']
  71             else:
  72                 fakeDD['dockerImageFormats'] = supported
  73
  74             err = tutil.StringIO()
  75             out = tutil.StringIO()
  76
  77             with tutil.redirected_streams(stdout=out), \
  78                  mock.patch('arvados.api') as api, \
  79                  mock.patch('arvados.commands.keepdocker.popen_docker',
  80                             return_value=subprocess.Popen(
  81                                 ['echo', img_id],
  82                                 stdout=subprocess.PIPE)), \
  83                  mock.patch('arvados.commands.keepdocker.prep_image_file',
  84                             side_effect=StopTest), \
  85                  self.assertRaises(StopTest if expect_ok else SystemExit):
  86
  87                 api()._rootDesc = fakeDD
  88                 self.run_arv_keepdocker(['--force', 'testimage'], err)
  89
  90             self.assertEqual(out.getvalue(), '')
  91             if expect_ok:
  92                 self.assertNotRegex(
  93                     err.getvalue(), "refusing to store",
  94                     msg=repr((supported, img_id)))
  95             else:
  96                 self.assertRegex(
  97                     err.getvalue(), "refusing to store",
  98                     msg=repr((supported, img_id)))
  99             if not supported:
 100                 self.assertRegex(
 101                     err.getvalue(),
 102                     "server does not specify supported image formats",
 103                     msg=repr((supported, img_id)))
 104
 105         fakeDD = arvados.api('v1')._rootDesc
 106         fakeDD['dockerImageFormats'] = ['v1']
 107         err = tutil.StringIO()
 108         out = tutil.StringIO()
 109         with tutil.redirected_streams(stdout=out), \
 110              mock.patch('arvados.api') as api, \
 111              mock.patch('arvados.commands.keepdocker.popen_docker',
 112                         return_value=subprocess.Popen(
 113                             ['echo', new_id],
 114                             stdout=subprocess.PIPE)), \
 115              mock.patch('arvados.commands.keepdocker.prep_image_file',
 116                         side_effect=StopTest), \
 117              self.assertRaises(StopTest):
 118             api()._rootDesc = fakeDD
 119             self.run_arv_keepdocker(
 120                 ['--force', '--force-image-format', 'testimage'], err)
 121         self.assertRegex(err.getvalue(), "forcing incompatible image")
 122
 123     def test_tag_given_twice(self):
 124         with tutil.redirected_streams(stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
 125             with self.assertRaises(SystemExit):
 126                 self.run_arv_keepdocker(['myrepo:mytag', 'extratag'], sys.stderr)
 127             self.assertRegex(err.getvalue(), "cannot add tag argument 'extratag'")
 128
 129     def test_image_given_as_repo_colon_tag(self):
 130         with self.assertRaises(StopTest), \
 131              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 132                         side_effect=StopTest) as find_image_mock:
 133             self.run_arv_keepdocker(['repo:tag'], sys.stderr)
 134         find_image_mock.assert_called_with('repo', 'tag')
 135
 136         with self.assertRaises(StopTest), \
 137              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 138                         side_effect=StopTest) as find_image_mock:
 139             self.run_arv_keepdocker(['myreg.example:8888/repo/img:tag'], sys.stderr)
 140         find_image_mock.assert_called_with('myreg.example:8888/repo/img', 'tag')
 141
 142     def test_image_has_colons(self):
 143         with self.assertRaises(StopTest), \
 144              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 145                         side_effect=StopTest) as find_image_mock:
 146             self.run_arv_keepdocker(['[::1]:8888/repo/img'], sys.stderr)
 147         find_image_mock.assert_called_with('[::1]:8888/repo/img', 'latest')
 148
 149         with self.assertRaises(StopTest), \
 150              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 151                         side_effect=StopTest) as find_image_mock:
 152             self.run_arv_keepdocker(['[::1]/repo/img'], sys.stderr)
 153         find_image_mock.assert_called_with('[::1]/repo/img', 'latest')
 154
 155     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
 156                 return_value=['abc123'])
 157     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 158                 return_value='abc123')
 159     def test_collection_property_update(self, _1, _2):
 160         image_id = 'sha256:'+hashlib.sha256(b'image').hexdigest()
 161         fakeDD = arvados.api('v1')._rootDesc
 162         fakeDD['dockerImageFormats'] = ['v2']
 163
 164         err = tutil.StringIO()
 165         out = tutil.StringIO()
 166         File = collections.namedtuple('File', ['name'])
 167         mocked_file = File(name='docker_image')
 168         mocked_collection = {
 169             'uuid': 'new-collection-uuid',
 170             'properties': {
 171                 'responsible_person_uuid': 'person_uuid',
 172             }
 173         }
 174
 175         with tutil.redirected_streams(stdout=out), \
 176              mock.patch('arvados.api') as api, \
 177              mock.patch('arvados.commands.keepdocker.popen_docker',
 178                         return_value=subprocess.Popen(
 179                             ['echo', image_id],
 180                             stdout=subprocess.PIPE)), \
 181              mock.patch('arvados.commands.keepdocker.prep_image_file',
 182                         return_value=(mocked_file, False)), \
 183              mock.patch('arvados.commands.put.main',
 184                         return_value='new-collection-uuid'), \
 185              self.assertRaises(StopTest):
 186
 187             api()._rootDesc = fakeDD
 188             api().collections().get().execute.return_value = copy.deepcopy(mocked_collection)
 189             api().collections().update().execute.side_effect = StopTest
 190             self.run_arv_keepdocker(['--force', 'testimage'], err)
 191
 192         updated_properties = mocked_collection['properties']
 193         updated_properties.update({'docker-image-repo-tag': 'testimage:latest'})
 194         api().collections().update.assert_called_with(
 195             uuid=mocked_collection['uuid'],
 196             body={'properties': updated_properties})