Add basic filter implementations and tests.

csv_object_reader/__init__.py

@@ -1,6 +1,9 @@
 from csv_object_reader.object_reader import ObjectReader
+from csv_object_reader.filter import Filter, RegexFilter
 
 __author__ = "fireyone29"
 __version__ = "0.1.1"
 
-__all__ = (ObjectReader.__name__,)
+__all__ = (ObjectReader.__name__,
+           Filter.__name__,
+           RegexFilter.__name__,)

csv_object_reader/filter.py

@@ -0,0 +1,64 @@
+"""Filter Objects"""
+
+import re
+
+
+class Filter(object):
+    """Basic Filter, uses == to compare values."""
+    def __init__(self, field, value=None, invert=False, missing_is_pass=False):
+        """
+        :param field: name of field to filter on
+        :param value: required value of the specified field, if None
+            checks that the field exists
+        :param invert: invert the results of the filter. Ignored if
+            value is None.
+        :param missing_is_pass: if True and field is not present,
+            pass this entry instead of failing it (the default).
+            Ignored if value is None.
+        """
+        if not isinstance(field, str):
+            raise TypeError
+        self._field = field
+        self._value = value
+        self._invert = invert
+        self._missing_is_pass = missing_is_pass
+
+    def _compare(self, value):
+        """Compare the expected value to the present value with ==."""
+        return self._value == value
+
+    def test(self, entry):
+        """
+        Test the provided entry against this filer.
+
+        :param entry: entry to test against the filter.
+        :returns: True if the entry passes the filter, otherwise False.
+        """
+        value = None
+        missing = False
+        try:
+            value = getattr(entry, self._field)
+        except AttributeError:
+            missing = True
+        if missing:
+            if self._value is None:
+                return False
+            else:
+                return self._missing_is_pass
+        elif self._value is None:
+            return True
+        else:
+            return self._compare(value) ^ self._invert
+
+
+class RegexFilter(Filter):
+    """Filter which uses regex match to compare values."""
+    def __init__(self, field, value=None, invert=False, missing_is_pass=False):
+        super(RegexFilter, self).__init__(field, value, invert,
+                                          missing_is_pass)
+        if self._value is not None:
+            self._regex = re.compile(self._value)
+
+    def _compare(self, value):
+        """Treat the expected value as a regex expression to match."""
+        return bool(self._regex.match(str(value)))

docs/api/csv_object_reader.filter.rst

@@ -0,0 +1,7 @@
+csv_object_reader.filter module
+===============================
+
+.. automodule:: csv_object_reader.filter
+    :members:
+    :undoc-members:
+    :show-inheritance:

docs/api/csv_object_reader.rst

@@ -6,6 +6,7 @@ Submodules
 
 .. toctree::
 
+   csv_object_reader.filter
    csv_object_reader.object_reader
 
 Module contents

tests/test_filter.py

@@ -0,0 +1,132 @@
+import unittest
+from collections import namedtuple
+from csv_object_reader import Filter, RegexFilter
+
+
+class TestFilterValueNone(unittest.TestCase):
+    def setUp(self):
+        super(TestFilterValueNone, self).setUp()
+        self.field = "field1"
+        self.filter1 = Filter(self.field)
+
+    def test_field_present(self):
+        Entry = namedtuple("Entry", [self.field])
+        entry = Entry("a")
+        self.assertTrue(self.filter1.test(entry))
+
+    def test_field_missing(self):
+        Entry = namedtuple("Entry", [self.field + "a"])
+        entry = Entry("a")
+        self.assertFalse(self.filter1.test(entry))
+
+
+class TestFilterValueNoneInvert(TestFilterValueNone):
+    def setUp(self):
+        super(TestFilterValueNoneInvert, self).setUp()
+        self.filter1 = Filter(self.field, invert=True)
+
+
+class TestFilterValueNoneMissingPass(TestFilterValueNone):
+    def setUp(self):
+        super(TestFilterValueNoneMissingPass, self).setUp()
+        self.filter1 = Filter(self.field, missing_is_pass=True)
+
+
+class TestRegexFilterValueNone(TestFilterValueNone):
+    def setUp(self):
+        super(TestRegexFilterValueNone, self).setUp()
+        self.filter1 = RegexFilter(self.field)
+
+
+class TestRegexFilterValueNonePassMissing(TestFilterValueNone):
+    def setUp(self):
+        super(TestRegexFilterValueNonePassMissing, self).setUp()
+        self.filter1 = RegexFilter(self.field, missing_is_pass=True)
+
+
+class TestRegexFilterValueNoneInvert(TestFilterValueNone):
+    def setUp(self):
+        super(TestRegexFilterValueNoneInvert, self).setUp()
+        self.filter1 = RegexFilter(self.field, invert=True)
+
+
+class TestFilter(unittest.TestCase):
+    def setUp(self):
+        super(TestFilter, self).setUp()
+        self.field = "field1"
+        self.pass_value = 2
+        self.fail_value = "2"
+        self.filter1 = Filter(self.field, self.pass_value)
+
+    def test_pass(self):
+        Entry = namedtuple("Entry", [self.field])
+        entry = Entry(self.pass_value)
+        self.assertTrue(self.filter1.test(entry))
+
+    def test_fail(self):
+        Entry = namedtuple("Entry", [self.field])
+        entry = Entry(self.fail_value)
+        self.assertFalse(self.filter1.test(entry))
+
+    def test_field_missing(self):
+        Entry = namedtuple("Entry", [self.field + "a"])
+        entry = Entry(self.pass_value)
+        self.assertFalse(self.filter1.test(entry))
+
+
+class TestFilterInvert(TestFilter):
+    def setUp(self):
+        super(TestFilterInvert, self).setUp()
+        self.pass_value = "2"
+        self.fail_value = 2
+        self.filter1 = Filter(self.field, self.fail_value, invert=True)
+
+
+class TestFilterMissingPass(TestFilter):
+    def setUp(self):
+        super(TestFilterMissingPass, self).setUp()
+        self.filter1 = Filter(self.field, self.pass_value,
+                              missing_is_pass=True)
+
+    def test_field_missing(self):
+        Entry = namedtuple("Entry", [self.field + "a"])
+        entry = Entry(self.pass_value)
+        self.assertTrue(self.filter1.test(entry))
+
+
+class TestRegexFilterInvalidPattern(unittest.TestCase):
+    def test_invalid_regex_pattern(self):
+        with self.assertRaises(TypeError):
+            RegexFilter("x", 2)
+
+    def test_invalid_field(self):
+        with self.assertRaises(TypeError):
+            RegexFilter(2, "2")
+
+
+class TestRegexFilter(TestFilter):
+    def setUp(self):
+        super(TestRegexFilter, self).setUp()
+        self.pass_value = "val1"
+        self.fail_value = "1val"
+        self.pattern = "val[0-9]"
+        self.filter1 = RegexFilter(self.field, self.pattern)
+
+
+class TestRegexFilterInvert(TestRegexFilter):
+    def setUp(self):
+        super(TestRegexFilterInvert, self).setUp()
+        self.pattern = "[0-9]val"
+        self.filter1 = RegexFilter(self.field, self.pattern, invert=True)
+
+
+class TestRegexFilterMissingPass(TestRegexFilter):
+    def setUp(self):
+        super(TestRegexFilterMissingPass, self).setUp()
+        self.filter1 = RegexFilter(self.field, self.pattern,
+                                   missing_is_pass=True)
+
+    def test_field_missing(self):
+        Entry = namedtuple("Entry", [self.field + "a"])
+        entry = Entry(self.pass_value)
+        self.assertTrue(self.filter1.test(entry))