Allura/allura/tests/test_zarkov_helpers.py

# -*- coding: utf-8 -*-

#       Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF licenses this file
#       to you 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.

import unittest
from calendar import timegm
from datetime import datetime

import bson
import mock

from allura.lib import zarkov_helpers as zh

class TestZarkovClient(unittest.TestCase):

    def setUp(self):
        addr = 'tcp://0.0.0.0:0'
        ctx = mock.Mock()
        self.socket = mock.Mock()
        ctx.socket = mock.Mock(return_value=self.socket)
        PUSH=mock.Mock()
        with mock.patch('allura.lib.zarkov_helpers.zmq') as zmq:
            zmq.PUSH=PUSH
            zmq.Context.instance.return_value = ctx
            self.client = zh.ZarkovClient(addr)
        zmq.Context.instance.assert_called_once_with()
        ctx.socket.assert_called_once_with(PUSH)
        self.socket.connect.assert_called_once_with(addr)

    def test_event(self):
        self.client.event('test', dict(user='testuser'))
        obj = bson.BSON.encode(dict(
                type='test',
                context=dict(user='testuser'),
                extra=None))
        self.socket.send.assert_called_once_with(obj)

class TestZeroFill(unittest.TestCase):

    def setUp(self):
        self.dt_begin = datetime(2010, 6, 1)
        self.dt_end = datetime(2011, 7, 1)
        ts_begin = timegm(self.dt_begin.timetuple())
        ts_end = timegm(self.dt_end.timetuple())
        self.ts_ms_begin = ts_begin * 1000.0
        self.ts_ms_end = ts_end * 1000.0
        self.zarkov_data = dict(
            a=dict(
                a1=[ (self.ts_ms_begin, 1000), (self.ts_ms_end, 1000) ],
                a2=[ (self.ts_ms_begin, 1000), (self.ts_ms_end, 1000) ] ),
            b=dict(
                b1=[ (self.ts_ms_begin, 2000), (self.ts_ms_end, 2000) ],
                b2=[ (self.ts_ms_begin, 2000), (self.ts_ms_end, 2000) ] ))

    def test_to_utc_timestamp(self):
        self.assertEqual(
            zh.to_utc_timestamp(self.dt_begin),
            self.ts_ms_begin)
        self.assertEqual(
            zh.to_utc_timestamp(self.dt_end),
            self.ts_ms_end)

    def test_zero_fill_time_series_month(self):
        result = zh.zero_fill_time_series(
            self.zarkov_data['a']['a1'], 'month',
            datetime(2010, 5, 1), datetime(2011, 9, 1))
        self.assertEqual(result[0][1], 0)
        self.assertEqual(result[-1][1], 0)
        self.assertEqual(len(result), 17)
        self.assertEqual(result[1][1], 1000)
        self.assertEqual(result[-3][1], 1000)
        days_ms = 24 * 3600 * 1000
        min_delta = 28 * days_ms
        max_delta= 31 * days_ms
        for p1, p2 in zip(result, result[1:]):
            delta = p2[0]-p1[0]
            assert min_delta <= delta <= max_delta, delta

    def test_zero_fill_time_series_date(self):
        result = zh.zero_fill_time_series(
            self.zarkov_data['a']['a1'], 'date',
            datetime(2010, 5, 1), datetime(2011, 9, 1))
        self.assertEqual(len(result), 489)
        days_ms = 24 * 3600 * 1000
        for p1, p2 in zip(result, result[1:]):
            delta = p2[0]-p1[0]
            assert delta == days_ms

    def test_zero_fill_zarkov_month_dt(self):
        result = zh.zero_fill_zarkov_result(
            self.zarkov_data, 'month',
            datetime(2010, 5, 1), datetime(2011, 9, 1))
        a_result = result['a']['a1']
        b_result = result['b']['b2']
        self.assertEqual(a_result[0][1], 0)
        self.assertEqual(a_result[-1][1], 0)
        self.assertEqual(len(a_result), 17)
        self.assertEqual(a_result[1][1], 1000)
        self.assertEqual(a_result[-3][1], 1000)
        self.assertEqual(b_result[0][1], 0)
        self.assertEqual(b_result[-1][1], 0)
        self.assertEqual(len(b_result), 17)
        self.assertEqual(b_result[1][1], 2000)
        self.assertEqual(b_result[-3][1], 2000)

    def test_zero_fill_zarkov_month_str(self):
        result0 = zh.zero_fill_zarkov_result(
            self.zarkov_data, 'month',
            datetime(2010, 5, 1), datetime(2011, 9, 1))
        result1 = zh.zero_fill_zarkov_result(
            self.zarkov_data, 'month',
            '2010-5-1', '2011-09-1')
        self.assertEqual(result0, result1)