Permalink
Browse files

Merged expressions unit tests from pypercube

  • Loading branch information...
1 parent 272eb36 commit d3715dd95f92be14de80c8c0f7755852ca3afa23 @tsileo committed Oct 1, 2013
Showing with 180 additions and 0 deletions.
  1. +115 −0 cube/tests/test_event_expression.py
  2. +65 −0 cube/tests/test_metric_expression.py
@@ -0,0 +1,115 @@
+# -*- encoding: utf-8 -*-
+
+"""
+Copyright (c) 2012 Steven Buss
+Originally from:
+https://raw.github.com/sbuss/pypercube/master/tests/test_event_expression.py
+"""
+
+import unittest
+
+from cube.expression import EventExpression
+
+
+class TestEventExpressions(unittest.TestCase):
+ def test_copy(self):
+ e1 = EventExpression('request', ['path', 'elapsed_ms'])
+ e2 = e1.copy()
+ self.assertEqual(e1, e2)
+ e1 = e1.eq('path', '/')
+ e3 = e1.copy()
+ self.assertNotEqual(e1, e2)
+ self.assertEqual(e1, e3)
+ self.assertNotEqual(e2, e3)
+
+ def test_equality(self):
+ e1 = EventExpression('request')
+ e2 = EventExpression('request')
+ self.assertEqual(e1, e2)
+ e1 = EventExpression('request', 'path')
+ self.assertNotEqual(e1, e2)
+ e2 = EventExpression('request', 'path')
+ self.assertEqual(e1, e2)
+ e1 = EventExpression('request', ['path', 'elapsed_ms'])
+ self.assertNotEqual(e1, e2)
+ e2 = EventExpression('request', ['path', 'elapsed_ms'])
+ self.assertEqual(e1, e2)
+ e1 = EventExpression('request', ['path', 'elapsed_ms']).eq('path', '/')
+ self.assertNotEqual(e1, e2)
+ e2 = EventExpression('request', ['path', 'elapsed_ms']).eq('path', '/')
+ self.assertEqual(e1, e2)
+ e1 = EventExpression('request', ['path', 'elapsed_ms']).eq(
+ 'path', '/').gt('elapsed_ms', 500)
+ self.assertNotEqual(e1, e2)
+ e2 = EventExpression('request', ['path', 'elapsed_ms']).eq(
+ 'path', '/').gt('elapsed_ms', 500)
+ self.assertEqual(e1, e2)
+
+ def _test_filter(self, filter_str):
+ e = EventExpression('test')
+ self.assertTrue(hasattr(e, filter_str))
+ f = getattr(e, filter_str)
+ filtered = f('elapsed_ms', 500)
+ self.assertTrue(isinstance(filtered, EventExpression))
+ self.assertTrue(len(filtered.filters), 1)
+ self.assertEqual("%s" % filtered,
+ "test.{filter_str}(elapsed_ms, 500)".format(
+ filter_str=filter_str))
+
+ def test_eq(self):
+ self._test_filter('eq')
+
+ def test_lt(self):
+ self._test_filter('lt')
+
+ def test_le(self):
+ self._test_filter('le')
+
+ def test_gt(self):
+ self._test_filter('gt')
+
+ def test_ge(self):
+ self._test_filter('ge')
+
+ def test_ne(self):
+ self._test_filter('ne')
+
+ def test_re(self):
+ self._test_filter('re')
+
+ def test_startswith(self):
+ e = EventExpression('test')
+ f = e.startswith('foo', 'bar')
+ self.assertEqual("%s" % f, 'test.re(foo, "^bar")')
+
+ def test_endswith(self):
+ e = EventExpression('test')
+ f = e.endswith('foo', 'bar')
+ self.assertEqual("%s" % f, 'test.re(foo, ".*bar$")')
+
+ def test_contains(self):
+ e = EventExpression('test')
+ f = e.contains('foo', 'bar')
+ self.assertEqual("%s" % f, 'test.re(foo, ".*bar.*")')
+
+ def test_in_array(self):
+ e = EventExpression('test')
+ f = e.in_array('foo', ['bar', 'baz'])
+ self.assertEqual("%s" % f, 'test.in(foo, ["bar", "baz"])')
+
+ f = e.in_array('foo', 'bar')
+ self.assertEqual("%s" % f, 'test.in(foo, ["b", "a", "r"])')
+
+ def test_filter_chaining(self):
+ e = EventExpression('test')
+ e = e.eq('bar', 'baz')
+ self.assertTrue(isinstance(e, EventExpression))
+ self.assertEqual(len(e.filters), 1)
+ e = e.lt('fizz', 'bang')
+ self.assertTrue(isinstance(e, EventExpression))
+ self.assertEqual(len(e.filters), 2)
+ e = e.ge('foo', 4)
+ self.assertTrue(isinstance(e, EventExpression))
+ self.assertEqual(len(e.filters), 3)
+ self.assertEqual("%s" % e,
+ 'test.eq(bar, "baz").lt(fizz, "bang").ge(foo, 4)')
@@ -0,0 +1,65 @@
+import unittest
+
+from cube.expression import (Distinct, EventExpression,
+ Max, Median, MetricExpression, Min, Sum)
+
+
+class TestMetricExpressions(unittest.TestCase):
+ def setUp(self):
+ self.e = EventExpression('request')
+
+ def _test_op(self, op_str, op):
+ m1 = MetricExpression(op_str, self.e)
+ m2 = op(self.e)
+ self.assertEqual(m1, m2)
+ s = "{op}(request)".format(op=op_str)
+ self.assertEqual("%s" % m1, s)
+ self.assertEqual("%s" % m2, s)
+
+ def test_sum(self):
+ self._test_op("sum", Sum)
+
+ def test_min(self):
+ self._test_op("min", Min)
+
+ def test_max(self):
+ self._test_op("max", Max)
+
+ def test_median(self):
+ self._test_op("median", Median)
+
+ def test_distinct(self):
+ self._test_op("distinct", Distinct)
+
+ def test_equality(self):
+ e1 = EventExpression('request')
+ m1 = MetricExpression('sum', e1)
+ m2 = MetricExpression('sum', e1)
+ self.assertEqual(m1, m2)
+
+ e2 = EventExpression('other')
+ m2 = MetricExpression('sum', e2)
+ self.assertNotEqual(m1, m2)
+
+ m1 = MetricExpression('sum', e2)
+ self.assertEqual(m1, m2)
+
+ m1 = MetricExpression('min', e2)
+ self.assertNotEqual(m1, m2)
+
+ m2 = MetricExpression('min', e2)
+ self.assertEqual(m1, m2)
+
+ def test_invalid_params(self):
+ self.assertRaisesRegexp(ValueError,
+ "Events for Metrics may only select a single event property",
+ Sum, EventExpression('request', ['path', 'user_id']))
+ self.assertRaises(TypeError, Sum)
+
+ def test_filters(self):
+ e1 = EventExpression('request', 'elapsed_ms').eq('path', '/')
+ e2 = e1.gt('elapsed_ms', 500)
+ self.assertEqual("%s" % Sum(e1),
+ "sum(request(elapsed_ms).eq(path, \"/\"))")
+ self.assertEqual("%s" % Sum(e2),
+ "sum(request(elapsed_ms).eq(path, \"/\").gt(elapsed_ms, 500))")

0 comments on commit d3715dd

Please sign in to comment.