sdk/python/tests/test_arv_keepdocker.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 from __future__ import absolute_import
   5 import arvados
   6 import hashlib
   7 import io
   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):
  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         with self.assertRaises(SystemExit):
  37             self.run_arv_keepdocker(['-x=unknown'], sys.stderr)
  38
  39     def test_version_argument(self):
  40         err = io.BytesIO()
  41         out = io.BytesIO()
  42         with tutil.redirected_streams(stdout=out, stderr=err):
  43             with self.assertRaises(SystemExit):
  44                 self.run_arv_keepdocker(['--version'], sys.stderr)
  45         self.assertEqual(out.getvalue(), '')
  46         self.assertRegexpMatches(err.getvalue(), "[0-9]+\.[0-9]+\.[0-9]+")
  47
  48     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
  49                 return_value=['abc123'])
  50     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
  51                 return_value='abc123')
  52     def test_image_format_compatibility(self, _1, _2):
  53         old_id = hashlib.sha256('old').hexdigest()
  54         new_id = 'sha256:'+hashlib.sha256('new').hexdigest()
  55         for supported, img_id, expect_ok in [
  56                 (['v1'], old_id, True),
  57                 (['v1'], new_id, False),
  58                 (None, old_id, False),
  59                 ([], old_id, False),
  60                 ([], new_id, False),
  61                 (['v1', 'v2'], new_id, True),
  62                 (['v1'], new_id, False),
  63                 (['v2'], new_id, True)]:
  64
  65             fakeDD = arvados.api('v1')._rootDesc
  66             if supported is None:
  67                 del fakeDD['dockerImageFormats']
  68             else:
  69                 fakeDD['dockerImageFormats'] = supported
  70
  71             err = io.BytesIO()
  72             out = io.BytesIO()
  73
  74             with tutil.redirected_streams(stdout=out), \
  75                  mock.patch('arvados.api') as api, \
  76                  mock.patch('arvados.commands.keepdocker.popen_docker',
  77                             return_value=subprocess.Popen(
  78                                 ['echo', img_id],
  79                                 stdout=subprocess.PIPE)), \
  80                  mock.patch('arvados.commands.keepdocker.prep_image_file',
  81                             side_effect=StopTest), \
  82                  self.assertRaises(StopTest if expect_ok else SystemExit):
  83
  84                 api()._rootDesc = fakeDD
  85                 self.run_arv_keepdocker(['--force', 'testimage'], err)
  86
  87             self.assertEqual(out.getvalue(), '')
  88             if expect_ok:
  89                 self.assertNotRegexpMatches(
  90                     err.getvalue(), "refusing to store",
  91                     msg=repr((supported, img_id)))
  92             else:
  93                 self.assertRegexpMatches(
  94                     err.getvalue(), "refusing to store",
  95                     msg=repr((supported, img_id)))
  96             if not supported:
  97                 self.assertRegexpMatches(
  98                     err.getvalue(),
  99                     "server does not specify supported image formats",
 100                     msg=repr((supported, img_id)))
 101
 102         fakeDD = arvados.api('v1')._rootDesc
 103         fakeDD['dockerImageFormats'] = ['v1']
 104         err = io.BytesIO()
 105         out = io.BytesIO()
 106         with tutil.redirected_streams(stdout=out), \
 107              mock.patch('arvados.api') as api, \
 108              mock.patch('arvados.commands.keepdocker.popen_docker',
 109                         return_value=subprocess.Popen(
 110                             ['echo', new_id],
 111                             stdout=subprocess.PIPE)), \
 112              mock.patch('arvados.commands.keepdocker.prep_image_file',
 113                         side_effect=StopTest), \
 114              self.assertRaises(StopTest):
 115             api()._rootDesc = fakeDD
 116             self.run_arv_keepdocker(
 117                 ['--force', '--force-image-format', 'testimage'], err)
 118         self.assertRegexpMatches(err.getvalue(), "forcing incompatible image")