--- a
+++ b/Allura/allura/tests/test_zarkov_helpers.py
@@ -0,0 +1,110 @@
+# -*- 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)