<gh_stars>10-100
import os.path
import mock
import shutil
import tempfile
from unittest import TestCase

from hacheck import spool

se = mock.sentinel


class TestSpool(TestCase):
    def setUp(self):
        self.root = tempfile.mkdtemp()
        spool.configure(self.root)

    def tearDown(self):
        shutil.rmtree(self.root)

    def test_configure_creates_root(self):
        spool.configure(os.path.join(self.root, 'spool'))
        assert os.path.exists, 'spool'

    def test_configure_no_write(self):
        new_root = os.path.join(self.root, 'non_writable')
        os.mkdir(new_root)
        os.chmod(new_root, 0o555)
        self.assertRaises(ValueError, spool.configure, new_root, needs_write=True)

    def test_configure_no_write_no_needs_write(self):
        new_root = os.path.join(self.root, 'non_writable')
        os.mkdir(new_root)
        os.chmod(new_root, 0o555)
        spool.configure(new_root, needs_write=False)

    def test_basic(self):
        svcname = 'test_basic'
        self.assertEquals(True, spool.status(svcname)[0])
        self.assertEquals(True, spool.status(svcname, 1234)[0])
        self.assertEquals(True, spool.status(svcname, 1234, '10.93.10.1')[0])
        spool.down(svcname, port=1234)
        self.assertEquals(True, spool.status(svcname)[0])
        self.assertEquals(False, spool.status(svcname, 1234)[0])
        self.assertEquals(True, spool.status(svcname, 1234, '10.93.10.1')[0])
        spool.down(svcname)
        self.assertEquals(False, spool.status(svcname)[0])
        self.assertEquals(False, spool.is_up(svcname)[0])
        self.assertEquals(False, spool.is_up(svcname, 1234, '10.93.10.1')[0])
        spool.up(svcname, port=1234)
        self.assertEquals(True, spool.status(svcname, 1234)[0])
        spool.up(svcname, port=1234, host='10.93.10.1')
        self.assertEquals(True, spool.status(svcname, 1234, '10.93.10.1')[0])
        spool.up(svcname)
        self.assertEquals(True, spool.status(svcname)[0])

    def test_all(self):
        svcname = 'test_all'
        self.assertEquals(True, spool.status(svcname)[0])
        spool.down('all')
        self.assertEquals(True, spool.status(svcname)[0])
        self.assertEquals(False, spool.is_up(svcname)[0])

    def test_status_all_down(self):
        self.assertEqual(len(list(spool.status_all_down())), 0)
        spool.down('foo')
        self.assertEqual(
            list(spool.status_all_down()),
            [('foo', None, None, {'service': 'foo', 'reason': '', 'expiration': None, 'creation': mock.ANY})]
        )

    def test_repeated_ups_works(self):
        spool.up('all')
        spool.up('all')

    def test_spool_file_path(self):
        self.assertEqual(os.path.join(self.root, 'foo:1234'), spool.spool_file_path("foo", port=1234))
        self.assertEqual(os.path.join(self.root, 'foo'), spool.spool_file_path("foo", None))
        self.assertEqual(os.path.join(self.root, 'foo:1234'), spool.spool_file_path("foo", port=1234, host='127.0.0.1'))
        self.assertEqual(
            os.path.join(self.root, 'foo:10.93.10.1:1234'),
            spool.spool_file_path("foo", port=1234, host='10.93.10.1'),
        )

    def test_parse_spool_file_path(self):
        self.assertEqual(("foo", None, 1234), spool.parse_spool_file_path(spool.spool_file_path("foo", 1234)))
        self.assertEqual(
            ("foo", '10.93.10.1', 1234),
            spool.parse_spool_file_path(spool.spool_file_path("foo", 1234, '10.93.10.1')),
        )

    def test_serialize_spool_file_contents(self):
        actual = spool.serialize_spool_file_contents("hi", expiration=12345, creation=54321)
        assert '"reason": "hi"' in actual
        assert '"expiration": 12345' in actual
        assert '"creation": 54321' in actual

    def test_deserialize_spool_file_contents_legacy(self):
        actual = spool.deserialize_spool_file_contents("this is a reason")
        self.assertEqual(actual, {"reason": "this is a reason", "expiration": None, "creation": None})

    def test_deserialize_spool_file_contents_new(self):
        actual = spool.deserialize_spool_file_contents('{"reason": "hi", "expiration": 12345, "creation": 12344}')
        self.assertEqual(actual, {"reason": "hi", "expiration": 12345, "creation": 12344})

    def test_status_creation(self):
        now = 1000
        svcname = 'test_status_creation'

        with mock.patch('time.time', return_value=now):
            spool.down(svcname)
            up, info_dict = spool.status(svcname)
            spool.up(svcname)

        self.assertEqual(info_dict['creation'], 1000)

    def test_status_expiration(self):
        svcname = 'test_status_expiration'
        now = 1000
        future = now + 10
        past = now - 10

        with mock.patch('time.time', return_value=now):
            self.assertEquals(True, spool.status(svcname)[0])

            # First, check with expiration in future; everything should behave normally.
            spool.down(svcname, expiration=future)
            self.assertEquals(False, spool.status(svcname)[0])

            # Now, let's make sure we remove the spool file if its expiration is in the past.
            spool.down(svcname, expiration=past)
            self.assertEquals(True, os.path.exists(spool.spool_file_path(svcname, None)))

            self.assertEquals(True, spool.status(svcname)[0])
            self.assertEquals(False, os.path.exists(spool.spool_file_path(svcname, None)))

    def test_repeated_downs_leaves_original_creation_time(self):
        first_creation = 1000
        first_expiration = first_creation + 60
        second_creation = first_creation + 10
        second_expiration = second_creation + 60
        with mock.patch('time.time', return_value=first_creation):
            spool.down('all', expiration=first_expiration)
        with mock.patch('time.time', return_value=second_creation):
            spool.down('all', expiration=second_expiration)

        with mock.patch('time.time', return_value=second_creation):
            _, info = spool.status('all')

        self.assertEquals(info['creation'], first_creation)
        self.assertEquals(info['expiration'], second_expiration)

    def test_repeated_downs_with_different_reason_changes_creation_time(self):
        first_creation = 1000
        first_expiration = first_creation + 60
        second_creation = first_creation + 10
        second_expiration = second_creation + 60
        with mock.patch('time.time', return_value=first_creation):
            spool.down('all', reason="first reason", expiration=first_expiration)
        with mock.patch('time.time', return_value=second_creation):
            spool.down('all', reason="second reason", expiration=second_expiration)

        with mock.patch('time.time', return_value=second_creation):
            _, info = spool.status('all')

        self.assertEquals(info['creation'], second_creation)
        self.assertEquals(info['expiration'], second_expiration)

    def test_repeated_downs_with_creation_arg_changes_creation_time(self):
        first_creation = 1000
        first_expiration = first_creation + 60
        second_creation = first_creation + 10
        second_expiration = second_creation + 60
        spool.down('all', creation=first_creation, expiration=first_expiration)
        spool.down('all', creation=second_creation, expiration=second_expiration)

        with mock.patch('time.time', return_value=second_creation):
            _, info = spool.status('all')

        self.assertEquals(info['creation'], second_creation)
        self.assertEquals(info['expiration'], second_expiration)

import os.path
import mock
import shutil
import tempfile
from unittest import TestCase

from hacheck import spool

se = mock.sentinel


class TestSpool(TestCase):
    def setUp(self):
        self.root = tempfile.mkdtemp()
        spool.configure(self.root)

    def tearDown(self):
        shutil.rmtree(self.root)

    def test_configure_creates_root(self):
        spool.configure(os.path.join(self.root, 'spool'))
        assert os.path.exists, 'spool'

    def test_configure_no_write(self):
        new_root = os.path.join(self.root, 'non_writable')
        os.mkdir(new_root)
        os.chmod(new_root, 0o555)
        self.assertRaises(ValueError, spool.configure, new_root, needs_write=True)

    def test_configure_no_write_no_needs_write(self):
        new_root = os.path.join(self.root, 'non_writable')
        os.mkdir(new_root)
        os.chmod(new_root, 0o555)
        spool.configure(new_root, needs_write=False)

    def test_basic(self):
        svcname = 'test_basic'
        self.assertEquals(True, spool.status(svcname)[0])
        self.assertEquals(True, spool.status(svcname, 1234)[0])
        self.assertEquals(True, spool.status(svcname, 1234, '10.93.10.1')[0])
        spool.down(svcname, port=1234)
        self.assertEquals(True, spool.status(svcname)[0])
        self.assertEquals(False, spool.status(svcname, 1234)[0])
        self.assertEquals(True, spool.status(svcname, 1234, '10.93.10.1')[0])
        spool.down(svcname)
        self.assertEquals(False, spool.status(svcname)[0])
        self.assertEquals(False, spool.is_up(svcname)[0])
        self.assertEquals(False, spool.is_up(svcname, 1234, '10.93.10.1')[0])
        spool.up(svcname, port=1234)
        self.assertEquals(True, spool.status(svcname, 1234)[0])
        spool.up(svcname, port=1234, host='10.93.10.1')
        self.assertEquals(True, spool.status(svcname, 1234, '10.93.10.1')[0])
        spool.up(svcname)
        self.assertEquals(True, spool.status(svcname)[0])

    def test_all(self):
        svcname = 'test_all'
        self.assertEquals(True, spool.status(svcname)[0])
        spool.down('all')
        self.assertEquals(True, spool.status(svcname)[0])
        self.assertEquals(False, spool.is_up(svcname)[0])

    def test_status_all_down(self):
        self.assertEqual(len(list(spool.status_all_down())), 0)
        spool.down('foo')
        self.assertEqual(
            list(spool.status_all_down()),
            [('foo', None, None, {'service': 'foo', 'reason': '', 'expiration': None, 'creation': mock.ANY})]
        )

    def test_repeated_ups_works(self):
        spool.up('all')
        spool.up('all')

    def test_spool_file_path(self):
        self.assertEqual(os.path.join(self.root, 'foo:1234'), spool.spool_file_path("foo", port=1234))
        self.assertEqual(os.path.join(self.root, 'foo'), spool.spool_file_path("foo", None))
        self.assertEqual(os.path.join(self.root, 'foo:1234'), spool.spool_file_path("foo", port=1234, host='127.0.0.1'))
        self.assertEqual(
            os.path.join(self.root, 'foo:10.93.10.1:1234'),
            spool.spool_file_path("foo", port=1234, host='10.93.10.1'),
        )

    def test_parse_spool_file_path(self):
        self.assertEqual(("foo", None, 1234), spool.parse_spool_file_path(spool.spool_file_path("foo", 1234)))
        self.assertEqual(
            ("foo", '10.93.10.1', 1234),
            spool.parse_spool_file_path(spool.spool_file_path("foo", 1234, '10.93.10.1')),
        )

    def test_serialize_spool_file_contents(self):
        actual = spool.serialize_spool_file_contents("hi", expiration=12345, creation=54321)
        assert '"reason": "hi"' in actual
        assert '"expiration": 12345' in actual
        assert '"creation": 54321' in actual

    def test_deserialize_spool_file_contents_legacy(self):
        actual = spool.deserialize_spool_file_contents("this is a reason")
        self.assertEqual(actual, {"reason": "this is a reason", "expiration": None, "creation": None})

    def test_deserialize_spool_file_contents_new(self):
        actual = spool.deserialize_spool_file_contents('{"reason": "hi", "expiration": 12345, "creation": 12344}')
        self.assertEqual(actual, {"reason": "hi", "expiration": 12345, "creation": 12344})

    def test_status_creation(self):
        now = 1000
        svcname = 'test_status_creation'

        with mock.patch('time.time', return_value=now):
            spool.down(svcname)
            up, info_dict = spool.status(svcname)
            spool.up(svcname)

        self.assertEqual(info_dict['creation'], 1000)

    def test_status_expiration(self):
        svcname = 'test_status_expiration'
        now = 1000
        future = now + 10
        past = now - 10

        with mock.patch('time.time', return_value=now):
            self.assertEquals(True, spool.status(svcname)[0])

            # First, check with expiration in future; everything should behave normally.
            spool.down(svcname, expiration=future)
            self.assertEquals(False, spool.status(svcname)[0])

            # Now, let's make sure we remove the spool file if its expiration is in the past.
            spool.down(svcname, expiration=past)
            self.assertEquals(True, os.path.exists(spool.spool_file_path(svcname, None)))

            self.assertEquals(True, spool.status(svcname)[0])
            self.assertEquals(False, os.path.exists(spool.spool_file_path(svcname, None)))

    def test_repeated_downs_leaves_original_creation_time(self):
        first_creation = 1000
        first_expiration = first_creation + 60
        second_creation = first_creation + 10
        second_expiration = second_creation + 60
        with mock.patch('time.time', return_value=first_creation):
            spool.down('all', expiration=first_expiration)
        with mock.patch('time.time', return_value=second_creation):
            spool.down('all', expiration=second_expiration)

        with mock.patch('time.time', return_value=second_creation):
            _, info = spool.status('all')

        self.assertEquals(info['creation'], first_creation)
        self.assertEquals(info['expiration'], second_expiration)

    def test_repeated_downs_with_different_reason_changes_creation_time(self):
        first_creation = 1000
        first_expiration = first_creation + 60
        second_creation = first_creation + 10
        second_expiration = second_creation + 60
        with mock.patch('time.time', return_value=first_creation):
            spool.down('all', reason="first reason", expiration=first_expiration)
        with mock.patch('time.time', return_value=second_creation):
            spool.down('all', reason="second reason", expiration=second_expiration)

        with mock.patch('time.time', return_value=second_creation):
            _, info = spool.status('all')

        self.assertEquals(info['creation'], second_creation)
        self.assertEquals(info['expiration'], second_expiration)

    def test_repeated_downs_with_creation_arg_changes_creation_time(self):
        first_creation = 1000
        first_expiration = first_creation + 60
        second_creation = first_creation + 10
        second_expiration = second_creation + 60
        spool.down('all', creation=first_creation, expiration=first_expiration)
        spool.down('all', creation=second_creation, expiration=second_expiration)

        with mock.patch('time.time', return_value=second_creation):
            _, info = spool.status('all')

        self.assertEquals(info['creation'], second_creation)
        self.assertEquals(info['expiration'], second_expiration)
# First, check with expiration in future; everything should behave normally. # Now, let's make sure we remove the spool file if its expiration is in the past.