<commit_before>import json
from Queue import Queue as Queue
from Queue import Empty, Full
import os
import socket
import sys
import tempfile
import threading
import time
import unittest

# pylint: disable=import-error
import consul as pyconsul
import mock

import manage
from manage.containerpilot import ContainerPilot
from manage.libconsul import Consul
from manage.libmanta import Manta
from manage.libmysql import MySQL
from manage.utils import *


thread_data = threading.local()

def trace_wait(frame, event, arg):
    if event != 'call':
        return
    func_name = frame.f_code.co_name
    if func_name in thread_data.trace_into or func_name == '_mock_call':
        node_name = threading.current_thread().name
        timeout = thread_data.timeout
        try:
            # block until new work is available but no more than 5
            # seconds before killing the thread
            val = thread_data.queue.get(True, timeout)
            if val == 'quit':
                raise Empty
            thread_data.queue.task_done()
        except Empty:
            sys.exit(0)


class TestNode(object):
    """
    TestNode is a context containing a manage.Node object and which runs
    the function under test in its own thread. The function under test
    is stepped thru with each `tick()` call on this object.
    """
    def __init__(self, node=None, fn=None, trace_into=[]):
        self.node = node
        self._nodeq = Queue(1)
        self._thread = threading.Thread(
            target=self.step_thru,
            name=node.name,
            args=(fn, node),
            kwargs=dict(queue=self._nodeq, trace_into=trace_into))
        self._thread.start()

    def step_thru(self, fn, *args, **kwargs):
        """
        Step thru the function `fn`. The function will block every time it
        calls a function w/ a name listed in `trace_into` until a message
        is pushed onto the `queue` or a number of seconds equal to
        `test_timeout` passes. Exceptions are allowed to bubble up and
        crash the thread.
        """
        thread_data.queue = kwargs.pop('queue')
        thread_data.trace_into = kwargs.pop('trace_into', None)
        thread_data.timeout = kwargs.pop('test_timeout', 5)
        sys.settrace(trace_wait)
        fn(*args, **kwargs)

    def tick(self, val=True):
        print('tick!: {}'.format(val))
        try:
            self._nodeq.put(val, True, 1)
        except Full:
            raise Full('{} raised Full at {}'.format(self.node.name, val))
        time.sleep(0.01) # ensure we yield the main thread

    def end(self):
        try:
            self._nodeq.put('quit', True, 1)
        except Full:
            pass
        finally:
            self._thread.join()


class TestPreStart(unittest.TestCase):

    def setUp(self):
        logging.getLogger('manage').setLevel(logging.WARN)
        consul = mock.MagicMock()
        manta = mock.MagicMock()
        my = mock.MagicMock()
        my.datadir = tempfile.mkdtemp()
        _node1 = manage.Node(consul=consul, manta=manta, mysql=my)
        trace_into=['render', 'has_snapshot', 'get_backup',
                    'restore_from_snapshot', 'initialize_db', 'get']
        self.node_1 = TestNode(node=_node1, fn=manage.pre_start,
                               trace_into=trace_into)

    def tearDown(self):
        logging.getLogger('manage').setLevel(logging.DEBUG)
        self.node_1.end()

    def test_pre_start_first_node(self):
        """
        The first node will not attempt to download a snapshot from Manta.
        """
        self.node_1.node.consul.has_snapshot.return_value = False
        manage.pre_start(self.node_1.node)
        self.node_1.node.consul.has_snapshot.assert_called_once()
        self.node_1.node.mysql.initialize_db.assert_called_once()
        self.assertFalse(self.node_1.node.manta.get_backup.called)
        self.assertFalse(self.node_1.node.mysql.restore_from_snapshot.called)

    def test_pre_start_snapshot_complete(self):
        """
        Given a successful snapshot by the first node, a new node will
        download the snapshot from Manta
        """
        self.node_1.node.consul.has_snapshot.return_value = True
        manage.pre_start(self.node_1.node)
        self.node_1.node.consul.has_snapshot.assert_called_once()
        self.node_1.node.manta.get_backup.assert_called_once()
        self.node_1.node.mysql.restore_from_snapshot.assert_called_once()
        self.assertFalse(self.node_1.node.mysql.initialize_db.called)

    def test_pre_start_no_reinitialization(self):
        """
        Given a node that's restarted, pre_start should not try
        to re-initialize the node.
        """
        os.mkdir(os.path.join(self.node_1.node.mysql.datadir, 'mysql'))
        self.node_1.node.consul.has_snapshot.return_value = True
        manage.pre_start(self.node_1.node)
        self.assertFalse(self.node_1.node.consul.has_snapshot.called)

    def test_pre_start_snapshot_incomplete(self):
        """
        Given a snapshot that has been marked successful but not
        completed, a new node will wait and not crash.
        """
        self.node_1.node.consul = Consul(TEST_ENVIRON)
        self.node_1.node.consul.client = mock.MagicMock()
        self.node_1.node.consul.client.kv.get.side_effect = pyconsul.ConsulException('')

        self.node_1.tick('pre_start')
        self.node_1.tick('render')
        self.node_1.tick('has_snapshot')
        self.node_1.tick('get')
        self.node_1.tick('get')
        self.node_1.node.consul.client.kv.get.side_effect = None
        self.node_1.node.consul.client.kv.get.return_value = [0,{'Value': 'ok'}]
        self.node_1.tick('get_backup')
        self.node_1.tick('restore_from_snapshot')
        self.node_1.end()

        self.node_1.node.manta.get_backup.assert_called_once()
        self.assertEqual(self.node_1.node.consul.client.kv.get.call_count, 2)
        self.node_1.node.mysql.restore_from_snapshot.assert_called_once()
        self.assertFalse(self.node_1.node.mysql.initialize_db.called)


class TestMySQL(unittest.TestCase):
<commit_msg>Unit testing for snapshot_task<commit_after>import json
from Queue import Queue as Queue
from Queue import Empty, Full
import os
import socket
import sys
import tempfile
import threading
import time
import unittest

# pylint: disable=import-error
import consul as pyconsul
import mock

import manage
from manage.containerpilot import ContainerPilot
from manage.libconsul import Consul
from manage.libmanta import Manta
from manage.libmysql import MySQL
from manage.utils import *


thread_data = threading.local()

def trace_wait(frame, event, arg):
    if event != 'call':
        return
    func_name = frame.f_code.co_name
    if func_name in thread_data.trace_into or func_name == '_mock_call':
        node_name = threading.current_thread().name
        timeout = thread_data.timeout
        try:
            # block until new work is available but no more than 5
            # seconds before killing the thread
            val = thread_data.queue.get(True, timeout)
            if val == 'quit':
                raise Empty
            thread_data.queue.task_done()
        except Empty:
            sys.exit(0)


class TestNode(object):
    """
    TestNode is a context containing a manage.Node object and which runs
    the function under test in its own thread. The function under test
    is stepped thru with each `tick()` call on this object.
    """
    def __init__(self, node=None, fn=None, trace_into=[]):
        self.node = node
        self._nodeq = Queue(1)
        self._thread = threading.Thread(
            target=self.step_thru,
            name=node.name,
            args=(fn, node),
            kwargs=dict(queue=self._nodeq, trace_into=trace_into))
        self._thread.start()

    def step_thru(self, fn, *args, **kwargs):
        """
        Step thru the function `fn`. The function will block every time it
        calls a function w/ a name listed in `trace_into` until a message
        is pushed onto the `queue` or a number of seconds equal to
        `test_timeout` passes. Exceptions are allowed to bubble up and
        crash the thread.
        """
        thread_data.queue = kwargs.pop('queue')
        thread_data.trace_into = kwargs.pop('trace_into', None)
        thread_data.timeout = kwargs.pop('test_timeout', 5)
        sys.settrace(trace_wait)
        fn(*args, **kwargs)

    def tick(self, val=True):
        print('tick!: {}'.format(val))
        try:
            self._nodeq.put(val, True, 1)
        except Full:
            raise Full('{} raised Full at {}'.format(self.node.name, val))
        time.sleep(0.01) # ensure we yield the main thread

    def end(self):
        try:
            self._nodeq.put('quit', True, 1)
        except Full:
            pass
        finally:
            self._thread.join()


class TestPreStart(unittest.TestCase):

    def setUp(self):
        logging.getLogger('manage').setLevel(logging.WARN)
        consul = mock.MagicMock()
        manta = mock.MagicMock()
        my = mock.MagicMock()
        my.datadir = tempfile.mkdtemp()
        _node1 = manage.Node(consul=consul, manta=manta, mysql=my)
        trace_into=['render', 'has_snapshot', 'get_backup',
                    'restore_from_snapshot', 'initialize_db', 'get']
        self.node_1 = TestNode(node=_node1, fn=manage.pre_start,
                               trace_into=trace_into)

    def tearDown(self):
        logging.getLogger('manage').setLevel(logging.DEBUG)
        self.node_1.end()

    def test_pre_start_first_node(self):
        """
        The first node will not attempt to download a snapshot from Manta.
        """
        self.node_1.node.consul.has_snapshot.return_value = False
        manage.pre_start(self.node_1.node)
        self.node_1.node.consul.has_snapshot.assert_called_once()
        self.node_1.node.mysql.initialize_db.assert_called_once()
        self.assertFalse(self.node_1.node.manta.get_backup.called)
        self.assertFalse(self.node_1.node.mysql.restore_from_snapshot.called)

    def test_pre_start_snapshot_complete(self):
        """
        Given a successful snapshot by the first node, a new node will
        download the snapshot from Manta
        """
        self.node_1.node.consul.has_snapshot.return_value = True
        manage.pre_start(self.node_1.node)
        self.node_1.node.consul.has_snapshot.assert_called_once()
        self.node_1.node.manta.get_backup.assert_called_once()
        self.node_1.node.mysql.restore_from_snapshot.assert_called_once()
        self.assertFalse(self.node_1.node.mysql.initialize_db.called)

    def test_pre_start_no_reinitialization(self):
        """
        Given a node that's restarted, pre_start should not try
        to re-initialize the node.
        """
        os.mkdir(os.path.join(self.node_1.node.mysql.datadir, 'mysql'))
        self.node_1.node.consul.has_snapshot.return_value = True
        manage.pre_start(self.node_1.node)
        self.assertFalse(self.node_1.node.consul.has_snapshot.called)

    def test_pre_start_snapshot_incomplete(self):
        """
        Given a snapshot that has been marked successful but not
        completed, a new node will wait and not crash.
        """
        self.node_1.node.consul = Consul(TEST_ENVIRON)
        self.node_1.node.consul.client = mock.MagicMock()
        self.node_1.node.consul.client.kv.get.side_effect = pyconsul.ConsulException('')

        self.node_1.tick('pre_start')
        self.node_1.tick('render')
        self.node_1.tick('has_snapshot')
        self.node_1.tick('get')
        self.node_1.tick('get')
        self.node_1.node.consul.client.kv.get.side_effect = None
        self.node_1.node.consul.client.kv.get.return_value = [0,{'Value': 'ok'}]
        self.node_1.tick('get_backup')
        self.node_1.tick('restore_from_snapshot')
        self.node_1.end()

        self.node_1.node.manta.get_backup.assert_called_once()
        self.assertEqual(self.node_1.node.consul.client.kv.get.call_count, 2)
        self.node_1.node.mysql.restore_from_snapshot.assert_called_once()
        self.assertFalse(self.node_1.node.mysql.initialize_db.called)


class TestHealth(unittest.TestCase):
    pass

class TestOnChange(unittest.TestCase):
    pass

class TestSnapshotTask(unittest.TestCase):

    def setUp(self):
        logging.getLogger('manage').setLevel(logging.WARN)
        consul = mock.MagicMock()
        manta = mock.MagicMock()
        my = mock.MagicMock()
        cp = ContainerPilot()
        cp.state = PRIMARY
        my.datadir = tempfile.mkdtemp()
        self.node = manage.Node(consul=consul, cp=cp, manta=manta, mysql=my)

    def tearDown(self):
        logging.getLogger('manage').setLevel(logging.DEBUG)
        try:
            os.remove('/tmp/mysql-backup-run')
        except OSError as ex:
            pass

    def test_not_snapshot_node(self):
        """ Don't snapshot if this isn't the snapshot node """
        # TODO update when this logic changes
        self.node.cp.state = REPLICA
        manage.snapshot_task(self.node)
        self.assertFalse(self.node.mysql.query.called)

    def test_binlog_stale(self):
        """ Snapshot if the binlog is stale even if its not time to do so """
        self.node.consul.is_check_healthy.return_value = True
        self.node.consul.get.return_value = 'mybackup1'
        self.node.mysql.query.return_value = [['mybackup2']]

        with mock.patch('manage.write_snapshot') as ws:
            manage.snapshot_task(self.node)
            self.node.mysql.query.assert_called_once()
            self.node.consul.get.assert_called_once()
            self.assertTrue(ws.called)

    def test_backup_already_running(self):
        """ Don't snapshot if there's already a snapshot running """
        self.node.consul.is_check_healthy.return_value = False
        self.node.consul.get.return_value = 'mybackup1'
        self.node.mysql.query.return_value = [['mybackup2']]

        with mock.patch('manage.write_snapshot') as ws:
            lockfile_name = '/tmp/mysql-backup-run'
            try:
                backup_lock = open(lockfile_name, 'w')
                fcntl.flock(backup_lock, fcntl.LOCK_EX|fcntl.LOCK_NB)
                manage.snapshot_task(self.node)
            finally:
                fcntl.flock(backup_lock, fcntl.LOCK_UN)
                backup_lock.close()
            self.assertFalse(ws.called)

    def test_time_to_snapshot(self):
        """ Snapshot if the timer has elapsed even if the binlog isn't stale"""
        self.node.consul.is_check_healthy.return_value = False
        self.node.consul.get.return_value = 'mybackup1'
        self.node.mysql.query.return_value = [['mybackup1']]
        with mock.patch('manage.write_snapshot') as ws:
            manage.snapshot_task(self.node)
            self.assertTrue(ws.called)


class TestMySQL(unittest.TestCase):

{'fraction_non_alphanumeric': 0.10377611140867703, 'fraction_numerical': 0.006829137653990359, 'mean_word_length': 2.547138446924721, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5670878', 'n_tokens_mistral': 4660, 'n_tokens_neox': 4585, 'n_words': 965}