| # -*- 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) |