Repository: google/turbinia
Pull request #437 (2019-08-31 00:12:30)

Job tracking and new finalize Tasks
Some notes about this change:
- Sorry for the big change, I didn't anticipate it being this large, but a decent chunk of this is just new tests.
- Jobs are instantiated per request instead of keeping the same Job around for the lifecycle of the task manager.
- Tasks are now tracked by adding them to the instantiated Job that generated them instead of one big list.
- Evidence returned from tasks is saved in an EvidenceCollection object in the Job so that we can later pass it to the "finalize" tasks
- The task manager tracks all Job objects until the Request and the finalize tasks have fully completed.
- This adds the new concepts of Job finalization and Request finalization Tasks.  When a Job ends, `TurbiniaJob.create_final_task()` gets called, which can generate a final task to process all of the other output from the Job.  Likewise when all Jobs for the entire processing request have completed, a special RequestFinalizationJob is run and can create final new Tasks.  These can be used to combine multiple Task outputs into one output, and will allow Jobs to be broken up into more parallel Tasks.  These new Tasks that are generated get passed the entire EvidenceCollection for the given Job/Request.
- This adds a specific Request finalization Task that will generate and save the full report out to GCS, along with the associated metadata file.  This can be used to generate a GCS/pubsub trigger so Timesketch can import the report as a story.
- Renamed `finalize_result()` to `process_result()` to disambiguate between processing and finalizing.  Also refactored some of this out into `process_job()`. 
- Adds a new FinalReport Evidence type that is configured to save metadata file
- A new EvidenceCollection type is added, which extends Evidence so we can pass it around as evidence, and also has a `collection` list property so we can save other Evidence objects there.
- self.tasks becomes a @property instead of a manually managed list.
- Adds tests for the Task manager, and brings test coverage from 58% to 73%

aarontp: Ok, this one should be ready to review now, thanks!

Patch:
diff --git a/turbinia/client.py b/turbinia/client.py
index 473bd18..abe01b5 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -37,6 +37,7 @@ from turbinia.workers import Priority
 from turbinia.workers.artifact import FileArtifactExtractionTask
 from turbinia.workers.analysis.wordpress import WordpressAccessLogAnalysisTask
 from turbinia.workers.analysis.jenkins import JenkinsAnalysisTask
+from turbinia.workers.finalize_request import FinalizeRequestTask
 from turbinia.workers.grep import GrepTask
 from turbinia.workers.hadoop import HadoopAnalysisTask
 from turbinia.workers.hindsight import HindsightTask
@@ -54,6 +55,7 @@ from turbinia.workers.worker_stat import StatTask
 TASK_MAP = {
     'fileartifactextractiontask': FileArtifactExtractionTask,
     'wordpressaccessloganalysistask': WordpressAccessLogAnalysisTask,
+    'finalizerequesttask': FinalizeRequestTask,
     'jenkinsanalysistask': JenkinsAnalysisTask,
     'greptask': GrepTask,
     'hadoopanalysistask': HadoopAnalysisTask,
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index 6dae495..62e0a31 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -16,6 +16,7 @@
 
 from __future__ import unicode_literals
 
+import copy
 import json
 import os
 import sys
@@ -63,8 +64,12 @@ def evidence_decode(evidence_dict):
         'No Evidence object of type {0:s} in evidence module'.format(type_))
 
   evidence.__dict__ = evidence_dict
-  if evidence_dict['parent_evidence']:
+  if evidence_dict.get('parent_evidence'):
     evidence.parent_evidence = evidence_decode(evidence_dict['parent_evidence'])
+  if evidence_dict.get('collection'):
+    evidence.collection = [
+        evidence_decode(e) for e in evidence_dict['collection']
+    ]
   return evidence
 
 
@@ -138,7 +143,7 @@ class Evidence(object):
 
   def serialize(self):
     """Return JSON serializable object."""
-    serialized_evidence = self.__dict__
+    serialized_evidence = copy.deepcopy(self.__dict__)
     if self.parent_evidence:
       serialized_evidence['parent_evidence'] = self.parent_evidence.serialize()
     return serialized_evidence
@@ -222,6 +227,33 @@ class Evidence(object):
         raise TurbiniaException(message)
 
 
+class EvidenceCollection(Evidence):
+  """A Collection of Evidence objects.
+
+  Attributes:
+    collection(list): The underlying Evidence objects
+  """
+
+  def __init__(self, collection=None, *args, **kwargs):
+    """Initialization for Evidence Collection object."""
+    super(EvidenceCollection, self).__init__(*args, **kwargs)
+    self.collection = collection if collection else []
+
+  def serialize(self):
+    """Return JSON serializable object."""
+    serialized_evidence = super(EvidenceCollection, self).serialize()
+    serialized_evidence['collection'] = [e.serialize() for e in self.collection]
+    return serialized_evidence
+
+  def add_evidence(self, evidence):
+    """Adds evidence to the collection.
+
+    Args:
+      evidence (Evidence): The evidence to add.
+    """
+    self.collection.append(evidence)
+
+
 class Directory(Evidence):
   """Filesystem directory evidence."""
   pass
@@ -428,6 +460,14 @@ class ReportText(Evidence):
     self.copyable = True
 
 
+class FinalReport(ReportText):
+  """Report format for the final complete Turbinia request report."""
+
+  def __init__(self, *args, **kwargs):
+    super(FinalReport, self).__init__(*args, **kwargs)
+    self.save_metadata = True
+
+
 class TextFile(Evidence):
   """Text data."""
 
@@ -446,9 +486,9 @@ class ExportedFileArtifact(Evidence):
 
   REQUIRED_ATTRIBUTES = ['artifact_name']
 
-  def __init__(self, artifact_name=None):
+  def __init__(self, artifact_name=None, *args, **kwargs):
     """Initializes an exported file artifact."""
-    super(ExportedFileArtifact, self).__init__()
+    super(ExportedFileArtifact, self).__init__(*args, **kwargs)
     self.artifact_name = artifact_name
     self.copyable = True
 
diff --git a/turbinia/jobs/__init__.py b/turbinia/jobs/__init__.py
index 7df9221..e916d72 100644
--- a/turbinia/jobs/__init__.py
+++ b/turbinia/jobs/__init__.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Turbinia jobs."""
+from turbinia.jobs import finalize_request
 from turbinia.jobs import grep
 from turbinia.jobs import hadoop
 from turbinia.jobs import http_access_logs
diff --git a/turbinia/jobs/finalize_request.py b/turbinia/jobs/finalize_request.py
new file mode 100644
index 0000000..6c446e1
--- /dev/null
+++ b/turbinia/jobs/finalize_request.py
@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Job to finalize a request."""
+
+from __future__ import unicode_literals
+
+from turbinia.evidence import FinalReport
+from turbinia.jobs import interface
+from turbinia.jobs import manager
+from turbinia.workers.finalize_request import FinalizeRequestTask
+
+
+class FinalizeRequestJob(interface.TurbiniaJob):
+  """Runs request finalization Tasks."""
+  # An empty evidence input here means that this Job will never get
+  # automatically created.  This is generated by the TaskManager implicitly when
+  # all of the other Jobs in the request have completed.
+  evidence_input = []
+  evidence_output = [FinalReport]
+
+  NAME = 'FinalizeRequestJob'
+
+  def __init__(self):
+    super(FinalizeRequestJob, self).__init__()
+    self.is_finalize_job = True
+
+  def create_tasks(self, evidence):
+    """Create finalize Tasks.
+
+    Args:
+      evidence (list[Evidence]): Evidence to process
+
+    Returns:
+      list[FinalizeRequestTask]: A list of FinalizeRequestTasks.
+    """
+    return [FinalizeRequestTask() for _ in evidence]
+
+
+manager.JobsManager.RegisterJob(FinalizeRequestJob)
diff --git a/turbinia/jobs/interface.py b/turbinia/jobs/interface.py
index e84b23c..a2a29ba 100644
--- a/turbinia/jobs/interface.py
+++ b/turbinia/jobs/interface.py
@@ -15,22 +15,53 @@
 """Interface for Jobs."""
 import uuid
 
+import logging
+
+from turbinia.evidence import EvidenceCollection
+
+log = logging.getLogger('turbinia')
+
 
 class TurbiniaJob(object):
   """Base class for Turbinia Jobs.
 
   Attributes:
-    name: Name of Job
-    id: Id of job
-    priority: Job priority from 0-100, lowest == highest priority
+    name (str): Name of Job
+    id (str): ID of job
+    is_finalize_job (bool): True if this is a Job for finalizing the request.
+    is_finalized (bool): True if the request has been finalized.
+    request_id (str): ID of the request
+    priority (int): Job priority from 0-100, lowest == highest priority
+    tasks (list): A list of Tasks generated by this Job
+    completed_task_count (int): The number of Tasks that have been removed.
+    evidence (EvidenceCollection): The Evidence returned by Tasks from this Job.
   """
 
   NAME = 'name'
 
-  def __init__(self):
+  def __init__(self, request_id=None, evidence_config=None):
     self.name = self.NAME
     self.id = uuid.uuid4().hex
+    self.is_finalize_job = False
+    self.is_finalized = False
     self.priority = 100
+    self.request_id = request_id
+    self.tasks = []
+    self.completed_task_count = 0
+    self.evidence = EvidenceCollection()
+    self.evidence.request_id = request_id
+    self.evidence.config = evidence_config if evidence_config else {}
+
+  def check_done(self):
+    """Check to see if all Tasks for this Job have completed.
+
+    Returns:
+      bool: True if all Tasks have completed, else False.
+    """
+    if self.completed_task_count and not self.tasks:
+      return True
+    else:
+      return False
 
   def create_tasks(self, evidence):
     """Create Turbinia tasks to be run.
@@ -39,6 +70,46 @@ class TurbiniaJob(object):
       evidence: A list of evidence objects
 
     Returns:
-      A List of TurbiniaTask objects.
+      list[TurbiniaTask]: The newly created Tasks.
     """
     raise NotImplementedError
+
+  # pylint: disable=unused-argument
+  def create_final_task(self):
+    """Create Final Turbinia task to finalize the Job.
+
+    This Task will be run after all of the main Tasks for the Job have been
+    completed.  This can be used to "join" or "reduce" all of the output
+    Evidence generated by the other Tasks.  These can also generate new
+    Evidence, and so these Evidence types should be added to the
+    `output_evidence` class variable.
+
+    Returns:
+      TurbiniaTask: The Task to run
+    """
+    return None
+
+  def remove_task(self, task_id):
+    """Removes a Task from the Job.
+
+    Args:
+      task_id (str): The ID of the task to remove.
+
+    Returns:
+      bool: True for success, else False.
+    """
+    remove_task = None
+    for task in self.tasks:
+      if task_id == task.id:
+        remove_task = task
+        break
+
+    if remove_task:
+      self.tasks.remove(remove_task)
+      log.debug('Removed task {0:s} from Job {1:s}'.format(task_id, self.name))
+      self.completed_task_count += 1
+    else:
+      log.debug(
+          'Could not find task {0:s} to remove from Job {1:s}'.format(
+              task_id, self.name))
+    return bool(remove_task)
diff --git a/turbinia/jobs/manager.py b/turbinia/jobs/manager.py
index ba9030d..53fd6b1 100644
--- a/turbinia/jobs/manager.py
+++ b/turbinia/jobs/manager.py
@@ -71,9 +71,9 @@ class JobsManager(object):
     Returns:
      list[TurbiniaJob]: Job objects
     """
-    job_names = [job.name.lower() for job in jobs]
+    job_names = [job.NAME.lower() for job in jobs]
     job_names = cls.FilterJobNames(job_names, jobs_blacklist, jobs_whitelist)
-    return [job for job in jobs if job.name.lower() in job_names]
+    return [job for job in jobs if job.NAME.lower() in job_names]
 
   @classmethod
   def DeregisterJob(cls, job_class):
@@ -140,9 +140,12 @@ class JobsManager(object):
     return cls._job_classes.keys()
 
   @classmethod
-  def GetJobs(cls):
+  def GetJobs(cls, job_names=None):
     """Retrieves the registered jobs.
 
+    Args:
+      job_names (list[str]): names of the jobs to retrieve.
+
     Yields:
       tuple: containing:
 
@@ -150,7 +153,11 @@ class JobsManager(object):
         type: the job class.
     """
     for job_name, job_class in iter(cls._job_classes.items()):
-      yield job_name, job_class
+      if job_names:
+        if job_name in job_names:
+          yield job_name, job_class
+      else:
+        yield job_name, job_class
 
   @classmethod
   def RegisterJob(cls, job_class):
diff --git a/turbinia/state_manager.py b/turbinia/state_manager.py
index 9816fed..5fef4c3 100644
--- a/turbinia/state_manager.py
+++ b/turbinia/state_manager.py
@@ -187,6 +187,7 @@ class DatastoreStateManager(BaseStateManager):
     return data
 
   def update_task(self, task):
+    task.touch()
     try:
       with self.client.transaction():
         entity = self.client.get(task.state_key)
@@ -271,6 +272,7 @@ class RedisStateManager(BaseStateManager):
     return tasks
 
   def update_task(self, task):
+    task.touch()
     key = task.state_key
     if not self.client.get(key):
       self.write_new_task(task)
diff --git a/turbinia/task_manager.py b/turbinia/task_manager.py
index 11f6ec4..7eddb60 100644
--- a/turbinia/task_manager.py
+++ b/turbinia/task_manager.py
@@ -24,6 +24,7 @@ from turbinia import workers
 from turbinia import evidence
 from turbinia import config
 from turbinia import state_manager
+from turbinia import TurbiniaException
 from turbinia.jobs import manager as jobs_manager
 
 config.LoadConfig()
@@ -89,18 +90,29 @@ class BaseTaskManager(object):
   processes results from Tasks that have run.
 
   Attributes:
-    jobs: A list of instantiated job objects
-    evidence: A list of evidence objects to process
-    state_manager: State manager object to handle syncing with storage
-    tasks: A list of outstanding TurbiniaTask objects
+    jobs (list[TurbiniaJob]): Uninstantiated job classes.
+    running_jobs (list[TurbiniaJob]): A list of jobs that are
+        currently running.
+    evidence (list): A list of evidence objects to process.
+    state_manager (DatastoreStateManager|RedisStateManager): State manager
+        object to handle syncing with storage.
+    tasks (list[TurbiniaTask]): Running tasks.
   """
 
   def __init__(self):
     self.jobs = []
-    self.evidence = []
-    self.tasks = []
+    self.running_jobs = []
     self.state_manager = state_manager.get_state_manager()
 
+  @property
+  def tasks(self):
+    """A property that returns all outstanding Tasks.
+
+    Returns:
+      list[TurbiniaTask]: All outstanding Tasks.
+    """
+    return [task for job in self.running_jobs for task in job.tasks]
+
   def _backend_setup(self, *args, **kwargs):
     """Sets up backend dependencies.
 
@@ -117,7 +129,6 @@ class BaseTaskManager(object):
       jobs_whitelist (list): The only Jobs will be included to run
     """
     self._backend_setup(*args, **kwargs)
-    # TODO(aarontp): Consider instantiating a job per evidence object
     job_names = jobs_manager.JobsManager.GetJobNames()
     if jobs_blacklist or jobs_whitelist:
       log.info(
@@ -125,7 +136,7 @@ class BaseTaskManager(object):
               jobs_whitelist, jobs_blacklist))
       job_names = jobs_manager.JobsManager.FilterJobNames(
           job_names, jobs_blacklist, jobs_whitelist)
-    self.jobs = jobs_manager.JobsManager.GetJobInstances(job_names)
+    self.jobs = [job for _, job in jobs_manager.JobsManager.GetJobs(job_names)]
     log.debug('Registered job list: {0:s}'.format(str(job_names)))
 
   def add_evidence(self, evidence_):
@@ -143,7 +154,6 @@ class BaseTaskManager(object):
       raise turbinia.TurbiniaException(
           'Jobs must be registered before evidence can be added')
     log.info('Adding new evidence: {0:s}'.format(str(evidence_)))
-    self.evidence.append(evidence_)
     job_count = 0
     jobs_whitelist = evidence_.config.get('jobs_whitelist', [])
     jobs_blacklist = evidence_.config.get('jobs_blacklist', [])
@@ -165,14 +175,15 @@ class BaseTaskManager(object):
       # comment figured out.
       # pylint: disable=unidiomatic-typecheck
       if [True for t in job.evidence_input if type(evidence_) == t]:
+        job_instance = job(
+            request_id=evidence_.request_id, evidence_config=evidence_.config)
+        self.running_jobs.append(job_instance)
         log.info(
             'Adding {0:s} job to process {1:s}'.format(
-                job.name, evidence_.name))
+                job_instance.name, evidence_.name))
         job_count += 1
-        for task in job.create_tasks([evidence_]):
-          task.base_output_dir = config.OUTPUT_DIR
-          task.requester = evidence_.config.get('requester')
-          self.add_task(task, evidence_)
+        for task in job_instance.create_tasks([evidence_]):
+          self.add_task(task, job_instance, evidence_)
 
     if not job_count:
       log.warning(
@@ -181,42 +192,164 @@ class BaseTaskManager(object):
           'Evidence as input'.format(str(evidence_)))
 
   def check_done(self):
-    """Checks to see if we have any outstanding tasks.
+    """Checks if we have any outstanding tasks.
 
     Returns:
-      Bool indicating whether we are done.
+      bool: Indicating whether we are done.
     """
     return not bool(len(self.tasks))
 
+  def check_request_done(self, request_id):
+    """Checks if we have any outstanding tasks for the request ID.
+
+    Args:
+      request_id (str): The request ID to check for completion
+
+    Returns:
+      bool: Indicating whether all Jobs are done.
+    """
+    job_completion = []
+    for job in self.running_jobs:
+      if request_id == job.request_id:
+        job_completion.append(job.check_done())
+
+    return min(job_completion)
+
+  def check_request_finalized(self, request_id):
+    """Checks if the the request is done and finalized.
+
+    A request can be done but not finalized if all of the Tasks created by the
+    original Jobs have completed, but the "finalize" Job/Tasks have not been
+    run.  These finalize Job/Tasks are created after all of the original
+    Jobs/Tasks have completed. Only one Job needs to be marked as finalized for
+    the entire request to be considered finalized.
+
+    Args:
+      request_id (str): The request ID to check for finalization.
+
+    Returns:
+      bool: Indicating whether all Jobs are done.
+    """
+    request_finalized = False
+    for job in self.running_jobs:
+      if request_id == job.request_id and job.is_finalized:
+        request_finalized = True
+        break
+
+    return request_finalized and self.check_request_done(request_id)
+
   def get_evidence(self):
     """Checks for new evidence to process.
 
     Returns:
-      A list of Evidence objects
+      list[evidence.Evidence]: The evidence to process.
     """
     raise NotImplementedError
 
-  def add_task(self, task, evidence_):
+  def get_job(self, job_id):
+    """Gets the running Job instance from the given Job ID
+
+    Args:
+      job_id (str): The Job id to get the job for.
+
+    Returns:
+      TurbiniaJob|None: Job instance if found, else None
+    """
+    job = None
+    for job_instance in self.running_jobs:
+      if job_id == job_instance.id:
+        job = job_instance
+        break
+
+    return job
+
+  def generate_request_finalize_tasks(self, job):
+    """Generates the Tasks to finalize the given request ID.
+
+    Args:
+      job (TurbiniaJob): The last Job that was run for this request.
+    """
+    request_id = job.request_id
+    final_job = jobs_manager.JobsManager.GetJobInstance('FinalizeRequestJob')
+    final_job.request_id = request_id
+    final_job.evidence.config = job.evidence.config
+    log.debug(
+        'Request {0:s} done, but not finalized, creating FinalizeRequestJob '
+        '{1:s}'.format(request_id, final_job.id))
+
+    # Finalize tasks use EvidenceCollection with all evidence created by the
+    # request or job.
+    final_evidence = evidence.EvidenceCollection()
+    final_evidence.request_id = request_id
+    self.running_jobs.append(final_job)
+
+    # Gather evidence created by every Job in the request.
+    for running_job in self.running_jobs:
+      if running_job.request_id == request_id:
+        final_evidence.collection.extend(running_job.evidence.collection)
+
+    for finalize_task in final_job.create_tasks([final_evidence]):
+      self.add_task(finalize_task, final_job, final_evidence)
+
+  def add_task(self, task, job, evidence_):
     """Adds a task and evidence to process to the task manager.
 
     Args:
       task: An instantiated Turbinia Task
       evidence_: An Evidence object to be processed.
+
+    Raises:
+      TurbiniaException: If the request_id cannot be found.
     """
-    task.request_id = evidence_.request_id
-    self.tasks.append(task)
+    if evidence_.request_id:
+      task.request_id = evidence_.request_id
+    elif job and job.request_id:
+      task.request_id = job.request_id
+    else:
+      raise TurbiniaException(
+          'Request ID not found in Evidence {0!s} or Task {1!s}'.format(
+              evidence_, task))
+
+    evidence_.config = job.evidence.config
+    task.base_output_dir = config.OUTPUT_DIR
+    task.requester = evidence_.config.get('requester')
+    if job:
+      task.job_id = job.id
+      job.tasks.append(task)
     self.state_manager.write_new_task(task)
     self.enqueue_task(task, evidence_)
 
-  def remove_task(self, task):
-    """Removes a task from the queue; Usually after completion or failure.
+  def remove_jobs(self, request_id):
+    """Removes the all Jobs for the given request ID.
 
     Args:
-      task: A TurbiniaTask object
+      request_id (str): The ID of the request we want to remove jobs for.
     """
-    task.touch()
-    self.state_manager.update_task(task)
-    self.tasks.remove(task)
+    remove_jobs = [j for j in self.running_jobs if j.request_id == request_id]
+    log.debug(
+        'Removing {0:d} completed Job(s) for request ID {1:s}.'.format(
+            len(remove_jobs), request_id))
+    # pylint: disable=expression-not-assigned
+    [self.remove_job(j.id) for j in remove_jobs]
+
+  def remove_job(self, job_id):
+    """Removes a Job from the running jobs list.
+
+    Args:
+      job_id (str): The ID of the job to remove.
+
+    Returns:
+      bool: True if Job removed, else False.
+    """
+    remove_job = None
+    for job in self.running_jobs:
+      if job_id == job.id:
+        remove_job = job
+        break
+
+    if remove_job:
+      self.running_jobs.remove(remove_job)
+    return bool(remove_job)
 
   def enqueue_task(self, task, evidence_):
     """Enqueues a task and evidence in the implementation specific task queue.
@@ -227,15 +360,21 @@ class BaseTaskManager(object):
     """
     raise NotImplementedError
 
-  def finalize_result(self, task_result):
+  def process_result(self, task_result):
     """Runs final task results recording.
 
     self.process_tasks handles things that have failed at the task queue layer
     (i.e. PSQ), and this method handles tasks that have potentially failed
     below that layer (i.e. somewhere in our Task code).
 
+    This also adds the Evidence to the running jobs and running requests so we
+    can process those later in 'finalize' Tasks.
+
     Args:
       task_result: The TurbiniaTaskResult object
+
+    Returns:
+      TurbiniaJob|None: The Job for the processed task, else None
     """
     if not task_result.successful:
       log.error(
@@ -248,17 +387,27 @@ class BaseTaskManager(object):
               task_result.status))
 
     if not isinstance(task_result.evidence, list):
-      log.info(
+      log.warning(
           'Task {0:s} from {1:s} did not return evidence list'.format(
               task_result.task_name, task_result.worker_name))
-      return
+      task_result.evidence = []
 
+    job = self.get_job(task_result.job_id)
+    if not job:
+      log.warning(
+          'Received task results for unknown Job from Task ID {0:s}'.format(
+              task_result.task_id))
+
+    # Reprocess new evidence and save instance for later consumption by finalize
+    # tasks.
     for evidence_ in task_result.evidence:
       if isinstance(evidence_, evidence.Evidence):
         log.info(
             'Task {0:s} from {1:s} returned Evidence {2:s}'.format(
                 task_result.task_name, task_result.worker_name, evidence_.name))
         self.add_evidence(evidence_)
+        if job:
+          job.evidence.add_evidence(evidence_)
       else:
         log.error(
             'Task {0:s} from {1:s} returned non-Evidence output type '
@@ -266,15 +415,55 @@ class BaseTaskManager(object):
                 task_result.task_name, task_result.worker_name,
                 type(task_result.evidence)))
 
+    return job
+
+  def process_job(self, job, task):
+    """Processes the Job after Task completes.
+
+    This removes the Task from the running Job and generates the "finalize"
+    Tasks after all the Tasks for the Job and Request have completed.  It also
+    removes all Jobs from the running Job list once everything is complete.
+
+    Args:
+      job (TurbiniaJob): The Job to process
+      task (TurbiniaTask): The Task that just completed.
+    """
+    log.debug(
+        'Processing Job {0:s} for completed Task {1:s}'.format(
+            job.name, task.id))
+    self.state_manager.update_task(task)
+    job.remove_task(task.id)
+    if job.check_done() and not (job.is_finalize_job or task.is_finalize_task):
+      log.debug(
+          'Job {0:s} completed, creating Job finalize tasks'.format(job.name))
+      final_task = job.create_final_task()
+      if final_task:
+        final_task.is_finalize_task = True
+        self.add_task(final_task, job, job.evidence)
+    elif job.check_done() and job.is_finalize_job:
+      job.is_finalized = True
+
+    request_id = job.request_id
+    request_done = self.check_request_done(request_id)
+    request_finalized = self.check_request_finalized(request_id)
+    # If the request is done but not finalized, we generate the finalize tasks.
+    if request_done and not request_finalized:
+      self.generate_request_finalize_tasks(job)
+
+    # If the Job has been finalized then we can remove all the Jobs for this
+    # request since everything is complete.
+    elif request_done and request_finalized:
+      self.remove_jobs(request_id)
+
   def process_tasks(self):
     """Process any tasks that need to be processed.
 
     Returns:
-      A list of tasks that have completed.
+      list[TurbiniaTask]: Tasks to process that have completed.
     """
     raise NotImplementedError
 
-  def run(self):
+  def run(self, under_test=False):
     """Main run loop for TaskManager."""
     log.info('Starting Task Manager run loop')
     while True:
@@ -283,14 +472,18 @@ class BaseTaskManager(object):
 
       for task in self.process_tasks():
         if task.result:
-          self.finalize_result(task.result)
-        self.remove_task(task)
+          job = self.process_result(task.result)
+          if job:
+            self.process_job(job, task)
 
       [self.state_manager.update_task(t) for t in self.tasks]
       if config.SINGLE_RUN and self.check_done():
         log.info('No more tasks to process.  Exiting now.')
         return
 
+      if under_test:
+        break
+
       time.sleep(config.SLEEP_TIME)
 
 
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index c8ef47e..8610b71 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -65,6 +65,7 @@ class TurbiniaTaskResult(object):
       evidence: List of newly created Evidence objects.
       id: Unique Id of result (string of hex)
       input_evidence: The evidence this task processed.
+      job_id (str): The ID of the Job that generated this Task/TaskResult
       report_data (string): Markdown data that can be used in a Turbinia report.
       report_priority (int): Value between 0-100 (0 is the highest priority) to
           be used to order report sections.
@@ -89,13 +90,14 @@ class TurbiniaTaskResult(object):
 
   def __init__(
       self, evidence=None, input_evidence=None, base_output_dir=None,
-      request_id=None):
+      request_id=None, job_id=None):
     """Initialize the TurbiniaTaskResult object."""
 
     self.closed = False
     self.evidence = evidence if evidence else []
     self.input_evidence = input_evidence
     self.id = uuid.uuid4().hex
+    self.job_id = job_id
     self.base_output_dir = base_output_dir
     self.request_id = request_id
 
@@ -299,29 +301,35 @@ class TurbiniaTask(object):
   """Base class for Turbinia tasks.
 
   Attributes:
-      base_output_dir: The base directory that output will go into.  Per-task
-          directories will be created under this.
-      id: Unique Id of task (string of hex)
-      last_update: A datetime object with the last time the task was updated.
-      name: Name of task
-      output_dir: The directory output will go into (including per-task folder).
-      output_manager: An output manager object
-      result: A TurbiniaTaskResult object.
-      request_id: The id of the initial request to process this evidence.
-      run_local: Whether we are running locally without a Worker or not.
-      state_key: A key used to manage task state
-      stub: The task manager implementation specific task stub that exists
-          server side to keep a reference to the remote task objects.  For PSQ
-          this is a task result object, but other implementations have their
-          own stub objects.
-      tmp_dir: Temporary directory for Task to write to.
-      requester: The user who requested the task.
+      base_output_dir (str): The base directory that output will go into.
+          Per-task directories will be created under this.
+      id (str): Unique Id of task (string of hex)
+      is_finalize_task (bool): Whether this is a finalize Task or not.
+      job_id (str): Job ID the Task was created by.
+      last_update (datetime): A datetime object with the last time the task was
+          updated.
+      name (str): Name of task
+      output_dir (str): The directory output will go into (including per-task
+          folder).
+      output_manager (OutputManager): The object that manages saving output.
+      result (TurbiniaTaskResult): A TurbiniaTaskResult object.
+      request_id (str): The id of the initial request to process this evidence.
+      run_local (bool): Whether we are running locally without a Worker or not.
+      state_key (str): A key used to manage task state
+      stub (psq.task.TaskResult|celery.app.Task): The task manager
+          implementation specific task stub that exists server side to keep a
+          reference to the remote task objects.  For PSQ this is a task result
+          object, but other implementations have their own stub objects.
+      tmp_dir (str): Temporary directory for Task to write to.
+      requester (str): The user who requested the task.
       _evidence_config (dict): The config that we want to pass to all new
             evidence created from this task.
   """
 
   # The list of attributes that we will persist into storage
-  STORED_ATTRIBUTES = ['id', 'last_update', 'name', 'request_id', 'requester']
+  STORED_ATTRIBUTES = [
+      'id', 'job_id', 'last_update', 'name', 'request_id', 'requester'
+  ]
 
   def __init__(
       self, name=None, base_output_dir=None, request_id=None, requester=None):
@@ -332,6 +340,8 @@ class TurbiniaTask(object):
       self.base_output_dir = config.OUTPUT_DIR
 
     self.id = uuid.uuid4().hex
+    self.is_finalize_task = False
+    self.job_id = None
     self.last_update = datetime.now()
     self.name = name if name else self.__class__.__name__
     self.output_dir = None
@@ -493,7 +503,7 @@ class TurbiniaTask(object):
     if not self.result:
       self.result = TurbiniaTaskResult(
           input_evidence=evidence, base_output_dir=self.base_output_dir,
-          request_id=self.request_id)
+          request_id=self.request_id, job_id=self.job_id)
       self.result.setup(self)
 
     if not self.run_local:
@@ -552,7 +562,8 @@ class TurbiniaTask(object):
         old_status = 'No previous status'
 
       result = TurbiniaTaskResult(
-          base_output_dir=self.base_output_dir, request_id=self.request_id)
+          base_output_dir=self.base_output_dir, request_id=self.request_id,
+          job_id=self.job_id)
       result.setup(self)
       result.status = '{0:s}. Previous status: [{1:s}]'.format(
           bad_message, old_status)
diff --git a/turbinia/workers/finalize_request.py b/turbinia/workers/finalize_request.py
new file mode 100644
index 0000000..263ae3c
--- /dev/null
+++ b/turbinia/workers/finalize_request.py
@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Task to finalize the request."""
+
+from __future__ import unicode_literals
+
+import os
+
+from turbinia import config
+from turbinia.evidence import FinalReport
+from turbinia.workers import TurbiniaTask
+
+
+class FinalizeRequestTask(TurbiniaTask):
+  """Task to finalize the Turbinia request."""
+
+  def run(self, evidence, result):
+    """Main entry point for Task.
+
+    This generates a final report.
+
+    Args:
+        evidence (EvidenceCollection): All Evidence that has been generated as
+            part of this request.
+        result (TurbiniaTaskResult): The result to place task output into.
+
+    Returns:
+        TurbiniaTaskResult: Task execution results.
+    """
+    # Doing a delayed import to avoid circular dependencies.
+    from turbinia.client import TurbiniaClient
+    report = FinalReport()
+    client = TurbiniaClient()
+
+    report_file = os.path.join(
+        self.tmp_dir, 'final_turbinia_report_{0:s}.md'.format(self.id))
+    report.local_path = report_file
+    report_data = client.format_task_status(
+        config.INSTANCE_ID, config.TURBINIA_PROJECT, config.TURBINIA_REGION,
+        request_id=evidence.request_id, full_report=True)
+
+    result.log('Writing report data to [{0:s}]'.format(report.local_path))
+    with open(report.local_path, 'wb') as file_handle:
+      file_handle.write(report_data.encode('utf-8'))
+
+    result.add_evidence(report, evidence.config)
+    result.close(self, True)
+    return result


Test patch:
diff --git a/turbinia/evidence_test.py b/turbinia/evidence_test.py
index d1c02f4..9759be1 100644
--- a/turbinia/evidence_test.py
+++ b/turbinia/evidence_test.py
@@ -34,10 +34,42 @@ class TestTurbiniaEvidence(unittest.TestCase):
     self.assertTrue(isinstance(rawdisk_json, str))
 
     rawdisk_new = evidence.evidence_decode(json.loads(rawdisk_json))
-    self.assertTrue(isinstance(rawdisk_new, evidence.RawDisk))
+    self.assertIsInstance(rawdisk_new, evidence.RawDisk)
     self.assertEqual(rawdisk_new.name, 'My Evidence')
     self.assertEqual(rawdisk_new.mount_path, '/mnt/foo')
 
+  def testEvidenceCollectionDeserialization(self):
+    """Test that EvidenceCollection deserializes."""
+    rawdisk = evidence.RawDisk(
+        name='My Evidence', local_path='/tmp/foo', mount_path='/mnt/foo')
+    collection = evidence.EvidenceCollection()
+    collection.name = 'testCollection'
+    collection.add_evidence(rawdisk)
+    collection_json = collection.to_json()
+    self.assertTrue(isinstance(collection_json, str))
+
+    collection_new = evidence.evidence_decode(json.loads(collection_json))
+    rawdisk_new = collection_new.collection[0]
+    # Make sure that both the collection, and the things in the collection
+    # deserializd to the correct types.
+    self.assertIsInstance(collection_new, evidence.EvidenceCollection)
+    self.assertIsInstance(rawdisk_new, evidence.RawDisk)
+    self.assertEqual(collection_new.name, 'testCollection')
+    self.assertEqual(rawdisk_new.name, 'My Evidence')
+    self.assertEqual(rawdisk_new.mount_path, '/mnt/foo')
+
+  def testEvidenceCollectionSerialization(self):
+    """Test that EvidenceCollection serializes/unserializes."""
+    evidence_ = evidence.EvidenceCollection()
+    rawdisk = evidence.RawDisk(
+        name='My Evidence', local_path='/tmp/foo', mount_path='/mnt/foo')
+    evidence_.add_evidence(rawdisk)
+    serialized_evidence = evidence_.serialize()
+    collection_evidence = serialized_evidence['collection'][0]
+
+    self.assertIsInstance(serialized_evidence, dict)
+    self.assertEqual(collection_evidence['name'], 'My Evidence')
+
   def testEvidenceSerializationBadType(self):
     """Test that evidence_decode throws error on non-dict type."""
     self.assertRaises(TurbiniaException, evidence.evidence_decode, [1, 2])
diff --git a/turbinia/jobs/interface_test.py b/turbinia/jobs/interface_test.py
new file mode 100644
index 0000000..e442833
--- /dev/null
+++ b/turbinia/jobs/interface_test.py
@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for Turbinia Jobs interface."""
+
+from __future__ import unicode_literals
+
+from turbinia import evidence
+from turbinia.jobs import interface
+from turbinia.workers.workers_test import TestTurbiniaTaskBase
+
+
+class TestJobsInterface(TestTurbiniaTaskBase):
+  """Tests Jobs interface."""
+
+  def setUp(self):
+    super(TestJobsInterface, self).setUp()
+    self.job = interface.TurbiniaJob()
+
+  def testCheckDone(self):
+    """Tests check_done() method."""
+    self.job.completed_task_count = 1
+    self.assertTrue(self.job.check_done())
+
+  def testCheckDoneNoCompletedCount(self):
+    """Tests check_done() method."""
+    self.assertFalse(self.job.check_done())
+
+  def testCheckDoneWithTasks(self):
+    """Tests check_done() method."""
+    self.job.completed_task_count = 1
+    self.job.tasks.append(self.task)
+    self.assertFalse(self.job.check_done())
+
+  def testRemoveTask(self):
+    """Tests remove_task."""
+    task_id = self.task.id
+    self.job.tasks.append(self.task)
+    self.assertTrue(self.job.remove_task(task_id))
+    self.assertListEqual(self.job.tasks, [])
+
+  def testRemoveTaskUnknownTask(self):
+    """Tests remove_task."""
+    task_id = 'noSuchTask'
+    self.job.tasks.append(self.task)
+    self.assertFalse(self.job.remove_task(task_id))
+    self.assertListEqual(self.job.tasks, [self.task])
diff --git a/turbinia/task_manager_test.py b/turbinia/task_manager_test.py
new file mode 100644
index 0000000..456c55f
--- /dev/null
+++ b/turbinia/task_manager_test.py
@@ -0,0 +1,265 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for Turbinia task_manager module."""
+
+from __future__ import unicode_literals
+
+import mock
+
+from turbinia import task_manager
+from turbinia.jobs import manager as jobs_manager
+from turbinia.jobs import plaso
+from turbinia.jobs import strings
+from turbinia.workers.workers_test import TestTurbiniaTaskBase
+
+
+class TestTaskManager(TestTurbiniaTaskBase):
+  """Tests the task_manager module."""
+
+  @mock.patch('turbinia.task_manager.state_manager.get_state_manager')
+  def setUp(self, _):
+    """Sets up the test class."""
+    super(TestTaskManager, self).setUp()
+    self.manager = task_manager.BaseTaskManager()
+    self.job1 = plaso.PlasoJob()
+    self.job2 = strings.StringsJob()
+
+  def testTaskManagerTasksProperty(self):
+    """Basic test for task_manager Tasks property."""
+    self.setResults()
+    job = jobs_manager.JobsManager.GetJobInstance('PlasoJob')
+    job.tasks.extend([self.task, self.task])
+    self.manager.running_jobs.extend([job, job])
+    self.assertEqual(len(self.manager.tasks), 4)
+
+  def testAddEvidence(self):
+    """Tests add_evidence method."""
+    self.setResults()
+    request_id = 'testRequestID'
+    self.evidence.request_id = request_id
+    self.manager.add_task = mock.MagicMock()
+    job = plaso.PlasoJob
+    job.create_tasks = mock.MagicMock(return_value=[self.task])
+    self.manager.jobs = [job]
+    self.manager.add_evidence(self.evidence)
+
+    self.manager.add_task.assert_called()
+    test_job = self.manager.running_jobs[0]
+    test_job.create_tasks.assert_called()
+    self.assertEqual(test_job.request_id, request_id)
+    self.assertEqual(test_job.evidence.request_id, request_id)
+    self.assertIn(test_job, self.manager.running_jobs)
+
+  def testAddEvidenceBlackList(self):
+    """Tests add_evidence method."""
+    self.setResults()
+    self.manager.add_task = mock.MagicMock()
+    self.job1 = plaso.PlasoJob
+    self.job2 = strings.StringsJob
+    self.job1.create_tasks = mock.MagicMock(return_value=[self.task])
+    self.job2.create_tasks = mock.MagicMock(return_value=[self.task])
+    self.manager.jobs = [self.job1, self.job2]
+    self.evidence.config['jobs_blacklist'] = ['StringsJob']
+    self.manager.add_evidence(self.evidence)
+
+    # Only one Plaso job is queued after one is blacklisted
+    self.assertEqual(len(self.manager.running_jobs), 1)
+    test_job = self.manager.running_jobs[0]
+    self.assertEqual(test_job.name, 'PlasoJob')
+
+  def testAddEvidenceWhitelist(self):
+    """Tests add_evidence method."""
+    self.setResults()
+    self.manager.add_task = mock.MagicMock()
+    self.job1 = plaso.PlasoJob
+    self.job2 = strings.StringsJob
+    self.job1.create_tasks = mock.MagicMock(return_value=[self.task])
+    self.job2.create_tasks = mock.MagicMock(return_value=[self.task])
+    self.manager.jobs = [self.job1, self.job2]
+    self.evidence.config['jobs_whitelist'] = ['PlasoJob']
+    self.manager.add_evidence(self.evidence)
+
+    # Only one Plaso job is queued after one is blacklisted
+    self.assertEqual(len(self.manager.running_jobs), 1)
+    test_job = self.manager.running_jobs[0]
+    self.assertEqual(test_job.name, 'PlasoJob')
+
+  def testCheckRequestDoneIsDone(self):
+    """Basic test for check_request_done for when the request is done."""
+    request_id = 'testId'
+    self.job1.request_id = request_id
+    self.job2.request_id = request_id
+    # We want completed tasks for each Job, but no Tasks pending so that it
+    # thinks they are completed.
+    self.job1.completed_task_count = 1
+    self.job2.completed_task_count = 1
+    self.manager.running_jobs = [self.job1, self.job2]
+    self.assertTrue(self.manager.check_request_done(request_id))
+
+  def testCheckRequestDoneNoCompletedTasks(self):
+    """Test for check_request_done no tasks have been completed."""
+    request_id = 'testId'
+    self.job1.request_id = request_id
+    self.job2.request_id = request_id
+    self.manager.running_jobs = [self.job1, self.job2]
+    # With no completed tasks the Jobs will show as not yet done.
+    self.assertFalse(self.manager.check_request_done(request_id))
+
+  def testCheckRequestDonePendingTasks(self):
+    """Test for check_request_done when Tasks are pending."""
+    request_id = 'testId'
+    self.job1.request_id = request_id
+    self.job2.request_id = request_id
+    self.job1.completed_task_count = 1
+    self.job2.completed_task_count = 1
+    self.job1.tasks = [self.task]
+    self.manager.running_jobs = [self.job1, self.job2]
+    # With no completed tasks the Jobs will show as not yet done.
+    self.assertFalse(self.manager.check_request_done(request_id))
+
+  def testGetJob(self):
+    """Tests get_job method."""
+    self.setResults()
+    job_id = 'testID'
+    self.job1.id = job_id
+    self.job2.id = 'NotMyJob'
+    self.manager.running_jobs = [self.job1, self.job2]
+    test_job = self.manager.get_job(job_id)
+    self.assertEqual(test_job.name, 'PlasoJob')
+    self.assertEqual(test_job.id, job_id)
+
+  def testAddTask(self):
+    """Tests add_task method."""
+    request_id = 'testID'
+    self.evidence.request_id = request_id
+    self.manager.enqueue_task = mock.MagicMock()
+    self.manager.add_task(self.task, self.job1, self.evidence)
+    self.assertEqual(self.task.request_id, request_id)
+    self.assertListEqual(self.job1.tasks, [self.task])
+    self.manager.enqueue_task.assert_called()
+
+  def testRemoveJob(self):
+    """Tests remove_job method."""
+    job_id = 'testID'
+    self.job1.id = job_id
+    self.manager.running_jobs.extend([self.job1, self.job2])
+    self.assertTrue(self.manager.remove_job(job_id))
+    self.assertListEqual(self.manager.running_jobs, [self.job2])
+
+  def testFinalizeResult(self):
+    """Tests process_result method."""
+    job_id = 'testJobID'
+    self.job1.id = job_id
+    self.result.job_id = job_id
+    self.result.evidence.append(self.evidence)
+    self.manager.add_evidence = mock.MagicMock()
+    self.manager.running_jobs.append(self.job1)
+    test_job = self.manager.process_result(self.result)
+    self.assertEqual(test_job.id, job_id)
+    self.assertEqual(test_job, self.manager.running_jobs[0])
+    self.assertEqual(test_job.evidence.collection[0], self.evidence)
+    self.manager.add_evidence.assert_called_with(self.evidence)
+
+  def testFinalizeResultBadEvidence(self):
+    """Tests process_result method with bad input evidence."""
+    job_id = 'testJobID'
+    self.job1.id = job_id
+    self.result.job_id = job_id
+    self.result.evidence = None
+    self.assertIsNone(self.manager.process_result(self.result))
+    self.assertIsInstance(self.result.evidence, list)
+
+  def testFinalizeJobGenerateJobFinalizeTasks(self):
+    """Tests process_job method generates Job finalize Task."""
+    request_id = 'testRequestID'
+    self.task.id = 'createdFinalizeTask'
+    self.plaso_task.id = 'originalTask'
+    self.job1.request_id = request_id
+    # We'll use self.task as the finalize task that gets created.
+    self.job1.create_final_task = mock.MagicMock(return_value=self.task)
+    self.job1.evidence.add_evidence(self.evidence)
+    self.manager.enqueue_task = mock.MagicMock()
+    self.job1.tasks.append(self.plaso_task)
+    self.manager.running_jobs.append(self.job1)
+    # Job has one task that is not a finalize task, so it will generate job
+    # finalize tasks.
+    self.manager.process_job(self.job1, self.plaso_task)
+
+    self.job1.create_final_task.assert_called()
+    self.manager.enqueue_task.assert_called()
+    test_task, test_evidence = self.manager.enqueue_task.call_args[0]
+    # Make sure the evidence originally associated with the job is the same as
+    # the new evidence collection.
+    self.assertListEqual(test_evidence.collection, [self.evidence])
+    self.assertEqual(test_task.id, 'createdFinalizeTask')
+    self.assertFalse(self.job1.is_finalized)
+    # We should only have our new finalize task running, and the old task should
+    # be gone.
+    self.assertListEqual(self.manager.running_jobs[0].tasks, [self.task])
+
+  def testFinalizeJobGenerateRequestFinalizeTasks(self):
+    """Tests process_job method generates Request finalize Task."""
+    request_id = 'testRequestID'
+    self.task.is_finalize_task = True
+    self.job1.request_id = request_id
+    self.job1.evidence.add_evidence(self.evidence)
+    self.job1.create_final_task = mock.MagicMock()
+    self.job1.tasks.append(self.task)
+    self.manager.generate_request_finalize_tasks = mock.MagicMock()
+    self.manager.remove_jobs = mock.MagicMock()
+    self.manager.running_jobs.append(self.job1)
+    # Job has one task, and it is a finalze_task.
+    self.manager.process_job(self.job1, self.task)
+
+    # Job won't be finalized because the finalize job task has not completed.
+    self.assertFalse(self.job1.is_finalized)
+    # Because the task is a finalize task it shouldn't generate job finalize.
+    self.job1.create_final_task.assert_not_called()
+    self.manager.generate_request_finalize_tasks.assert_called_with(self.job1)
+    self.manager.remove_jobs.assert_not_called()
+
+  def testFinalizeJobClosingRequest(self):
+    """Tests that process_job method removes jobs when request is finalized."""
+    request_id = 'testRequestID'
+    self.plaso_task.request_id = request_id
+    self.job1.request_id = request_id
+    self.job2.request_id = 'ThisIsADifferentRequest'
+    self.job1.evidence.add_evidence(self.evidence)
+    self.job1.tasks.append(self.plaso_task)
+    self.job1.completed_task_count = 1
+    self.job1.is_finalize_job = True
+    self.manager.generate_request_finalize_tasks = mock.MagicMock()
+    self.manager.running_jobs.extend([self.job1, self.job2])
+    self.manager.process_job(self.job1, self.plaso_task)
+
+    self.manager.generate_request_finalize_tasks.assert_not_called()
+    # The Job for our request was removed, but the second job still remains
+    self.assertListEqual(self.manager.running_jobs, [self.job2])
+    self.assertListEqual(self.job1.tasks, [])
+    self.assertTrue(self.job1.is_finalized)
+
+  def testRun(self):
+    """Test the run() method."""
+    self.manager.get_evidence = mock.MagicMock(return_value=[self.evidence])
+    self.manager.add_evidence = mock.MagicMock()
+    self.task.result = self.result
+    self.manager.process_tasks = mock.MagicMock(return_value=[self.task])
+    self.manager.process_result = mock.MagicMock(return_value=self.job1)
+    self.manager.process_job = mock.MagicMock()
+    self.manager.run(under_test=True)
+
+    self.manager.add_evidence.assert_called_with(self.evidence)
+    self.manager.process_result.assert_called_with(self.result)
+    self.manager.process_job.assert_called_with(self.job1, self.task)