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 hashlib
   8 import mock
   9 import os
  10 import subprocess
  11 import sys
  12 import tempfile
  13 import unittest
  14 import logging
  15
  16 import arvados.commands.keepdocker as arv_keepdocker
  17 from . import arvados_testutil as tutil
  18 from . import run_test_server
  19
  20
  21 class StopTest(Exception):
  22     pass
  23
  24
  25 class ArvKeepdockerTestCase(unittest.TestCase, tutil.VersionChecker):
  26     def run_arv_keepdocker(self, args, err):
  27         sys.argv = ['arv-keepdocker'] + args
  28         log_handler = logging.StreamHandler(err)
  29         arv_keepdocker.logger.addHandler(log_handler)
  30         try:
  31             return arv_keepdocker.main()
  32         finally:
  33             arv_keepdocker.logger.removeHandler(log_handler)
  34
  35     def test_unsupported_arg(self):
  36         out = tutil.StringIO()
  37         with tutil.redirected_streams(stdout=out, stderr=out), \
  38              self.assertRaises(SystemExit):
  39             self.run_arv_keepdocker(['-x=unknown'], sys.stderr)
  40         self.assertRegex(out.getvalue(), 'unrecognized arguments')
  41
  42     def test_version_argument(self):
  43         with tutil.redirected_streams(
  44                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  45             with self.assertRaises(SystemExit):
  46                 self.run_arv_keepdocker(['--version'], sys.stderr)
  47         self.assertVersionOutput(out, err)
  48
  49     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
  50                 return_value=['abc123'])
  51     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
  52                 return_value='abc123')
  53     def test_image_format_compatibility(self, _1, _2):
  54         old_id = hashlib.sha256(b'old').hexdigest()
  55         new_id = 'sha256:'+hashlib.sha256(b'new').hexdigest()
  56         for supported, img_id, expect_ok in [
  57                 (['v1'], old_id, True),
  58                 (['v1'], new_id, False),
  59                 (None, old_id, False),
  60                 ([], old_id, False),
  61                 ([], new_id, False),
  62                 (['v1', 'v2'], new_id, True),
  63                 (['v1'], new_id, False),
  64                 (['v2'], new_id, True)]:
  65
  66             fakeDD = arvados.api('v1')._rootDesc
  67             if supported is None:
  68                 del fakeDD['dockerImageFormats']
  69             else:
  70                 fakeDD['dockerImageFormats'] = supported
  71
  72             err = tutil.StringIO()
  73             out = tutil.StringIO()
  74
  75             with tutil.redirected_streams(stdout=out), \
  76                  mock.patch('arvados.api') as api, \
  77                  mock.patch('arvados.commands.keepdocker.popen_docker',
  78                             return_value=subprocess.Popen(
  79                                 ['echo', img_id],
  80                                 stdout=subprocess.PIPE)), \
  81                  mock.patch('arvados.commands.keepdocker.prep_image_file',
  82                             side_effect=StopTest), \
  83                  self.assertRaises(StopTest if expect_ok else SystemExit):
  84
  85                 api()._rootDesc = fakeDD
  86                 self.run_arv_keepdocker(['--force', 'testimage'], err)
  87
  88             self.assertEqual(out.getvalue(), '')
  89             if expect_ok:
  90                 self.assertNotRegex(
  91                     err.getvalue(), "refusing to store",
  92                     msg=repr((supported, img_id)))
  93             else:
  94                 self.assertRegex(
  95                     err.getvalue(), "refusing to store",
  96                     msg=repr((supported, img_id)))
  97             if not supported:
  98                 self.assertRegex(
  99                     err.getvalue(),
 100                     "server does not specify supported image formats",
 101                     msg=repr((supported, img_id)))
 102
 103         fakeDD = arvados.api('v1')._rootDesc
 104         fakeDD['dockerImageFormats'] = ['v1']
 105         err = tutil.StringIO()
 106         out = tutil.StringIO()
 107         with tutil.redirected_streams(stdout=out), \
 108              mock.patch('arvados.api') as api, \
 109              mock.patch('arvados.commands.keepdocker.popen_docker',
 110                         return_value=subprocess.Popen(
 111                             ['echo', new_id],
 112                             stdout=subprocess.PIPE)), \
 113              mock.patch('arvados.commands.keepdocker.prep_image_file',
 114                         side_effect=StopTest), \
 115              self.assertRaises(StopTest):
 116             api()._rootDesc = fakeDD
 117             self.run_arv_keepdocker(
 118                 ['--force', '--force-image-format', 'testimage'], err)
 119         self.assertRegex(err.getvalue(), "forcing incompatible image")
 120
 121     def test_tag_given_twice(self):
 122         with tutil.redirected_streams(stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
 123             with self.assertRaises(SystemExit):
 124                 self.run_arv_keepdocker(['myrepo:mytag', 'extratag'], sys.stderr)
 125             self.assertRegex(err.getvalue(), "cannot add tag argument 'extratag'")
 126
 127     def test_image_given_as_repo_colon_tag(self):
 128         with self.assertRaises(StopTest), \
 129              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 130                         side_effect=StopTest) as find_image_mock:
 131             self.run_arv_keepdocker(['repo:tag'], sys.stderr)
 132         find_image_mock.assert_called_with('repo', 'tag')
 133
 134         with self.assertRaises(StopTest), \
 135              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 136                         side_effect=StopTest) as find_image_mock:
 137             self.run_arv_keepdocker(['myreg.example:8888/repo/img:tag'], sys.stderr)
 138         find_image_mock.assert_called_with('myreg.example:8888/repo/img', 'tag')
 139
 140     def test_image_has_colons(self):
 141         with self.assertRaises(StopTest), \
 142              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 143                         side_effect=StopTest) as find_image_mock:
 144             self.run_arv_keepdocker(['[::1]:8888/repo/img'], sys.stderr)
 145         find_image_mock.assert_called_with('[::1]:8888/repo/img', 'latest')
 146
 147         with self.assertRaises(StopTest), \
 148              mock.patch('arvados.commands.keepdocker.find_one_image_hash',
 149                         side_effect=StopTest) as find_image_mock:
 150             self.run_arv_keepdocker(['[::1]/repo/img'], sys.stderr)
 151         find_image_mock.assert_called_with('[::1]/repo/img', 'latest')