sdk/python/tests/test_arv_keepdocker.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 import arvados
   5 import hashlib
   6 import io
   7 import mock
   8 import os
   9 import subprocess
  10 import sys
  11 import tempfile
  12 import unittest
  13 import logging
  14
  15 import arvados.commands.keepdocker as arv_keepdocker
  16 import arvados_testutil as tutil
  17 import run_test_server
  18
  19
  20 class StopTest(Exception):
  21     pass
  22
  23
  24 class ArvKeepdockerTestCase(unittest.TestCase):
  25     def run_arv_keepdocker(self, args, err):
  26         sys.argv = ['arv-keepdocker'] + args
  27         log_handler = logging.StreamHandler(err)
  28         arv_keepdocker.logger.addHandler(log_handler)
  29         try:
  30             return arv_keepdocker.main()
  31         finally:
  32             arv_keepdocker.logger.removeHandler(log_handler)
  33
  34     def test_unsupported_arg(self):
  35         with self.assertRaises(SystemExit):
  36             self.run_arv_keepdocker(['-x=unknown'], sys.stderr)
  37
  38     def test_version_argument(self):
  39         err = io.BytesIO()
  40         out = io.BytesIO()
  41         with tutil.redirected_streams(stdout=out, stderr=err):
  42             with self.assertRaises(SystemExit):
  43                 self.run_arv_keepdocker(['--version'], sys.stderr)
  44         self.assertEqual(out.getvalue(), '')
  45         self.assertRegexpMatches(err.getvalue(), "[0-9]+\.[0-9]+\.[0-9]+")
  46
  47     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
  48                 return_value=['abc123'])
  49     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
  50                 return_value='abc123')
  51     def test_image_format_compatibility(self, _1, _2):
  52         old_id = hashlib.sha256('old').hexdigest()
  53         new_id = 'sha256:'+hashlib.sha256('new').hexdigest()
  54         for supported, img_id, expect_ok in [
  55                 (['v1'], old_id, True),
  56                 (['v1'], new_id, False),
  57                 (None, old_id, False),
  58                 ([], old_id, False),
  59                 ([], new_id, False),
  60                 (['v1', 'v2'], new_id, True),
  61                 (['v1'], new_id, False),
  62                 (['v2'], new_id, True)]:
  63
  64             fakeDD = arvados.api('v1')._rootDesc
  65             if supported is None:
  66                 del fakeDD['dockerImageFormats']
  67             else:
  68                 fakeDD['dockerImageFormats'] = supported
  69
  70             err = io.BytesIO()
  71             out = io.BytesIO()
  72
  73             with tutil.redirected_streams(stdout=out), \
  74                  mock.patch('arvados.api') as api, \
  75                  mock.patch('arvados.commands.keepdocker.popen_docker',
  76                             return_value=subprocess.Popen(
  77                                 ['echo', img_id],
  78                                 stdout=subprocess.PIPE)), \
  79                  mock.patch('arvados.commands.keepdocker.prep_image_file',
  80                             side_effect=StopTest), \
  81                  self.assertRaises(StopTest if expect_ok else SystemExit):
  82
  83                 api()._rootDesc = fakeDD
  84                 self.run_arv_keepdocker(['--force', 'testimage'], err)
  85
  86             self.assertEqual(out.getvalue(), '')
  87             if expect_ok:
  88                 self.assertNotRegexpMatches(
  89                     err.getvalue(), "refusing to store",
  90                     msg=repr((supported, img_id)))
  91             else:
  92                 self.assertRegexpMatches(
  93                     err.getvalue(), "refusing to store",
  94                     msg=repr((supported, img_id)))
  95             if not supported:
  96                 self.assertRegexpMatches(
  97                     err.getvalue(),
  98                     "server does not specify supported image formats",
  99                     msg=repr((supported, img_id)))
 100
 101         fakeDD = arvados.api('v1')._rootDesc
 102         fakeDD['dockerImageFormats'] = ['v1']
 103         err = io.BytesIO()
 104         out = io.BytesIO()
 105         with tutil.redirected_streams(stdout=out), \
 106              mock.patch('arvados.api') as api, \
 107              mock.patch('arvados.commands.keepdocker.popen_docker',
 108                         return_value=subprocess.Popen(
 109                             ['echo', new_id],
 110                             stdout=subprocess.PIPE)), \
 111              mock.patch('arvados.commands.keepdocker.prep_image_file',
 112                         side_effect=StopTest), \
 113              self.assertRaises(StopTest):
 114             api()._rootDesc = fakeDD
 115             self.run_arv_keepdocker(
 116                 ['--force', '--force-image-format', 'testimage'], err)
 117         self.assertRegexpMatches(err.getvalue(), "forcing incompatible image")