#!/usr/bin/env python
"""Tests for grr_response_server.checks.filters."""
from __future__ import absolute_import
from __future__ import division
from __future__ import unicode_literals

import collections

from absl import app
from future.utils import iteritems
from future.utils import itervalues

from grr_response_core.lib.rdfvalues import anomaly as rdf_anomaly
from grr_response_core.lib.rdfvalues import client_fs as rdf_client_fs
from grr_response_core.lib.rdfvalues import paths as rdf_paths
from grr_response_core.lib.rdfvalues import protodict as rdf_protodict
from grr_response_server.check_lib import checks
from grr_response_server.check_lib import filters
from grr.test_lib import test_lib

# Just a named tuple that can be used to test objectfilter expressions.
Sample = collections.namedtuple("Sample", ["x", "y"])


class BaseFilterTests(test_lib.GRRBaseTest):
  """Test base filter methods and operations."""

  def testEnforceList(self):
    filt = filters.Filter()
    self.assertRaises(filters.ProcessingError, filt.Parse, "not_a_list", False)

  def testValidate(self):
    filt = filters.Filter()
    self.assertRaises(NotImplementedError, filt.Validate, "anything")

  def testParse(self):
    filt = filters.Filter()
    self.assertRaises(NotImplementedError, filt.Parse, [], "do nothing")


class AttrFilterTests(test_lib.GRRBaseTest):
  """Test attribute filter methods and operations."""

  def testValidate(self):
    filt = filters.AttrFilter()
    self.assertRaises(filters.DefinitionError, filt.Validate, " ")
    self.assertFalse(filt.Validate("cfg1"))
    self.assertFalse(filt.Validate("cfg1 cfg1.test1"))

  def testParse(self):
    filt = filters.AttrFilter()

    hit1 = rdf_protodict.AttributedDict(k1="hit1", k2="found1", k3=[3, 4])
    hit2 = rdf_protodict.AttributedDict(k1="hit2", k2="found2")
    meta = rdf_protodict.AttributedDict(one=hit1, two=hit2)
    objs = [hit1, hit2, meta]

    results = filt.Parse(objs, "k1 k2 one.k3")
    self.assertLen(results, 5)
    r1, r2, r3, r4, r5 = results
    self.assertEqual("k1", r1.key)
    self.assertEqual("hit1", r1.value)
    self.assertEqual("k1", r2.key)
    self.assertEqual("hit2", r2.value)
    self.assertEqual("k2", r3.key)
    self.assertEqual("found1", r3.value)
    self.assertEqual("k2", r4.key)
    self.assertEqual("found2", r4.value)
    self.assertEqual("one.k3", r5.key)
    self.assertEqual([3, 4], r5.value)


class ItemFilterTests(test_lib.GRRBaseTest):
  """Test item filter methods and operations."""

  def testParse(self):
    filt = filters.ItemFilter()

    one = rdf_protodict.AttributedDict(test1="1", test2=[2, 3])
    foo = rdf_protodict.AttributedDict(test1="foo", test2=["bar", "baz"])
    fs = rdf_client_fs.Filesystem(device="/dev/sda1", mount_point="/root")
    objs = [one, foo, fs]

    results = filt.Parse(objs, u"test1 is '1'")
    self.assertLen(results, 1)
    self.assertEqual("test1", results[0].key)
    self.assertEqual("1", results[0].value)

    results = filt.Parse(objs, u"test1 is '2'")
    self.assertFalse(results)

    results = filt.Parse(objs, u"test2 contains 3")
    self.assertLen(results, 1)
    self.assertEqual("test2", results[0].key)
    self.assertEqual([2, 3], results[0].value)

    results = filt.Parse(objs, u"test1 is '1' or test1 contains 'foo'")
    self.assertLen(results, 2)
    self.assertEqual("test1", results[0].key)
    self.assertEqual("1", results[0].value)
    self.assertEqual("test1", results[1].key)
    self.assertEqual("foo", results[1].value)

    results = filt.Parse(objs, u"mount_point is '/root'")
    self.assertLen(results, 1)
    self.assertEqual("mount_point", results[0].key)
    self.assertEqual("/root", results[0].value)


class ForEachTests(test_lib.GRRBaseTest):
  """Test ForEach filter methods and operations."""

  def testValidate(self):
    filt = filters.ForEach()
    self.assertRaises(filters.DefinitionError, filt.Validate, " ")
    self.assertRaises(filters.DefinitionError, filt.Validate, "attr1 attr2")
    self.assertFalse(filt.Validate("attr1"))

  def testParse(self):
    filt = filters.ForEach()

    hit1 = rdf_protodict.AttributedDict(k1="v1", k2="v2", k3="v3")
    hit2 = rdf_protodict.AttributedDict(k1="v4", k2="v5", k3="v6")
    meta = rdf_protodict.AttributedDict(
        foo=["foo", "bar"], target=[hit1, hit2], null=[])
    objs = [meta]

    results = filt.Parse(objs, "target")
    self.assertLen(results, 2)
    self.assertCountEqual([hit1, hit2], [r.item for r in results])

    results = filt.Parse(objs, "foo")
    self.assertLen(results, 2)
    self.assertCountEqual(["foo", "bar"], [r.item for r in results])

    results = filt.Parse(objs, "null")
    self.assertEmpty(results)