# Copyright (c) 2010-2012 OpenStack Foundation
#
# 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.

from __future__ import print_function
import unittest
from contextlib import contextmanager
import os
import logging
import errno
import math
import time
from shutil import rmtree, copy
from tempfile import mkdtemp, NamedTemporaryFile
import json

import mock
from mock import patch, call
from six.moves import reload_module

from swift.container.backend import DATADIR
from swift.common import db_replicator
from swift.common.utils import (normalize_timestamp, hash_path,
                                storage_directory)
from swift.common.exceptions import DriveNotMounted
from swift.common.swob import HTTPException

from test import unit
from test.unit.common.test_db import ExampleBroker


TEST_ACCOUNT_NAME = 'a c t'
TEST_CONTAINER_NAME = 'c o n'


def teardown_module():
    "clean up my monkey patching"
    reload_module(db_replicator)


@contextmanager
def lock_parent_directory(filename):
    yield True


class FakeRing(object):
    class Ring(object):
        devs = []

        def __init__(self, path, reload_time=15, ring_name=None):
            pass

        def get_part(self, account, container=None, obj=None):
            return 0

        def get_part_nodes(self, part):
            return []

        def get_more_nodes(self, *args):
            return []


class FakeRingWithSingleNode(object):
    class Ring(object):
        devs = [dict(
            id=1, weight=10.0, zone=1, ip='1.1.1.1', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.1', replication_port=6200, region=1
        )]

        def __init__(self, path, reload_time=15, ring_name=None):
            pass

        def get_part(self, account, container=None, obj=None):
            return 0

        def get_part_nodes(self, part):
            return self.devs

        def get_more_nodes(self, *args):
            return (d for d in self.devs)


class FakeRingWithNodes(object):
    class Ring(object):
        devs = [dict(
            id=1, weight=10.0, zone=1, ip='1.1.1.1', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.1', replication_port=6200, region=1
        ), dict(
            id=2, weight=10.0, zone=2, ip='1.1.1.2', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.2', replication_port=6200, region=2
        ), dict(
            id=3, weight=10.0, zone=3, ip='1.1.1.3', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.3', replication_port=6200, region=1
        ), dict(
            id=4, weight=10.0, zone=4, ip='1.1.1.4', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.4', replication_port=6200, region=2
        ), dict(
            id=5, weight=10.0, zone=5, ip='1.1.1.5', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.5', replication_port=6200, region=1
        ), dict(
            id=6, weight=10.0, zone=6, ip='1.1.1.6', port=6200, device='sdb',
            meta='', replication_ip='1.1.1.6', replication_port=6200, region=2
        )]

        def __init__(self, path, reload_time=15, ring_name=None):
            pass

        def get_part(self, account, container=None, obj=None):
            return 0

        def get_part_nodes(self, part):
            return self.devs[:3]

        def get_more_nodes(self, *args):
            return (d for d in self.devs[3:])


class FakeProcess(object):
    def __init__(self, *codes):
        self.codes = iter(codes)
        self.args = None
        self.kwargs = None

    def __call__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs

        class Failure(object):
            def communicate(innerself):
                next_item = next(self.codes)
                if isinstance(next_item, int):
                    innerself.returncode = next_item
                    return next_item
                raise next_item
        return Failure()


@contextmanager
def _mock_process(*args):
    orig_process = db_replicator.subprocess.Popen
    db_replicator.subprocess.Popen = FakeProcess(*args)
    yield db_replicator.subprocess.Popen
    db_replicator.subprocess.Popen = orig_process


class ReplHttp(object):
    def __init__(self, response=None, set_status=200):
        self.response = response
        self.set_status = set_status
    replicated = False
    host = 'localhost'

    def replicate(self, *args):
        self.replicated = True

        class Response(object):
            status = self.set_status
            data = self.response

            def read(innerself):
                return self.response
        return Response()


class ChangingMtimesOs(object):
    def __init__(self):
        self.mtime = 0

    def __call__(self, *args, **kwargs):
        self.mtime += 1
        return self.mtime


class FakeBroker(object):
    db_file = __file__
    get_repl_missing_table = False
    stub_replication_info = None
    db_type = 'container'
    db_contains_type = 'object'
    info = {'account': TEST_ACCOUNT_NAME, 'container': TEST_CONTAINER_NAME}

    def __init__(self, *args, **kwargs):
        self.locked = False
        return None

    @contextmanager
    def lock(self):
        self.locked = True
        yield True
        self.locked = False

    def get_sync(self, *args, **kwargs):
        return 5

    def get_syncs(self):
        return []

    def get_items_since(self, point, *args):
        if point == 0:
            return [{'ROWID': 1}]
        if point == -1:
            return [{'ROWID': 1}, {'ROWID': 2}]
        return []

    def merge_syncs(self, *args, **kwargs):
        self.args = args

    def merge_items(self, *args):
        self.args = args

    def get_replication_info(self):
        if self.get_repl_missing_table:
            raise Exception('no such table')
        info = dict(self.info)
        info.update({
            'hash': 12345,
            'delete_timestamp': 0,
            'put_timestamp': 1,
            'created_at': 1,
            'count': 0,
        })
        if self.stub_replication_info:
            info.update(self.stub_replication_info)
        return info

    def reclaim(self, item_timestamp, sync_timestamp):
        pass

    def newid(self, remote_d):
        pass

    def update_metadata(self, metadata):
        self.metadata = metadata

    def merge_timestamps(self, created_at, put_timestamp, delete_timestamp):
        self.created_at = created_at
        self.put_timestamp = put_timestamp
        self.delete_timestamp = delete_timestamp


class FakeAccountBroker(FakeBroker):
    db_type = 'account'
    db_contains_type = 'container'
    info = {'account': TEST_ACCOUNT_NAME}


class TestReplicator(db_replicator.Replicator):
    server_type = 'container'
    ring_file = 'container.ring.gz'
    brokerclass = FakeBroker
    datadir = DATADIR
    default_port = 1000


class TestDBReplicator(unittest.TestCase):
    def setUp(self):
        db_replicator.ring = FakeRing()
        self.delete_db_calls = []
        self._patchers = []
        # recon cache path
        self.recon_cache = mkdtemp()
        rmtree(self.recon_cache, ignore_errors=1)
        os.mkdir(self.recon_cache)

    def tearDown(self):
        for patcher in self._patchers:
            patcher.stop()
        rmtree(self.recon_cache, ignore_errors=1)

    def _patch(self, patching_fn, *args, **kwargs):
        patcher = patching_fn(*args, **kwargs)
        patched_thing = patcher.start()
        self._patchers.append(patcher)
        return patched_thing

    def stub_delete_db(self, broker):
        self.delete_db_calls.append('/path/to/file')

    def test_creation(self):
        # later config should be extended to assert more config options
        replicator = TestReplicator({'node_timeout': '3.5'})
        self.assertEqual(replicator.node_timeout, 3.5)

    def test_repl_connection(self):
        node = {'replication_ip': '127.0.0.1', 'replication_port': 80,
                'device': 'sdb1'}
        conn = db_replicator.ReplConnection(node, '1234567890', 'abcdefg',
                                            logging.getLogger())

        def req(method, path, body, headers):
            self.assertEqual(method, 'REPLICATE')
            self.assertEqual(headers['Content-Type'], 'application/json')

        class Resp(object):
            def read(self):
                return 'data'
        resp = Resp()
        conn.request = req
        conn.getresponse = lambda *args: resp
        self.assertEqual(conn.replicate(1, 2, 3), resp)

        def other_req(method, path, body, headers):
            raise Exception('blah')
        conn.request = other_req
        self.assertEqual(conn.replicate(1, 2, 3), None)

    def test_rsync_file(self):
        replicator = TestReplicator({})
        with _mock_process(-1):
            self.assertEqual(
                False,
                replicator._rsync_file('/some/file', 'remote:/some/file'))
        with _mock_process(0):
            self.assertEqual(
                True,
                replicator._rsync_file('/some/file', 'remote:/some/file'))

    def test_rsync_file_popen_args(self):
        replicator = TestReplicator({})
        with _mock_process(0) as process:
            replicator._rsync_file('/some/file', 'remote:/some_file')
            exp_args = ([
                'rsync', '--quiet', '--no-motd',
                '--timeout=%s' % int(math.ceil(replicator.node_timeout)),
                '--contimeout=%s' % int(math.ceil(replicator.conn_timeout)),
                '--whole-file', '/some/file', 'remote:/some_file'],)
            self.assertEqual(exp_args, process.args)

    def test_rsync_file_popen_args_whole_file_false(self):
        replicator = TestReplicator({})
        with _mock_process(0) as process:
            replicator._rsync_file('/some/file', 'remote:/some_file', False)
            exp_args = ([
                'rsync', '--quiet', '--no-motd',
                '--timeout=%s' % int(math.ceil(replicator.node_timeout)),
                '--contimeout=%s' % int(math.ceil(replicator.conn_timeout)),
                '/some/file', 'remote:/some_file'],)
            self.assertEqual(exp_args, process.args)

    def test_rsync_file_popen_args_different_region_and_rsync_compress(self):
        replicator = TestReplicator({})
        for rsync_compress in (False, True):
            replicator.rsync_compress = rsync_compress
            for different_region in (False, True):
                with _mock_process(0) as process:
                    replicator._rsync_file('/some/file', 'remote:/some_file',
                                           False, different_region)
                    if rsync_compress and different_region:
                        # --compress arg should be passed to rsync binary
                        # only when rsync_compress option is enabled
                        # AND destination node is in a different
                        # region
                        self.assertTrue('--compress' in process.args[0])
                    else:
                        self.assertFalse('--compress' in process.args[0])

    def test_rsync_db(self):
        replicator = TestReplicator({})
        replicator._rsync_file = lambda *args, **kwargs: True
        fake_device = {'replication_ip': '127.0.0.1', 'device': 'sda1'}
        replicator._rsync_db(FakeBroker(), fake_device, ReplHttp(), 'abcd')

    def test_rsync_db_rsync_file_call(self):
        fake_device = {'ip': '127.0.0.1', 'port': '0',
                       'replication_ip': '127.0.0.1', 'replication_port': '0',
                       'device': 'sda1'}

        class MyTestReplicator(TestReplicator):
            def __init__(self, db_file, remote_file):
                super(MyTestReplicator, self).__init__({})
                self.db_file = db_file
                self.remote_file = remote_file

            def _rsync_file(self_, db_file, remote_file, whole_file=True,
                            different_region=False):
                self.assertEqual(self_.db_file, db_file)
                self.assertEqual(self_.remote_file, remote_file)
                self_._rsync_file_called = True
                return False

        broker = FakeBroker()
        remote_file = '127.0.0.1::container/sda1/tmp/abcd'
        replicator = MyTestReplicator(broker.db_file, remote_file)
        replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
        self.assertTrue(replicator._rsync_file_called)

    def test_rsync_db_rsync_file_failure(self):
        class MyTestReplicator(TestReplicator):
            def __init__(self):
                super(MyTestReplicator, self).__init__({})
                self._rsync_file_called = False

            def _rsync_file(self_, *args, **kwargs):
                self.assertEqual(
                    False, self_._rsync_file_called,
                    '_sync_file() should only be called once')
                self_._rsync_file_called = True
                return False

        with patch('os.path.exists', lambda *args: True):
            replicator = MyTestReplicator()
            fake_device = {'ip': '127.0.0.1', 'replication_ip': '127.0.0.1',
                           'device': 'sda1'}
            replicator._rsync_db(FakeBroker(), fake_device, ReplHttp(), 'abcd')
            self.assertEqual(True, replicator._rsync_file_called)

    def test_rsync_db_change_after_sync(self):
        class MyTestReplicator(TestReplicator):
            def __init__(self, broker):
                super(MyTestReplicator, self).__init__({})
                self.broker = broker
                self._rsync_file_call_count = 0

            def _rsync_file(self_, db_file, remote_file, whole_file=True,
                            different_region=False):
                self_._rsync_file_call_count += 1
                if self_._rsync_file_call_count == 1:
                    self.assertEqual(True, whole_file)
                    self.assertEqual(False, self_.broker.locked)
                elif self_._rsync_file_call_count == 2:
                    self.assertEqual(False, whole_file)
                    self.assertEqual(True, self_.broker.locked)
                else:
                    raise RuntimeError('_rsync_file() called too many times')
                return True

        # with journal file
        with patch('os.path.exists', lambda *args: True):
            broker = FakeBroker()
            replicator = MyTestReplicator(broker)
            fake_device = {'ip': '127.0.0.1', 'replication_ip': '127.0.0.1',
                           'device': 'sda1'}
            replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
            self.assertEqual(2, replicator._rsync_file_call_count)

        # with new mtime
        with patch('os.path.exists', lambda *args: False):
            with patch('os.path.getmtime', ChangingMtimesOs()):
                broker = FakeBroker()
                replicator = MyTestReplicator(broker)
                fake_device = {'ip': '127.0.0.1',
                               'replication_ip': '127.0.0.1',
                               'device': 'sda1'}
                replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
                self.assertEqual(2, replicator._rsync_file_call_count)

    def test_in_sync(self):
        replicator = TestReplicator({})
        self.assertEqual(replicator._in_sync(
            {'id': 'a', 'point': 0, 'max_row': 0, 'hash': 'b'},
            {'id': 'a', 'point': -1, 'max_row': 0, 'hash': 'b'},
            FakeBroker(), -1), True)
        self.assertEqual(replicator._in_sync(
            {'id': 'a', 'point': -1, 'max_row': 0, 'hash': 'b'},
            {'id': 'a', 'point': -1, 'max_row': 10, 'hash': 'b'},
            FakeBroker(), -1), True)
        self.assertEqual(bool(replicator._in_sync(
            {'id': 'a', 'point': -1, 'max_row': 0, 'hash': 'c'},
            {'id': 'a', 'point': -1, 'max_row': 10, 'hash': 'd'},
            FakeBroker(), -1)), False)

    def test_run_once_no_local_device_in_ring(self):
        logger = unit.debug_logger('test-replicator')
        replicator = TestReplicator({'recon_cache_path': self.recon_cache},
                                    logger=logger)
        with patch('swift.common.db_replicator.whataremyips',
                   return_value=['127.0.0.1']):
            replicator.run_once()
        expected = [
            "Can't find itself 127.0.0.1 with port 1000 "
            "in ring file, not replicating",
        ]
        self.assertEqual(expected, logger.get_lines_for_level('error'))

    def test_run_once_with_local_device_in_ring(self):
        logger = unit.debug_logger('test-replicator')
        base = 'swift.common.db_replicator.'
        with patch(base + 'whataremyips', return_value=['1.1.1.1']), \
                patch(base + 'ring', FakeRingWithNodes()):
            replicator = TestReplicator({'bind_port': 6200,
                                         'recon_cache_path': self.recon_cache},
                                        logger=logger)
            replicator.run_once()
        self.assertFalse(logger.get_lines_for_level('error'))

    def test_run_once_no_ips(self):
        replicator = TestReplicator({}, logger=unit.FakeLogger())
        self._patch(patch.object, db_replicator, 'whataremyips',
                    lambda *a, **kw: [])

        replicator.run_once()

        self.assertEqual(
            replicator.logger.log_dict['error'],
            [(('ERROR Failed to get my own IPs?',), {})])

    def test_run_once_node_is_not_mounted(self):
        db_replicator.ring = FakeRingWithSingleNode()
        # If a bind_ip is specified, it's plumbed into whataremyips() and
        # returned by itself.
        conf = {'mount_check': 'true', 'bind_ip': '1.1.1.1',
                'bind_port': 6200}
        replicator = TestReplicator(conf, logger=unit.FakeLogger())
        self.assertEqual(replicator.mount_check, True)
        self.assertEqual(replicator.port, 6200)

        def mock_ismount(path):
            self.assertEqual(path,
                             os.path.join(replicator.root,
                                          replicator.ring.devs[0]['device']))
            return False

        self._patch(patch.object, db_replicator, 'ismount', mock_ismount)
        replicator.run_once()

        self.assertEqual(
            replicator.logger.log_dict['warning'],
            [(('Skipping %(device)s as it is not mounted' %
               replicator.ring.devs[0],), {})])

    def test_run_once_node_is_mounted(self):
        db_replicator.ring = FakeRingWithSingleNode()
        conf = {'mount_check': 'true', 'bind_port': 6200}
        replicator = TestReplicator(conf, logger=unit.FakeLogger())
        self.assertEqual(replicator.mount_check, True)