#!/usr/bin/env python
import hashlib
import os

from grr_response_core.lib import rdfvalue
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_proto import jobs_pb2
from grr_response_proto import objects_pb2
from grr_response_server.databases import db
from grr_response_server.databases import db_test_utils
from grr_response_server.rdfvalues import mig_objects
from grr_response_server.rdfvalues import objects as rdf_objects


class DatabaseTestPathsMixin(object):
  """An abstract class for testing db.Database implementations.

  This mixin adds methods to test the handling of GRR path data.
  """

  def testWritePathInfosValidatesClientId(self):
    path = ["usr", "local"]

    with self.assertRaises(ValueError):
      self.db.WritePathInfos(
          "",
          [
              objects_pb2.PathInfo(
                  path_type=objects_pb2.PathInfo.PathType.OS, components=path
              )
          ],
      )

  def testWritePathInfosValidatesPathType(self):
    path = ["usr", "local"]
    client_id = db_test_utils.InitializeClient(self.db)

    with self.assertRaises(ValueError):
      self.db.WritePathInfos(client_id, [objects_pb2.PathInfo(components=path)])

  def testWritePathInfosValidatesClient(self):
    client_id = "C.0123456789012345"

    with self.assertRaises(db.UnknownClientError) as context:
      self.db.WritePathInfos(
          client_id,
          [
              objects_pb2.PathInfo(
                  path_type=objects_pb2.PathInfo.PathType.OS,
                  components=[],
                  directory=True,
              )
          ],
      )

    self.assertEqual(context.exception.client_id, client_id)

  def testWritePathInfosValidatesClientEvenIfEmpty(self):
    client_id = "C.0123456789012345"

    with self.assertRaises(db.UnknownClientError) as context:
      self.db.WritePathInfos(client_id, [])

    self.assertEqual(context.exception.client_id, client_id)

  def testWritePathInfosValidateConflictingWrites(self):
    client_id = db_test_utils.InitializeClient(self.db)

    with self.assertRaises(ValueError):
      self.db.WritePathInfos(
          client_id,
          [
              objects_pb2.PathInfo(
                  path_type=objects_pb2.PathInfo.PathType.OS,
                  components=["foo", "bar"],
                  directory=False,
              ),
              objects_pb2.PathInfo(
                  path_type=objects_pb2.PathInfo.PathType.OS,
                  components=["foo", "bar"],
                  directory=True,
              ),
          ],
      )

  def testWritePathInfosEmpty(self):
    client_id = db_test_utils.InitializeClient(self.db)
    self.db.WritePathInfos(client_id, [])  # Should not raise.

  def testWritePathInfosMetadata(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.TSK,
                components=["foo", "bar"],
                directory=True,
            )
        ],
    )

    results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.TSK, [("foo", "bar")]
    )

    result_path_info = results[("foo", "bar")]
    self.assertEqual(
        result_path_info.path_type, objects_pb2.PathInfo.PathType.TSK
    )
    self.assertEqual(result_path_info.components, ["foo", "bar"])
    self.assertEqual(result_path_info.directory, True)

  def testWritePathInfosMetadataTimestampUpdate(self):
    now = self.db.Now

    client_id = db_test_utils.InitializeClient(self.db)

    timestamp_0 = now()

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
            )
        ],
    )

    result = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(result.components, ["foo"])
    self.assertGreater(result.timestamp, timestamp_0)
    self.assertLess(result.timestamp, now())
    self.assertFalse(result.HasField("last_stat_entry_timestamp"))
    self.assertFalse(result.HasField("last_hash_entry_timestamp"))

    timestamp_1 = now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
    )
    path_info.stat_entry.st_mode = 42
    self.db.WritePathInfos(client_id, [path_info])

    result = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(result.components, ["foo"])
    self.assertEqual(result.stat_entry.st_mode, 42)
    self.assertGreater(result.timestamp, timestamp_1)
    self.assertLess(result.timestamp, now())
    self.assertGreater(result.last_stat_entry_timestamp, timestamp_1)
    self.assertLess(result.last_stat_entry_timestamp, now())

    timestamp_2 = now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
    )
    path_info.hash_entry.sha256 = b"foo"
    self.db.WritePathInfos(client_id, [path_info])

    result = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(result.components, ["foo"])
    self.assertEqual(result.hash_entry.sha256, b"foo")
    self.assertGreater(result.timestamp, timestamp_2)
    self.assertLess(result.timestamp, now())
    self.assertGreater(result.last_hash_entry_timestamp, timestamp_2)
    self.assertLess(result.last_hash_entry_timestamp, now())

    timestamp_3 = now()

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo"],
                directory=True,
            )
        ],
    )

    result = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(result.components, ["foo"])
    self.assertEqual(result.stat_entry.st_mode, 42)
    self.assertEqual(result.hash_entry.sha256, b"foo")
    self.assertTrue(result.directory)
    self.assertGreater(result.timestamp, timestamp_3)
    self.assertLess(result.timestamp, now())
    self.assertGreater(result.last_stat_entry_timestamp, timestamp_1)
    self.assertLess(result.last_stat_entry_timestamp, timestamp_2)
    self.assertGreater(result.last_hash_entry_timestamp, timestamp_2)
    self.assertLess(result.last_hash_entry_timestamp, timestamp_3)

    timestamp_4 = now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )
    path_info.stat_entry.st_mode = 108
    path_info.hash_entry.sha256 = b"norf"
    self.db.WritePathInfos(client_id, [path_info])

    result = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(result.components, ["foo"])
    self.assertEqual(result.stat_entry.st_mode, 108)
    self.assertEqual(result.hash_entry.sha256, b"norf")
    self.assertGreater(result.timestamp, timestamp_4)
    self.assertGreater(result.last_stat_entry_timestamp, timestamp_4)
    self.assertGreater(result.last_hash_entry_timestamp, timestamp_4)
    self.assertLess(result.timestamp, now())
    self.assertLess(result.last_stat_entry_timestamp, now())
    self.assertLess(result.last_hash_entry_timestamp, now())

  def testWritePathInfosStatEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    stat_entry = rdf_client_fs.StatEntry()
    stat_entry.pathspec.path = "foo/bar"
    stat_entry.pathspec.pathtype = rdf_paths.PathSpec.PathType.OS
    stat_entry.st_mode = 1337
    stat_entry.st_mtime = 108
    stat_entry.st_atime = 4815162342

    path_info = rdf_objects.PathInfo.FromStatEntry(stat_entry)
    proto_path_info = mig_objects.ToProtoPathInfo(path_info)
    self.db.WritePathInfos(client_id, [proto_path_info])

    results = self.db.ReadPathInfos(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        [
            (),
            ("foo",),
            ("foo", "bar"),
        ],
    )

    root_path_info = results[()]
    self.assertFalse(root_path_info.HasField("stat_entry"))

    foo_path_info = results[("foo",)]
    self.assertFalse(foo_path_info.HasField("stat_entry"))

    foobar_path_info = results[("foo", "bar")]
    self.assertTrue(foobar_path_info.HasField("stat_entry"))
    self.assertFalse(foobar_path_info.HasField("hash_entry"))
    self.assertEqual(foobar_path_info.stat_entry.st_mode, 1337)
    self.assertEqual(foobar_path_info.stat_entry.st_mtime, 108)
    self.assertEqual(foobar_path_info.stat_entry.st_atime, 4815162342)

  def testWritePathInfosHashEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    hash_entry = jobs_pb2.Hash()
    hash_entry.sha256 = hashlib.sha256(b"foo").digest()
    hash_entry.md5 = hashlib.md5(b"foo").digest()
    hash_entry.num_bytes = len(b"foo")

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "bar", "baz"],
    )
    path_info.hash_entry.CopyFrom(hash_entry)
    self.db.WritePathInfos(client_id, [path_info])

    result = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )

    self.assertEqual(result.components, ["foo", "bar", "baz"])
    self.assertTrue(result.HasField("hash_entry"))
    self.assertFalse(result.HasField("stat_entry"))
    self.assertEqual(result.hash_entry.sha256, hashlib.sha256(b"foo").digest())
    self.assertEqual(result.hash_entry.md5, hashlib.md5(b"foo").digest())
    self.assertEqual(result.hash_entry.num_bytes, len(b"foo"))

  def testWritePathInfosValidatesHashEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    hash_entry = jobs_pb2.Hash()
    hash_entry.md5 = hashlib.md5(b"foo").digest()
    hash_entry.sha1 = hashlib.sha1(b"bar").digest()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )
    path_info.hash_entry.CopyFrom(hash_entry)

    with self.assertRaises(ValueError):
      self.db.WritePathInfos(client_id, [path_info])

  def testWriteMultiplePathInfosHashEntry(self):

    def SHA256(data: bytes) -> bytes:
      return hashlib.sha256(data).digest()

    def MD5(data: bytes) -> bytes:
      return hashlib.md5(data).digest()

    client_id = db_test_utils.InitializeClient(self.db)

    files = {
        "foo": b"4815162342",
        "BAR": b"\xff\x00\xff",
        "bAz": b"\x00" * 42,
        "żółć": "Wpłynąłem na suchego przestwór oceanu".encode("utf-8"),
    }

    path_infos = []
    for name, content in files.items():
      content = name.encode("utf-8")

      hash_entry = jobs_pb2.Hash()
      hash_entry.sha256 = SHA256(content)
      hash_entry.md5 = MD5(content)
      hash_entry.num_bytes = len(content)

      path_info = objects_pb2.PathInfo(
          path_type=objects_pb2.PathInfo.PathType.OS,
          components=["foo", "bar", "baz", name],
      )
      path_info.hash_entry.CopyFrom(hash_entry)
      path_infos.append(path_info)

    self.db.WritePathInfos(client_id, path_infos)

    for name, content in files.items():
      content = name.encode("utf-8")

      result = self.db.ReadPathInfo(
          client_id,
          objects_pb2.PathInfo.PathType.OS,
          components=("foo", "bar", "baz", name),
      )

      self.assertEqual(result.components, ["foo", "bar", "baz", name])
      self.assertTrue(result.HasField("hash_entry"))
      self.assertFalse(result.HasField("stat_entry"))
      self.assertEqual(result.hash_entry.sha256, SHA256(content))
      self.assertEqual(result.hash_entry.md5, MD5(content))
      self.assertEqual(result.hash_entry.num_bytes, len(content))

  def testWritePathInfosHashAndStatEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    stat_entry = jobs_pb2.StatEntry(st_mode=1337)
    hash_entry = jobs_pb2.Hash(sha256=hashlib.sha256(b"foo").digest())

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "bar", "baz"],
    )
    path_info.stat_entry.CopyFrom(stat_entry)
    path_info.hash_entry.CopyFrom(hash_entry)
    self.db.WritePathInfos(client_id, [path_info])

    result = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )

    self.assertEqual(result.components, ["foo", "bar", "baz"])
    self.assertTrue(result.HasField("stat_entry"))
    self.assertTrue(result.HasField("hash_entry"))
    self.assertEqual(result.stat_entry, stat_entry)
    self.assertEqual(result.hash_entry, hash_entry)

  def testWritePathInfoHashAndStatEntrySeparateWrites(self):
    client_id = db_test_utils.InitializeClient(self.db)

    stat_entry = jobs_pb2.StatEntry(st_mode=1337)
    stat_entry_path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
    )
    stat_entry_path_info.stat_entry.CopyFrom(stat_entry)

    stat_entry_timestamp = self.db.Now()
    self.db.WritePathInfos(client_id, [stat_entry_path_info])

    hash_entry = jobs_pb2.Hash(sha256=hashlib.sha256(b"foo").digest())
    hash_entry_path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
    )
    hash_entry_path_info.hash_entry.CopyFrom(hash_entry)

    hash_entry_timestamp = self.db.Now()
    self.db.WritePathInfos(client_id, [hash_entry_path_info])

    result = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )

    now = self.db.Now()

    self.assertEqual(result.components, ["foo"])
    self.assertTrue(result.HasField("stat_entry"))
    self.assertTrue(result.HasField("hash_entry"))
    self.assertEqual(result.stat_entry, stat_entry)
    self.assertEqual(result.hash_entry, hash_entry)
    self.assertGreater(result.last_stat_entry_timestamp, stat_entry_timestamp)
    self.assertLess(result.last_stat_entry_timestamp, hash_entry_timestamp)
    self.assertGreater(result.last_hash_entry_timestamp, hash_entry_timestamp)
    self.assertLess(result.last_hash_entry_timestamp, now)

  def testWritePathInfosExpansion(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz"],
            ),
        ],
    )

    results = self.db.ReadPathInfos(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        [
            ("foo",),
            ("foo", "bar"),
            ("foo", "bar", "baz"),
        ],
    )

    self.assertLen(results, 3)

    foo = results[("foo",)]
    self.assertEqual(foo.components, ["foo"])
    self.assertTrue(foo.directory)

    foobar = results[("foo", "bar")]
    self.assertEqual(foobar.components, ["foo", "bar"])
    self.assertTrue(foobar.directory)

    foobarbaz = results[("foo", "bar", "baz")]
    self.assertEqual(foobarbaz.components, ["foo", "bar", "baz"])
    self.assertFalse(foobarbaz.directory)

  def testWritePathInfosTypeSeparated(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo"],
                directory=True,
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.TSK,
                components=["foo"],
                directory=False,
            ),
        ],
    )

    os_results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo",)]
    )
    self.assertLen(os_results, 1)
    self.assertTrue(os_results[("foo",)].directory)

    tsk_results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.TSK, [("foo",)]
    )
    self.assertLen(tsk_results, 1)
    self.assertFalse(tsk_results[("foo",)].directory)

  def testWritePathInfosUpdates(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz"],
                directory=False,
            ),
        ],
    )

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz"],
                directory=True,
            ),
        ],
    )

    results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo", "bar", "baz")]
    )

    result_path_info = results[("foo", "bar", "baz")]
    self.assertTrue(result_path_info.directory)

  def testWritePathInfosUpdatesAncestors(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo"],
                directory=False,
            ),
        ],
    )
    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
        ],
    )

    results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo",)]
    )

    self.assertLen(results, 1)
    self.assertTrue(results[("foo",)].directory)

  def testWritePathInfosWritesAncestorsWithTimestamps(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["ancestor", "bar"],
            ),
        ],
    )

    results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, [("ancestor",)]
    )

    self.assertLen(results, 1)
    self.assertIsNotNone(results[("ancestor",)].timestamp)

  def testWritePathInfosDuplicatedData(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
        ],
    )
    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
        ],
    )

    results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo", "bar")]
    )
    self.assertLen(results, 1)

    result_path_info = results[("foo", "bar")]
    self.assertEqual(result_path_info.components, ["foo", "bar"])
    self.assertEqual(result_path_info.directory, False)

  def testWritePathInfosStoresCopy(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo", "bar"]
    )

    path_info.stat_entry.st_size = 1337
    path_info.hash_entry.sha256 = b"foo"
    self.db.WritePathInfos(client_id, [path_info])

    timestamp_1 = self.db.Now()

    path_info.stat_entry.st_size = 42
    path_info.hash_entry.sha256 = b"bar"
    self.db.WritePathInfos(client_id, [path_info])

    timestamp_2 = self.db.Now()

    result_1 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar"),
        timestamp=timestamp_1,
    )
    self.assertEqual(result_1.stat_entry.st_size, 1337)
    self.assertEqual(result_1.hash_entry.sha256, b"foo")

    result_2 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar"),
        timestamp=timestamp_2,
    )
    self.assertEqual(result_2.stat_entry.st_size, 42)
    self.assertEqual(result_2.hash_entry.sha256, b"bar")

  def testReadPathInfosEmptyComponentsList(self):
    client_id = db_test_utils.InitializeClient(self.db)
    results = self.db.ReadPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, []
    )
    self.assertEqual(results, {})

  def testReadPathInfosNonExistent(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
        ],
    )

    results = self.db.ReadPathInfos(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        [
            ("foo", "bar"),
            ("foo", "baz"),
            ("quux", "norf"),
        ],
    )
    self.assertLen(results, 3)
    self.assertIsNotNone(results[("foo", "bar")])
    self.assertIsNone(results[("foo", "baz")])
    self.assertIsNone(results[("quux", "norf")])

  def testReadPathInfoValidatesTimestamp(self):
    client_id = db_test_utils.InitializeClient(self.db)

    with self.assertRaises(TypeError):
      self.db.ReadPathInfo(
          client_id,
          objects_pb2.PathInfo.PathType.REGISTRY,
          components=("foo", "bar", "baz"),
          timestamp=rdfvalue.Duration.From(10, rdfvalue.SECONDS),
      )

  def testReadPathInfoNonExistent(self):
    client_id = db_test_utils.InitializeClient(self.db)

    with self.assertRaises(db.UnknownPathError) as ctx:
      self.db.ReadPathInfo(
          client_id,
          objects_pb2.PathInfo.PathType.OS,
          components=("foo", "bar", "baz"),
      )

    self.assertEqual(ctx.exception.client_id, client_id)
    self.assertEqual(ctx.exception.path_type, objects_pb2.PathInfo.PathType.OS)
    self.assertEqual(ctx.exception.components, ("foo", "bar", "baz"))

  def testReadPathInfoTimestampStatEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    pathspec = rdf_paths.PathSpec(
        path="foo/bar/baz", pathtype=rdf_paths.PathSpec.PathType.OS
    )

    stat_entry = rdf_client_fs.StatEntry(pathspec=pathspec, st_size=42)
    path_info = rdf_objects.PathInfo.FromStatEntry(stat_entry)
    self.db.WritePathInfos(client_id, [mig_objects.ToProtoPathInfo(path_info)])
    timestamp_1 = self.db.Now()

    stat_entry = rdf_client_fs.StatEntry(pathspec=pathspec, st_size=101)
    path_info = rdf_objects.PathInfo.FromStatEntry(stat_entry)
    self.db.WritePathInfos(client_id, [mig_objects.ToProtoPathInfo(path_info)])
    timestamp_2 = self.db.Now()

    stat_entry = rdf_client_fs.StatEntry(pathspec=pathspec, st_size=1337)
    path_info = rdf_objects.PathInfo.FromStatEntry(stat_entry)
    self.db.WritePathInfos(client_id, [mig_objects.ToProtoPathInfo(path_info)])
    timestamp_3 = self.db.Now()

    path_info_last = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )
    self.assertEqual(path_info_last.stat_entry.st_size, 1337)
    self.assertEqual(path_info_last.components, ["foo", "bar", "baz"])

    path_info_1 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
        timestamp=timestamp_1,
    )
    self.assertEqual(path_info_1.stat_entry.st_size, 42)
    self.assertEqual(path_info_last.components, ["foo", "bar", "baz"])

    path_info_2 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
        timestamp=timestamp_2,
    )
    self.assertEqual(path_info_2.stat_entry.st_size, 101)
    self.assertEqual(path_info_last.components, ["foo", "bar", "baz"])

    path_info_3 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
        timestamp=timestamp_3,
    )
    self.assertEqual(path_info_3.stat_entry.st_size, 1337)
    self.assertEqual(path_info_last.components, ["foo", "bar", "baz"])

  def testReadPathInfoTimestampHashEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )

    path_info.hash_entry.sha256 = b"bar"
    self.db.WritePathInfos(client_id, [path_info])
    bar_timestamp = self.db.Now()

    path_info.hash_entry.sha256 = b"baz"
    self.db.WritePathInfos(client_id, [path_info])
    baz_timestamp = self.db.Now()

    path_info.hash_entry.sha256 = b"quux"
    self.db.WritePathInfos(client_id, [path_info])
    quux_timestamp = self.db.Now()

    bar_path_info = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=bar_timestamp,
    )
    self.assertEqual(bar_path_info.hash_entry.sha256, b"bar")

    baz_path_info = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=baz_timestamp,
    )
    self.assertEqual(baz_path_info.hash_entry.sha256, b"baz")

    quux_path_info = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=quux_timestamp,
    )
    self.assertEqual(quux_path_info.hash_entry.sha256, b"quux")

  def testReadPathInfosMany(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo", "bar"]
    )
    path_info_1.stat_entry.st_mode = 42
    path_info_1.hash_entry.md5 = b"foo"
    path_info_1.hash_entry.sha256 = b"bar"

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["baz", "quux", "norf"],
    )
    path_info_2.hash_entry.sha256 = b"bazquuxnorf"

    path_info_3 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["blargh"],
        directory=True,
    )
    path_info_3.stat_entry.st_size = 1337

    self.db.WritePathInfos(client_id, [path_info_1, path_info_2, path_info_3])

    results = self.db.ReadPathInfos(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        [
            ("foo", "bar"),
            ("baz", "quux", "norf"),
            ("blargh",),
        ],
    )
    result_path_info_1 = results[("foo", "bar")]
    self.assertEqual(result_path_info_1.components, ["foo", "bar"])
    self.assertEqual(result_path_info_1.stat_entry.st_mode, 42)
    self.assertEqual(result_path_info_1.hash_entry.md5, b"foo")
    self.assertEqual(result_path_info_1.hash_entry.sha256, b"bar")

    result_path_info_2 = results[("baz", "quux", "norf")]
    self.assertEqual(result_path_info_2.components, ["baz", "quux", "norf"])
    self.assertEqual(result_path_info_2.hash_entry.sha256, b"bazquuxnorf")

    result_path_info_3 = results[("blargh",)]
    self.assertEqual(result_path_info_3.components, ["blargh"])
    self.assertEqual(result_path_info_3.stat_entry.st_size, 1337)
    self.assertEqual(result_path_info_3.directory, True)

  def testReadPathInfoTimestampStatAndHashEntry(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )

    path_info.stat_entry.st_mode = 42
    path_info.ClearField("hash_entry")
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_1 = self.db.Now()

    path_info.ClearField("stat_entry")
    path_info.hash_entry.sha256 = b"quux"
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_2 = self.db.Now()

    path_info.stat_entry.st_mode = 1337
    path_info.ClearField("hash_entry")
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_3 = self.db.Now()

    path_info.stat_entry.st_mode = 4815162342
    path_info.hash_entry.sha256 = b"norf"
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_4 = self.db.Now()

    path_info_1 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_1,
    )
    self.assertEqual(path_info_1.stat_entry.st_mode, 42)
    self.assertFalse(path_info_1.HasField("hash_entry"))

    path_info_2 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_2,
    )
    self.assertEqual(path_info_2.stat_entry.st_mode, 42)
    self.assertEqual(path_info_2.hash_entry.sha256, b"quux")

    path_info_3 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_3,
    )
    self.assertEqual(path_info_3.stat_entry.st_mode, 1337)
    self.assertEqual(path_info_3.hash_entry.sha256, b"quux")

    path_info_4 = self.db.ReadPathInfo(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_4,
    )
    self.assertEqual(path_info_4.stat_entry.st_mode, 4815162342)
    self.assertEqual(path_info_4.hash_entry.sha256, b"norf")

  def testReadPathInfoOlder(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )
    path_info.stat_entry.st_mode = 42
    path_info.hash_entry.sha256 = b"foo"
    self.db.WritePathInfos(client_id, [path_info])

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["bar"]
    )
    path_info.stat_entry.st_mode = 1337
    path_info.hash_entry.sha256 = b"bar"
    self.db.WritePathInfos(client_id, [path_info])

    path_info = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(path_info.stat_entry.st_mode, 42)
    self.assertEqual(path_info.hash_entry.sha256, b"foo")

    path_info = self.db.ReadPathInfo(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("bar",)
    )
    self.assertEqual(path_info.stat_entry.st_mode, 1337)
    self.assertEqual(path_info.hash_entry.sha256, b"bar")

  def testListDescendantPathInfosAlwaysSucceedsOnRoot(self):
    client_id = db_test_utils.InitializeClient(self.db)

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=()
    )

    self.assertEmpty(results)

  def testListDescendantPathInfosAlwaysSucceedsOnRoot_ListVersion(self):
    client_id = db_test_utils.InitializeClient(self.db)

    results = self.db.ListDescendantPathInfos(
        client_id, rdf_objects.PathInfo.PathType.OS, components=[]
    )

    self.assertEmpty(results)

  def testListDescendantPathInfosNonexistentDirectory(self):
    client_id = db_test_utils.InitializeClient(self.db)

    with self.assertRaises(db.UnknownPathError):
      self.db.ListDescendantPathInfos(
          client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
      )

  def testListDescendantPathInfosNotDirectory(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        directory=False,
    )
    self.db.WritePathInfos(client_id, [path_info])

    with self.assertRaises(db.NotDirectoryPathError):
      self.db.ListDescendantPathInfos(
          client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
      )

  def testListDescendantPathInfosEmptyResult(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        directory=True,
    )
    self.db.WritePathInfos(client_id, [path_info])

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )

    self.assertEmpty(results)

  def testListDescendantPathInfosSingleResult(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )

    self.assertLen(results, 1)
    self.assertEqual(results[0].components, ["foo", "bar"])

  def testListDescendantPathInfosSingle(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz", "quux"],
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )

    self.assertLen(results, 3)
    self.assertEqual(results[0].components, ["foo", "bar"])
    self.assertEqual(results[1].components, ["foo", "bar", "baz"])
    self.assertEqual(results[2].components, ["foo", "bar", "baz", "quux"])

  def testListDescendantPathInfosBranching(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "quux"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "baz"],
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )

    self.assertLen(results, 3)
    self.assertEqual(results[0].components, ["foo", "bar"])
    self.assertEqual(results[1].components, ["foo", "bar", "quux"])
    self.assertEqual(results[2].components, ["foo", "baz"])

  def testListDescendantPathInfosLimited(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz", "quux"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "blargh"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "norf", "thud", "plugh"],
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        max_depth=2,
    )

    components = [tuple(path_info.components) for path_info in results]

    self.assertIn(("foo", "bar"), components)
    self.assertIn(("foo", "bar", "baz"), components)
    self.assertIn(("foo", "bar", "blargh"), components)

    self.assertNotIn(("foo", "bar", "baz", "quux"), components)
    self.assertNotIn(("foo", "norf", "thud", "plugh"), components)

  def testListDescendantPathInfosTypeSeparated(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["usr", "bin", "javac"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.TSK,
                components=["usr", "bin", "gdb"],
            ),
        ],
    )

    os_results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("usr", "bin")
    )
    self.assertLen(os_results, 1)
    self.assertEqual(os_results[0].components, ["usr", "bin", "javac"])

    tsk_results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.TSK, components=("usr", "bin")
    )
    self.assertLen(tsk_results, 1)
    self.assertEqual(tsk_results[0].components, ["usr", "bin", "gdb"])

  def testListDescendantPathInfosAll(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["baz", "quux"],
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=()
    )

    self.assertEqual(
        results[0].components,
        [
            "baz",
        ],
    )
    self.assertEqual(results[1].components, ["baz", "quux"])
    self.assertEqual(
        results[2].components,
        [
            "foo",
        ],
    )
    self.assertEqual(results[3].components, ["foo", "bar"])

  def testListDescendantPathInfosLimitedDirectory(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "bar", "baz"],
    )
    path_info_1.stat_entry.st_mode = 108

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo", "bar"]
    )
    path_info_2.stat_entry.st_mode = 1337

    path_info_3 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "norf", "quux"],
    )
    path_info_3.stat_entry.st_mode = 707

    self.db.WritePathInfos(client_id, [path_info_1, path_info_2, path_info_3])

    results = self.db.ListDescendantPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=(), max_depth=2
    )

    self.assertLen(results, 3)
    self.assertEqual(
        results[0].components,
        [
            "foo",
        ],
    )
    self.assertEqual(results[1].components, ["foo", "bar"])
    self.assertEqual(results[2].components, ["foo", "norf"])
    self.assertEqual(results[1].stat_entry.st_mode, 1337)

  def testListDescendantPathInfosDepthZero(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_3 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("baz",)
    )

    self.db.WritePathInfos(client_id, [path_info_1, path_info_2, path_info_3])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        max_depth=0,
    )
    self.assertEmpty(results)

  def testListDescendantPathInfosTimestampNow(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "bar", "baz"],
    )
    path_info.stat_entry.st_size = 1337
    self.db.WritePathInfos(client_id, [path_info])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=self.db.Now(),
    )

    self.assertLen(results, 3)
    self.assertEqual(
        results[0].components,
        [
            "foo",
        ],
    )
    self.assertEqual(results[1].components, ["foo", "bar"])
    self.assertEqual(results[2].components, ["foo", "bar", "baz"])
    self.assertEqual(results[2].stat_entry.st_size, 1337)

  def testListDescendantPathInfosTimestampMultiple(self):
    client_id = db_test_utils.InitializeClient(self.db)

    timestamp_0 = self.db.Now()

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "bar", "baz"],
    )
    path_info_1.stat_entry.st_size = 1
    self.db.WritePathInfos(client_id, [path_info_1])
    timestamp_1 = self.db.Now()

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "quux", "norf"],
    )
    path_info_2.stat_entry.st_size = 2
    self.db.WritePathInfos(client_id, [path_info_2])
    timestamp_2 = self.db.Now()

    path_info_3 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo", "quux", "thud"],
    )
    path_info_3.stat_entry.st_size = 3
    self.db.WritePathInfos(client_id, [path_info_3])
    timestamp_3 = self.db.Now()

    results_0 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_0,
    )
    self.assertEmpty(results_0)

    results_1 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_1,
    )
    self.assertLen(results_1, 3)
    self.assertEqual(results_1[0].components, ["foo"])
    self.assertEqual(results_1[1].components, ["foo", "bar"])
    self.assertEqual(results_1[2].components, ["foo", "bar", "baz"])

    results_2 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_2,
    )
    self.assertLen(results_2, 5)
    self.assertEqual(results_2[0].components, ["foo"])
    self.assertEqual(results_2[1].components, ["foo", "bar"])
    self.assertEqual(results_2[2].components, ["foo", "bar", "baz"])
    self.assertEqual(results_2[3].components, ["foo", "quux"])
    self.assertEqual(results_2[4].components, ["foo", "quux", "norf"])

    results_3 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_3,
    )
    self.assertLen(results_3, 6)
    self.assertEqual(results_3[0].components, ["foo"])
    self.assertEqual(results_3[1].components, ["foo", "bar"])
    self.assertEqual(results_3[2].components, ["foo", "bar", "baz"])
    self.assertEqual(results_3[3].components, ["foo", "quux"])
    self.assertEqual(results_3[4].components, ["foo", "quux", "norf"])
    self.assertEqual(results_3[5].components, ["foo", "quux", "thud"])

  def testListDescendantPathInfosTimestampStatValue(self):
    client_id = db_test_utils.InitializeClient(self.db)

    timestamp_0 = self.db.Now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )

    path_info.stat_entry.st_size = 1337
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_1 = self.db.Now()

    path_info.stat_entry.st_size = 42
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_2 = self.db.Now()

    results_0 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_0,
    )
    self.assertEmpty(results_0)

    results_1 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_1,
    )
    self.assertLen(results_1, 1)
    self.assertEqual(results_1[0].components, ["foo", "bar"])
    self.assertEqual(results_1[0].stat_entry.st_size, 1337)

    results_2 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_2,
    )
    self.assertLen(results_2, 1)
    self.assertEqual(results_2[0].components, ["foo", "bar"])
    self.assertEqual(results_2[0].stat_entry.st_size, 42)

  def testListDescendantPathInfosTimestampStatValue_ListVersion(self):
    client_id = db_test_utils.InitializeClient(self.db)

    timestamp_0 = self.db.Now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )

    path_info.stat_entry.st_size = 1337
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_1 = self.db.Now()

    path_info.stat_entry.st_size = 42
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_2 = self.db.Now()

    results_0 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
        timestamp=timestamp_0,
    )
    self.assertEmpty(results_0)

    results_1 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
        timestamp=timestamp_1,
    )
    self.assertLen(results_1, 1)
    self.assertEqual(results_1[0].components, ["foo", "bar"])
    self.assertEqual(results_1[0].stat_entry.st_size, 1337)

    results_2 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=["foo"],
        timestamp=timestamp_2,
    )
    self.assertLen(results_2, 1)
    self.assertEqual(results_2[0].components, ["foo", "bar"])
    self.assertEqual(results_2[0].stat_entry.st_size, 42)

  def testListDescendantPathInfosTimestampHashValue(self):
    client_id = db_test_utils.InitializeClient(self.db)

    timestamp_0 = self.db.Now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )

    path_info.hash_entry.md5 = b"quux"
    path_info.hash_entry.sha256 = b"thud"
    self.db.WritePathInfos(client_id, [path_info])

    timestamp_1 = self.db.Now()

    path_info.hash_entry.md5 = b"norf"
    path_info.hash_entry.sha256 = b"blargh"
    self.db.WritePathInfos(client_id, [path_info])

    timestamp_2 = self.db.Now()

    results_0 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_0,
    )
    self.assertEmpty(results_0)

    results_1 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_1,
    )
    self.assertLen(results_1, 1)
    self.assertEqual(results_1[0].hash_entry.md5, b"quux")
    self.assertEqual(results_1[0].hash_entry.sha256, b"thud")

    results_2 = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=(),
        timestamp=timestamp_2,
    )
    self.assertLen(results_2, 1)
    self.assertEqual(results_2[0].components, ["foo"])
    self.assertEqual(results_2[0].hash_entry.md5, b"norf")
    self.assertEqual(results_2[0].hash_entry.sha256, b"blargh")

  def testListDescendantPathInfosWildcards(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("foo", "quux"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("bar", "norf"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("___", "thud"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%%%", "ztesch"),
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("___",),
    )
    self.assertLen(results, 1)
    self.assertEqual(results[0].components, ["___", "thud"])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("%%%",),
    )
    self.assertLen(results, 1)
    self.assertEqual(results[0].components, ["%%%", "ztesch"])

  def testListDescendantPathInfosManyWildcards(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%", "%%", "%%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%", "%%%", "%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%%", "%", "%%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%%", "%%%", "%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%%%", "%%", "%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("__", "%%", "__"),
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("%",),
    )

    self.assertLen(results, 4)
    self.assertEqual(results[0].components, ["%", "%%"])
    self.assertEqual(results[1].components, ["%", "%%", "%%%"])
    self.assertEqual(results[2].components, ["%", "%%%"])
    self.assertEqual(results[3].components, ["%", "%%%", "%"])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("%%",),
    )

    self.assertLen(results, 4)
    self.assertEqual(results[0].components, ["%%", "%"])
    self.assertEqual(results[1].components, ["%%", "%", "%%%"])
    self.assertEqual(results[2].components, ["%%", "%%%"])
    self.assertEqual(results[3].components, ["%%", "%%%", "%"])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("__",),
    )

    self.assertLen(results, 2)
    self.assertEqual(results[0].components, ["__", "%%"])
    self.assertEqual(results[1].components, ["__", "%%", "__"])

  def testListDescendantPathInfosWildcardsWithMaxDepth(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%", "%%foo", "%%%bar", "%%%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%", "%%foo", "%%%baz", "%%%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%", "%%quux", "%%%norf", "%%%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%", "%%quux", "%%%thud", "%%%%"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%%", "%%bar", "%%%quux"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("%%", "%%baz", "%%%norf"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("__", "__bar__", "__quux__"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("__", "__baz__", "__norf__"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("blargh",),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("ztesch",),
            ),
        ],
    )

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("%",),
        max_depth=2,
    )

    self.assertLen(results, 6)
    self.assertEqual(results[0].components, ["%", "%%foo"])
    self.assertEqual(results[1].components, ["%", "%%foo", "%%%bar"])
    self.assertEqual(results[2].components, ["%", "%%foo", "%%%baz"])
    self.assertEqual(results[3].components, ["%", "%%quux"])
    self.assertEqual(results[4].components, ["%", "%%quux", "%%%norf"])
    self.assertEqual(results[5].components, ["%", "%%quux", "%%%thud"])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("%%",),
        max_depth=1,
    )
    self.assertLen(results, 2)
    self.assertEqual(results[0].components, ["%%", "%%bar"])
    self.assertEqual(results[1].components, ["%%", "%%baz"])

    results = self.db.ListDescendantPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("__",),
        max_depth=1,
    )
    self.assertLen(results, 2)
    self.assertEqual(results[0].components, ["__", "__bar__"])
    self.assertEqual(results[1].components, ["__", "__baz__"])

  def testListChildPathInfosRoot(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "baz"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["quux", "norf"],
            ),
        ],
    )

    results = self.db.ListChildPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=()
    )

    self.assertEqual(results[0].components, ["foo"])
    self.assertTrue(results[0].directory)
    self.assertEqual(results[1].components, ["quux"])
    self.assertTrue(results[1].directory)

  def testListChildPathInfosRootDeeper(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("foo", "bar", "baz"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("foo", "bar", "quux"),
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=("foo", "bar", "norf", "thud"),
            ),
        ],
    )

    results = self.db.ListChildPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=()
    )

    self.assertLen(results, 1)
    self.assertEqual(
        results[0].components,
        [
            "foo",
        ],
    )
    self.assertTrue(results[0].directory)

  def testListChildPathInfosDetails(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo", "bar"]
    )
    path_info.stat_entry.st_size = 42
    self.db.WritePathInfos(client_id, [path_info])

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo", "baz"]
    )
    path_info.hash_entry.md5 = b"quux"
    path_info.hash_entry.sha256 = b"norf"
    self.db.WritePathInfos(client_id, [path_info])

    results = self.db.ListChildPathInfos(
        client_id, objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    self.assertEqual(results[0].components, ["foo", "bar"])
    self.assertEqual(results[0].stat_entry.st_size, 42)
    self.assertEqual(results[1].components, ["foo", "baz"])
    self.assertEqual(results[1].hash_entry.md5, b"quux")
    self.assertEqual(results[1].hash_entry.sha256, b"norf")

  def testListChildPathInfosDeepSorted(self):
    client_id = db_test_utils.InitializeClient(self.db)

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz", "quux"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz", "norf"],
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=["foo", "bar", "baz", "thud"],
            ),
        ],
    )

    results = self.db.ListChildPathInfos(
        client_id,
        objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )
    self.assertEqual(results[0].components, ["foo", "bar", "baz", "norf"])
    self.assertEqual(results[1].components, ["foo", "bar", "baz", "quux"])
    self.assertEqual(results[2].components, ["foo", "bar", "baz", "thud"])

  def testListChildPathInfosTimestamp(self):
    client_id = db_test_utils.InitializeClient(self.db)

    timestamp_0 = self.db.Now()

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_1.stat_entry.st_size = 1
    self.db.WritePathInfos(client_id, [path_info_1])
    timestamp_1 = self.db.Now()

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "baz")
    )
    path_info_2.stat_entry.st_size = 2
    self.db.WritePathInfos(client_id, [path_info_2])
    timestamp_2 = self.db.Now()

    results_0 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_0,
    )
    self.assertEmpty(results_0)

    results_1 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_1,
    )
    self.assertLen(results_1, 1)
    self.assertEqual(results_1[0].components, ["foo", "bar"])
    self.assertEqual(results_1[0].stat_entry.st_size, 1)

    results_2 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        timestamp=timestamp_2,
    )
    self.assertLen(results_2, 2)
    self.assertEqual(results_2[0].components, ["foo", "bar"])
    self.assertEqual(results_2[0].stat_entry.st_size, 1)
    self.assertEqual(results_2[1].components, ["foo", "baz"])
    self.assertEqual(results_2[1].stat_entry.st_size, 2)

  def testListChildPathInfosTimestampStatAndHashValue(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )
    path_info.stat_entry.st_size = 42
    path_info.hash_entry.sha256 = b"quux"
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_1 = self.db.Now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )
    path_info.stat_entry.st_size = 108
    path_info.hash_entry.sha256 = b"norf"
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_2 = self.db.Now()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar", "baz"),
    )
    path_info.stat_entry.st_size = 1337
    path_info.hash_entry.sha256 = b"thud"
    self.db.WritePathInfos(client_id, [path_info])
    timestamp_3 = self.db.Now()

    results_1 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar"),
        timestamp=timestamp_1,
    )
    self.assertLen(results_1, 1)
    self.assertEqual(results_1[0].components, ["foo", "bar", "baz"])
    self.assertEqual(results_1[0].stat_entry.st_size, 42)
    self.assertEqual(results_1[0].hash_entry.sha256, b"quux")

    results_2 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar"),
        timestamp=timestamp_2,
    )
    self.assertLen(results_2, 1)
    self.assertEqual(results_2[0].components, ["foo", "bar", "baz"])
    self.assertEqual(results_2[0].stat_entry.st_size, 108)
    self.assertEqual(results_2[0].hash_entry.sha256, b"norf")

    results_3 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo", "bar"),
        timestamp=timestamp_3,
    )
    self.assertLen(results_3, 1)
    self.assertEqual(results_3[0].components, ["foo", "bar", "baz"])
    self.assertEqual(results_3[0].stat_entry.st_size, 1337)
    self.assertEqual(results_3[0].hash_entry.sha256, b"thud")

  def testListChildPathInfosBackslashes(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\", "\\\\", "\\\\\\"),
    )
    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\", "\\\\\\", "\\\\"),
    )
    path_info_3 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\", "foo\\bar", "baz"),
    )
    self.db.WritePathInfos(client_id, [path_info_1, path_info_2, path_info_3])

    results_0 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\",),
    )
    self.assertLen(results_0, 3)
    self.assertEqual(results_0[0].components, ["\\", "\\\\"])
    self.assertEqual(results_0[1].components, ["\\", "\\\\\\"])
    self.assertEqual(results_0[2].components, ["\\", "foo\\bar"])

    results_1 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\", "\\\\"),
    )
    self.assertLen(results_1, 1)
    self.assertEqual(results_1[0].components, ["\\", "\\\\", "\\\\\\"])

    results_2 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\", "\\\\\\"),
    )
    self.assertLen(results_2, 1)
    self.assertEqual(results_2[0].components, ["\\", "\\\\\\", "\\\\"])

    results_3 = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("\\", "foo\\bar"),
    )
    self.assertLen(results_3, 1)
    self.assertEqual(results_3[0].components, ["\\", "foo\\bar", "baz"])

  def testListChildPathInfosTSKRootVolume(self):
    client_id = db_test_utils.InitializeClient(self.db)
    volume = "\\\\?\\Volume{2d4fbbd3-0000-0000-0000-100000000000}"

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.TSK,
        components=(volume, "foobar.txt"),
    )
    path_info.stat_entry.st_size = 42
    self.db.WritePathInfos(client_id, [path_info])

    results = self.db.ListChildPathInfos(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.TSK,
        components=(volume,),
    )

    self.assertLen(results, 1)
    self.assertEqual(results[0].components, [volume, "foobar.txt"])
    self.assertEqual(results[0].stat_entry.st_size, 42)

  def testReadPathInfosHistoriesEmpty(self):
    client_id = db_test_utils.InitializeClient(self.db)
    result = self.db.ReadPathInfosHistories(
        client_id, objects_pb2.PathInfo.PathType.OS, []
    )
    self.assertEqual(result, {})

  def testReadPathInfosHistoriesDoesNotRaiseOnUnknownClient(self):
    results = self.db.ReadPathInfosHistories(
        "C.FFFF111122223333", objects_pb2.PathInfo.PathType.OS, [("foo",)]
    )

    self.assertEqual(results[("foo",)], [])

  def testReadPathInfosHistoriesWithSingleFileWithSingleHistoryItem(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )

    path_info.stat_entry.st_size = 42
    path_info.hash_entry.sha256 = b"quux"

    then = self.db.Now()
    self.db.WritePathInfos(client_id, [path_info])
    now = self.db.Now()

    path_infos = self.db.ReadPathInfosHistories(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo",)]
    )
    self.assertLen(path_infos, 1)

    pi = path_infos[("foo",)]
    self.assertLen(pi, 1)
    self.assertEqual(pi[0].stat_entry.st_size, 42)
    self.assertEqual(pi[0].hash_entry.sha256, b"quux")
    self.assertBetween(pi[0].timestamp, then, now)

  def testReadPathInfosHistoriesWithTwoFilesWithSingleHistoryItemEach(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )
    path_info_1.stat_entry.st_mode = 1337

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["bar"]
    )
    path_info_2.hash_entry.sha256 = b"quux"

    then = self.db.Now()
    self.db.WritePathInfos(client_id, [path_info_1, path_info_2])
    now = self.db.Now()

    path_infos = self.db.ReadPathInfosHistories(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo",), ("bar",)]
    )
    self.assertLen(path_infos, 2)

    pi = path_infos[("bar",)]
    self.assertLen(pi, 1)

    self.assertEqual(pi[0].components, ["bar"])
    self.assertEqual(pi[0].hash_entry.sha256, b"quux")
    self.assertBetween(pi[0].timestamp, then, now)

    pi = path_infos[("foo",)]
    self.assertLen(pi, 1)

    self.assertEqual(
        pi[0].components,
        ["foo"],
    )
    self.assertEqual(pi[0].stat_entry.st_mode, 1337)
    self.assertBetween(pi[0].timestamp, then, now)

  def testReadPathInfosHistoriesWithTwoFilesWithTwoHistoryItems(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )
    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["bar"]
    )

    timestamp_1 = self.db.Now()

    path_info_1.stat_entry.st_mode = 1337
    self.db.WritePathInfos(client_id, [path_info_1])

    timestamp_2 = self.db.Now()

    path_info_1.stat_entry.st_mode = 1338
    self.db.WritePathInfos(client_id, [path_info_1])

    timestamp_3 = self.db.Now()

    path_info_2.stat_entry.st_mode = 109
    self.db.WritePathInfos(client_id, [path_info_2])

    timestamp_4 = self.db.Now()

    path_info_2.stat_entry.st_mode = 110
    self.db.WritePathInfos(client_id, [path_info_2])

    timestamp_5 = self.db.Now()

    path_infos = self.db.ReadPathInfosHistories(
        client_id, objects_pb2.PathInfo.PathType.OS, [("foo",), ("bar",)]
    )
    self.assertLen(path_infos, 2)

    pi = path_infos[("bar",)]
    self.assertLen(pi, 2)

    self.assertEqual(pi[0].components, ["bar"])
    self.assertEqual(pi[0].stat_entry.st_mode, 109)
    self.assertBetween(pi[0].timestamp, timestamp_3, timestamp_4)

    self.assertEqual(pi[1].components, ["bar"])
    self.assertEqual(pi[1].stat_entry.st_mode, 110)
    self.assertBetween(pi[1].timestamp, timestamp_4, timestamp_5)

    pi = path_infos[("foo",)]
    self.assertLen(pi, 2)

    self.assertEqual(pi[0].components, ["foo"])
    self.assertEqual(pi[0].stat_entry.st_mode, 1337)
    self.assertBetween(pi[0].timestamp, timestamp_1, timestamp_2)

    self.assertEqual(pi[1].components, ["foo"])
    self.assertEqual(pi[1].stat_entry.st_mode, 1338)
    self.assertBetween(pi[1].timestamp, timestamp_2, timestamp_3)

  def testReadPathInfoHistoryTimestamp(self):
    client_id = db_test_utils.InitializeClient(self.db)

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=["foo"]
    )

    path_info.stat_entry.st_size = 0
    self.db.WritePathInfos(client_id, [path_info])

    path_info.stat_entry.st_size = 1
    self.db.WritePathInfos(client_id, [path_info])

    path_info.stat_entry.st_size = 2
    self.db.WritePathInfos(client_id, [path_info])

    cutoff = self.db.Now()

    path_info.stat_entry.st_size = 1337
    self.db.WritePathInfos(client_id, [path_info])

    path_infos = self.db.ReadPathInfoHistory(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=("foo",),
        cutoff=cutoff,
    )

    self.assertLen(path_infos, 3)
    self.assertEqual(path_infos[0].stat_entry.st_size, 0)
    self.assertEqual(path_infos[1].stat_entry.st_size, 1)
    self.assertEqual(path_infos[2].stat_entry.st_size, 2)

  def _WriteBlobReferences(self):
    blob_ref_1 = objects_pb2.BlobReference(
        offset=0, size=42, blob_id=(b"01234567" * 4)
    )
    blob_ref_2 = objects_pb2.BlobReference(
        offset=42, size=42, blob_id=(b"01234568" * 4)
    )
    hash_id_1 = rdf_objects.SHA256HashID(b"0a1b2c3d" * 4)
    hash_id_2 = rdf_objects.SHA256HashID(b"0a1b2c3e" * 4)
    data = {
        hash_id_1: [blob_ref_1],
        hash_id_2: [blob_ref_1, blob_ref_2],
    }
    self.db.WriteHashBlobReferences(data)

    return hash_id_1, hash_id_2

  def testReadLatestPathInfosReturnsNothingForNonExistingPaths(self):
    client_a_id = db_test_utils.InitializeClient(self.db)
    client_b_id = db_test_utils.InitializeClient(self.db)

    path_1 = db.ClientPath.OS(client_a_id, components=("foo", "baz"))
    path_2 = db.ClientPath.TSK(client_b_id, components=("foo", "baz"))

    results = self.db.ReadLatestPathInfosWithHashBlobReferences(
        [path_1, path_2]
    )
    self.assertEqual(results, {path_1: None, path_2: None})

  def testReadLatestPathInfosReturnsNothingWhenNoFilesCollected(self):
    client_a_id = db_test_utils.InitializeClient(self.db)
    client_b_id = db_test_utils.InitializeClient(self.db)

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    self.db.WritePathInfos(client_a_id, [path_info_1])
    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.TSK, components=("foo", "baz")
    )
    self.db.WritePathInfos(client_b_id, [path_info_2])

    path_1 = db.ClientPath.OS(client_a_id, components=("foo", "bar"))
    path_2 = db.ClientPath.TSK(client_b_id, components=("foo", "baz"))

    results = self.db.ReadLatestPathInfosWithHashBlobReferences(
        [path_1, path_2]
    )
    self.assertEqual(results, {path_1: None, path_2: None})

  def testReadLatestPathInfosFindsTwoCollectedFilesWhenTheyAreTheOnlyEntries(
      self,
  ):
    client_a_id = db_test_utils.InitializeClient(self.db)
    client_b_id = db_test_utils.InitializeClient(self.db)
    hash_id_1, hash_id_2 = self._WriteBlobReferences()

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_1.hash_entry.sha256 = hash_id_1.AsBytes()
    self.db.WritePathInfos(client_a_id, [path_info_1])
    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.TSK, components=("foo", "baz")
    )
    path_info_2.hash_entry.sha256 = hash_id_2.AsBytes()
    self.db.WritePathInfos(client_b_id, [path_info_2])

    path_1 = db.ClientPath.OS(client_a_id, components=("foo", "bar"))
    path_2 = db.ClientPath.TSK(client_b_id, components=("foo", "baz"))

    results = self.db.ReadLatestPathInfosWithHashBlobReferences(
        [path_1, path_2]
    )
    self.assertCountEqual(results.keys(), [path_1, path_2])
    self.assertEqual(results[path_1].hash_entry, path_info_1.hash_entry)
    self.assertEqual(results[path_2].hash_entry, path_info_2.hash_entry)
    self.assertTrue(results[path_1].timestamp)
    self.assertTrue(results[path_2].timestamp)

  def testReadLatestPathInfosCorrectlyFindsCollectedFileWithNonLatestEntry(
      self,
  ):
    client_id = db_test_utils.InitializeClient(self.db)
    hash_id, _ = self._WriteBlobReferences()

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_1.hash_entry.sha256 = hash_id.AsBytes()
    self.db.WritePathInfos(client_id, [path_info_1])

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    self.db.WritePathInfos(client_id, [path_info_2])

    path = db.ClientPath.OS(client_id, components=("foo", "bar"))
    results = self.db.ReadLatestPathInfosWithHashBlobReferences([path])

    self.assertCountEqual(results.keys(), [path])
    self.assertEqual(results[path].hash_entry, path_info_1.hash_entry)
    self.assertTrue(results[path].timestamp)

  def testReadLatestPathInfosCorrectlyFindsLatestOfTwoCollectedFiles(self):
    client_id = db_test_utils.InitializeClient(self.db)
    hash_id_1, hash_id_2 = self._WriteBlobReferences()

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_1.hash_entry.sha256 = hash_id_1.AsBytes()
    self.db.WritePathInfos(client_id, [path_info_1])

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_2.hash_entry.sha256 = hash_id_2.AsBytes()
    self.db.WritePathInfos(client_id, [path_info_2])

    path = db.ClientPath.OS(client_id, components=("foo", "bar"))
    results = self.db.ReadLatestPathInfosWithHashBlobReferences([path])
    self.assertCountEqual(results.keys(), [path])
    self.assertEqual(results[path].hash_entry, path_info_2.hash_entry)
    self.assertTrue(results[path].timestamp)

  def testReadLatestPathInfosCorrectlyFindsLatestCollectedFileBeforeTimestamp(
      self,
  ):
    client_id = db_test_utils.InitializeClient(self.db)
    hash_id_1, hash_id_2 = self._WriteBlobReferences()

    path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_1.hash_entry.sha256 = hash_id_1.AsBytes()
    self.db.WritePathInfos(client_id, [path_info_1])

    time_checkpoint = self.db.Now()

    path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info_2.hash_entry.sha256 = hash_id_2.AsBytes()
    self.db.WritePathInfos(client_id, [path_info_2])

    path = db.ClientPath.OS(client_id, components=("foo", "bar"))
    results = self.db.ReadLatestPathInfosWithHashBlobReferences(
        [path], max_timestamp=time_checkpoint
    )
    self.assertCountEqual(results.keys(), [path])
    self.assertEqual(results[path].hash_entry, path_info_1.hash_entry)
    self.assertTrue(results[path].timestamp)

  def testReadLatestPathInfosMaxTimestampMultiplePaths(self):
    client_id = db_test_utils.InitializeClient(self.db)

    before_hash_id = os.urandom(32)
    after_hash_id = os.urandom(32)

    blob_ref = objects_pb2.BlobReference()
    blob_ref.blob_id = os.urandom(32)

    self.db.WriteHashBlobReferences({
        rdf_objects.SHA256HashID(before_hash_id): [blob_ref],
        rdf_objects.SHA256HashID(after_hash_id): [blob_ref],
    })

    before_path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    before_path_info_1.hash_entry.sha256 = before_hash_id

    before_path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("bar",)
    )
    before_path_info_2.hash_entry.sha256 = before_hash_id

    self.db.WritePathInfos(client_id, [before_path_info_1, before_path_info_2])

    timestamp = self.db.Now()

    after_path_info_1 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo",)
    )
    after_path_info_1.hash_entry.sha256 = after_hash_id

    after_path_info_2 = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("bar",)
    )
    after_path_info_2.hash_entry.sha256 = after_hash_id

    self.db.WritePathInfos(client_id, [after_path_info_1, after_path_info_2])

    client_path_1 = db.ClientPath.OS(client_id, ("foo",))
    client_path_2 = db.ClientPath.OS(client_id, ("bar",))

    results = self.db.ReadLatestPathInfosWithHashBlobReferences(
        [client_path_1, client_path_2], max_timestamp=timestamp
    )

    self.assertLen(results, 2)
    self.assertEqual(results[client_path_1].hash_entry.sha256, before_hash_id)
    self.assertEqual(results[client_path_2].hash_entry.sha256, before_hash_id)

  def testReadLatestPathInfosIncludesStatEntryIfThereIsOneWithSameTimestamp(
      self,
  ):
    client_id = db_test_utils.InitializeClient(self.db)
    hash_id, _ = self._WriteBlobReferences()

    path_info = objects_pb2.PathInfo(
        path_type=objects_pb2.PathInfo.PathType.OS, components=("foo", "bar")
    )
    path_info.stat_entry.st_mode = 42
    path_info.hash_entry.sha256 = hash_id.AsBytes()
    self.db.WritePathInfos(client_id, [path_info])

    path = db.ClientPath.OS(client_id, components=("foo", "bar"))
    results = self.db.ReadLatestPathInfosWithHashBlobReferences([path])

    self.assertCountEqual(results.keys(), [path])
    self.assertEqual(results[path].stat_entry, path_info.stat_entry)
    self.assertEqual(results[path].hash_entry, path_info.hash_entry)
    self.assertTrue(results[path].timestamp)

  def testWriteLongPathInfosWithCommonPrefix(self):
    client_id = db_test_utils.InitializeClient(self.db)

    prefix = ["foobarbaz"] * 303
    quux_components = prefix + ["quux"]
    norf_components = prefix + ["norf"]

    self.db.WritePathInfos(
        client_id,
        [
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=quux_components,
            ),
            objects_pb2.PathInfo(
                path_type=objects_pb2.PathInfo.PathType.OS,
                components=norf_components,
            ),
        ],
    )

    quux_path_info = self.db.ReadPathInfo(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=quux_components,
    )
    self.assertEqual(quux_path_info.components, quux_components)

    norf_path_info = self.db.ReadPathInfo(
        client_id=client_id,
        path_type=objects_pb2.PathInfo.PathType.OS,
        components=norf_components,
    )
    self.assertEqual(norf_path_info.components, norf_components)


# This file is a test library and thus does not require a __main__ block.