# -*- coding: utf-8 -*-
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)