# -*- encoding: utf-8 -*-
#
# Copyright 2014 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

"""Tests for ImageCache class and helper functions."""

import os
import tempfile
import time
import uuid

import mock
from oslo_utils import uuidutils

from ironic.common import exception
from ironic.common import image_service
from ironic.common import images
from ironic.common import utils
from ironic.drivers.modules import image_cache
from ironic.tests import base


def touch(filename):
    open(filename, 'w').close()


@mock.patch.object(image_cache, '_fetch')
class TestImageCacheFetch(base.TestCase):

    def setUp(self):
        super(TestImageCacheFetch, self).setUp()
        self.master_dir = tempfile.mkdtemp()
        self.cache = image_cache.ImageCache(self.master_dir, None, None)
        self.dest_dir = tempfile.mkdtemp()
        self.dest_path = os.path.join(self.dest_dir, 'dest')
        self.uuid = uuidutils.generate_uuid()
        self.master_path = os.path.join(self.master_dir, self.uuid)

    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image_no_master_dir(self, mock_download, mock_clean_up,
                                       mock_fetch):
        self.cache.master_dir = None
        self.cache.fetch_image(self.uuid, self.dest_path)
        self.assertFalse(mock_download.called)
        mock_fetch.assert_called_once_with(
            None, self.uuid, self.dest_path, None, True)
        self.assertFalse(mock_clean_up.called)

    @mock.patch.object(os, 'unlink')
    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image_dest_and_master_exist_uptodate(self, mock_download,
            mock_clean_up, mock_unlink, mock_fetch):
        touch(self.master_path)
        os.link(self.master_path, self.dest_path)
        self.cache.fetch_image(self.uuid, self.dest_path)
        self.assertFalse(mock_unlink.called)
        self.assertFalse(mock_download.called)
        self.assertFalse(mock_fetch.called)
        self.assertFalse(mock_clean_up.called)

    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image_dest_and_master_exist_outdated(self, mock_download,
            mock_clean_up, mock_fetch):
        touch(self.master_path)
        touch(self.dest_path)
        self.assertNotEqual(os.stat(self.dest_path).st_ino,
                            os.stat(self.master_path).st_ino)
        self.cache.fetch_image(self.uuid, self.dest_path)
        self.assertFalse(mock_download.called)
        self.assertFalse(mock_fetch.called)
        self.assertTrue(os.path.isfile(self.dest_path))
        self.assertEqual(os.stat(self.dest_path).st_ino,
                         os.stat(self.master_path).st_ino)
        self.assertFalse(mock_clean_up.called)

    @mock.patch.object(os, 'unlink')
    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image_only_dest_exists(self, mock_download,
            mock_clean_up, mock_unlink, mock_fetch):
        touch(self.dest_path)
        self.cache.fetch_image(self.uuid, self.dest_path)
        mock_unlink.assert_called_once_with(self.dest_path)
        self.assertFalse(mock_fetch.called)
        mock_download.assert_called_once_with(
            self.uuid, self.master_path, self.dest_path,
            ctx=None, force_raw=True)
        self.assertTrue(mock_clean_up.called)

    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image_master_exists(self, mock_download, mock_clean_up,
                                       mock_fetch):
        touch(self.master_path)
        self.cache.fetch_image(self.uuid, self.dest_path)
        self.assertFalse(mock_download.called)
        self.assertFalse(mock_fetch.called)
        self.assertTrue(os.path.isfile(self.dest_path))
        self.assertEqual(os.stat(self.dest_path).st_ino,
                         os.stat(self.master_path).st_ino)
        self.assertFalse(mock_clean_up.called)

    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image(self, mock_download, mock_clean_up,
                         mock_fetch):
        self.cache.fetch_image(self.uuid, self.dest_path)
        self.assertFalse(mock_fetch.called)
        mock_download.assert_called_once_with(
            self.uuid, self.master_path, self.dest_path,
            ctx=None, force_raw=True)
        self.assertTrue(mock_clean_up.called)

    @mock.patch.object(image_cache.ImageCache, 'clean_up')
    @mock.patch.object(image_cache.ImageCache, '_download_image')
    def test_fetch_image_not_uuid(self, mock_download, mock_clean_up,
                                  mock_fetch):
        href = u'http://abc.com/ubuntu.qcow2'
        href_converted = str(uuid.uuid5(uuid.NAMESPACE_URL,
                                        href.encode('utf-8')))
        master_path = os.path.join(self.master_dir, href_converted)
        self.cache.fetch_image(href, self.dest_path)
        self.assertFalse(mock_fetch.called)
        mock_download.assert_called_once_with(
            href, master_path, self.dest_path,
            ctx=None, force_raw=True)
        self.assertTrue(mock_clean_up.called)

    def test__download_image(self, mock_fetch):
        def _fake_fetch(ctx, uuid, tmp_path, *args):
            self.assertEqual(self.uuid, uuid)
            self.assertNotEqual(self.dest_path, tmp_path)
            self.assertNotEqual(os.path.dirname(tmp_path), self.master_dir)
            with open(tmp_path, 'w') as fp:
                fp.write("TEST")

        mock_fetch.side_effect = _fake_fetch
        self.cache._download_image(self.uuid, self.master_path, self.dest_path)
        self.assertTrue(os.path.isfile(self.dest_path))
        self.assertTrue(os.path.isfile(self.master_path))
        self.assertEqual(os.stat(self.dest_path).st_ino,
                         os.stat(self.master_path).st_ino)
        with open(self.dest_path) as fp:
            self.assertEqual("TEST", fp.read())


class TestImageCacheCleanUp(base.TestCase):

    def setUp(self):
        super(TestImageCacheCleanUp, self).setUp()
        self.master_dir = tempfile.mkdtemp()
        self.cache = image_cache.ImageCache(self.master_dir,
                                            cache_size=10,
                                            cache_ttl=600)

    @mock.patch.object(image_cache.ImageCache, '_clean_up_ensure_cache_size')
    def test_clean_up_old_deleted(self, mock_clean_size):
        mock_clean_size.return_value = None
        files = [os.path.join(self.master_dir, str(i))
                 for i in range(2)]
        for filename in files:
            touch(filename)
        # NOTE(dtantsur): Can't alter ctime, have to set mtime to the future
        new_current_time = time.time() + 900
        os.utime(files[0], (new_current_time - 100, new_current_time - 100))
        with mock.patch.object(time, 'time', lambda: new_current_time):
            self.cache.clean_up()

        mock_clean_size.assert_called_once_with(mock.ANY, None)
        survived = mock_clean_size.call_args[0][0]
        self.assertEqual(1, len(survived))
        self.assertEqual(files[0], survived[0][0])
        # NOTE(dtantsur): do not compare milliseconds
        self.assertEqual(int(new_current_time - 100), int(survived[0][1]))
        self.assertEqual(int(new_current_time - 100),
                         int(survived[0][2].st_mtime))

    @mock.patch.object(image_cache.ImageCache, '_clean_up_ensure_cache_size')
    def test_clean_up_old_with_amount(self, mock_clean_size):
        files = [os.path.join(self.master_dir, str(i))
                 for i in range(2)]
        for filename in files:
            open(filename, 'wb').write('X')
        new_current_time = time.time() + 900
        with mock.patch.object(time, 'time', lambda: new_current_time):
            self.cache.clean_up(amount=1)

        self.assertFalse(mock_clean_size.called)
        # Exactly one file is expected to be deleted
        self.assertTrue(any(os.path.exists(f) for f in files))
        self.assertFalse(all(os.path.exists(f) for f in files))

    @mock.patch.object(image_cache.ImageCache, '_clean_up_ensure_cache_size')
    def test_clean_up_files_with_links_untouched(self, mock_clean_size):
        mock_clean_size.return_value = None
        files = [os.path.join(self.master_dir, str(i))
                 for i in range(2)]
        for filename in files:
            touch(filename)
            os.link(filename, filename + 'copy')

        new_current_time = time.time() + 900
        with mock.patch.object(time, 'time', lambda: new_current_time):
            self.cache.clean_up()

        for filename in files:
            self.assertTrue(os.path.exists(filename))
        mock_clean_size.assert_called_once_with([], None)

    @mock.patch.object(image_cache.ImageCache, '_clean_up_too_old')
    def test_clean_up_ensure_cache_size(self, mock_clean_ttl):
        mock_clean_ttl.side_effect = lambda *xx: xx
        # NOTE(dtantsur): Cache size in test is 10 bytes, we create 6 files
        # with 3 bytes each and expect 3 to be deleted
        files = [os.path.join(self.master_dir, str(i))
                 for i in range(6)]
        for filename in files:
            with open(filename, 'w') as fp:
                fp.write('123')
        # NOTE(dtantsur): Make 3 files 'newer' to check that
        # old ones are deleted first
        new_current_time = time.time() + 100
        for filename in files[:3]:
            os.utime(filename, (new_current_time, new_current_time))

        with mock.patch.object(time, 'time', lambda: new_current_time):
            self.cache.clean_up()

        for filename in files[:3]:
            self.assertTrue(os.path.exists(filename))
        for filename in files[3:]:
            self.assertFalse(os.path.exists(filename))

        mock_clean_ttl.assert_called_once_with(mock.ANY, None)

    @mock.patch.object(image_cache.ImageCache, '_clean_up_too_old')
    def test_clean_up_ensure_cache_size_with_amount(self, mock_clean_ttl):
        mock_clean_ttl.side_effect = lambda *xx: xx
        # NOTE(dtantsur): Cache size in test is 10 bytes, we create 6 files
        # with 3 bytes each and set amount to be 15, 5 files are to be deleted
        files = [os.path.join(self.master_dir, str(i))
                 for i in range(6)]
        for filename in files:
            with open(filename, 'w') as fp:
                fp.write('123')
        # NOTE(dtantsur): Make 1 file 'newer' to check that
        # old ones are deleted first
        new_current_time = time.time() + 100
        os.utime(files[0], (new_current_time, new_current_time))

        with mock.patch.object(time, 'time', lambda: new_current_time):
            self.cache.clean_up(amount=15)

        self.assertTrue(os.path.exists(files[0]))
        for filename in files[5:]:
            self.assertFalse(os.path.exists(filename))

        mock_clean_ttl.assert_called_once_with(mock.ANY, 15)

    @mock.patch.object(image_cache.LOG, 'info')
    @mock.patch.object(image_cache.ImageCache, '_clean_up_too_old')
    def test_clean_up_cache_still_large(self, mock_clean_ttl, mock_log):
        mock_clean_ttl.side_effect = lambda *xx: xx
        # NOTE(dtantsur): Cache size in test is 10 bytes, we create 2 files
        # than cannot be deleted and expected this to be logged
        files = [os.path.join(self.master_dir, str(i))
                 for i in range(2)]
        for filename in files:
            with open(filename, 'w') as fp:
                fp.write('123')
            os.link(filename, filename + 'copy')

        self.cache.clean_up()

        for filename in files:
            self.assertTrue(os.path.exists(filename))
        self.assertTrue(mock_log.called)
        mock_clean_ttl.assert_called_once_with(mock.ANY, None)

    @mock.patch.object(utils, 'rmtree_without_raise')
    @mock.patch.object(image_cache, '_fetch')
    def test_temp_images_not_cleaned(self, mock_fetch, mock_rmtree):
        def _fake_fetch(ctx, uuid, tmp_path, *args):
            with open(tmp_path, 'w') as fp:
                fp.write("TEST" * 10)

            # assume cleanup from another thread at this moment
            self.cache.clean_up()
            self.assertTrue(os.path.exists(tmp_path))

        mock_fetch.side_effect = _fake_fetch
        master_path = os.path.join(self.master_dir, 'uuid')
        dest_path = os.path.join(tempfile.mkdtemp(), 'dest')
        self.cache._download_image('uuid', master_path, dest_path)
        self.assertTrue(mock_rmtree.called)

    @mock.patch.object(utils, 'rmtree_without_raise')
    @mock.patch.object(image_cache, '_fetch')
    def test_temp_dir_exception(self, mock_fetch, mock_rmtree):
        mock_fetch.side_effect = exception.IronicException
        self.assertRaises(exception.IronicException,
                          self.cache._download_image,
                          'uuid', 'fake', 'fake')
        self.assertTrue(mock_rmtree.called)

    @mock.patch.object(image_cache.LOG, 'warn')
    @mock.patch.object(image_cache.ImageCache, '_clean_up_too_old')
    @mock.patch.object(image_cache.ImageCache, '_clean_up_ensure_cache_size')
    def test_clean_up_amount_not_satisfied(self, mock_clean_size,
                                           mock_clean_ttl, mock_log):
        mock_clean_ttl.side_effect = lambda *xx: xx
        mock_clean_size.side_effect = lambda listing, amount: amount
        self.cache.clean_up(amount=15)
        self.assertTrue(mock_log.called)

    def test_cleanup_ordering(self):

        class ParentCache(image_cache.ImageCache):
            def __init__(self):
                super(ParentCache, self).__init__('a', 1, 1, None)

        @image_cache.cleanup(priority=10000)
        class Cache1(ParentCache):
            pass

        @image_cache.cleanup(priority=20000)
        class Cache2(ParentCache):
            pass

        @image_cache.cleanup(priority=10000)
        class Cache3(ParentCache):
            pass

        self.assertEqual(image_cache._cache_cleanup_list[0][1], Cache2)

        # The order of caches with same prioirty is not deterministic.
        item_possibilities = [Cache1, Cache3]
        second_item_actual = image_cache._cache_cleanup_list[1][1]