Moving ScheduleNewAnalysis to break the cycle

appengine/findit/crash/crash_pipeline.py

Index: appengine/findit/crash/crash_pipeline.py
diff --git a/appengine/findit/crash/crash_pipeline.py b/appengine/findit/crash/crash_pipeline.py
index 9532858384a378b823d50d401baf93fd03345739..111b623b083fc553fd20e70f4c16d0ad4843659d 100644
--- a/appengine/findit/crash/crash_pipeline.py
+++ b/appengine/findit/crash/crash_pipeline.py
@@ -19,8 +19,8 @@ from lib.gitiles import gitiles_repository
 from model import analysis_status
 
 
-# TODO(http://crbug.com/659346): write complete coverage tests for this.
-def FinditForClientID(client_id): # pragma: no cover
+# TODO(http://crbug.com/659346): this needs complete coverage tests.
+def FinditForClientID(client_id):
   """Construct a Findit object from a client id string specifying the class.
 
   We cannot pass Findit objects to the various methods in
@@ -42,17 +42,16 @@ def FinditForClientID(client_id): # pragma: no cover
   # a dict; but that's buggy for some unknown reason.
   if client_id == CrashClient.FRACAS:
     cls = findit_for_chromecrash.FinditForFracas
-  elif client_id == CrashClient.CRACAS:
+  elif client_id == CrashClient.CRACAS: # pragma: no cover
     cls = findit_for_chromecrash.FinditForCracas
-  elif client_id == CrashClient.CLUSTERFUZZ:
+  elif client_id == CrashClient.CLUSTERFUZZ: # pragma: no cover
     cls = findit_for_clusterfuzz.FinditForClusterfuzz
-  else:
+  else: # pragma: no cover
     raise ValueError('FinditForClientID: '
         'unknown or unsupported client %s' % client_id)
 
-  return cls(
-      gitiles_repository.GitilesRepository(http_client=HttpClientAppengine()),
-      CrashWrapperPipeline)
+  return cls(gitiles_repository.GitilesRepository(
+      http_client=HttpClientAppengine()))
 
 
 # Some notes about the classes below, for people who are not
@@ -64,12 +63,17 @@ def FinditForClientID(client_id): # pragma: no cover
 # the same type. In practice, we end up passing all the arguments to the
 # constructors, because we need to have the fields around for logging
 # (e.g., in ``finalized``); thus, there's nothing that needs to be passed
-# to ``run``. Another thing to bear in mind is that whatever objects
+# to ``run``. However, for some inscrutable reason, whatever arguments
+# are passed to the constructors will also be passed to ``run`` (it's
+# unclear what sort of terrible introspection AppEngine does in order to

[stgao, 2016/11/02 01:28:27]

It is not AppEngine, but the AppEngine Pipeline framework
https://github.com/GoogleCloudPlatform/appengine-pipelines/wiki/Python

That's by design in the AppEngine Pipeline. One reason I could think of is that
users of the framework don't have to create constructor unless other functions
like ``finalized`` also need to access those parameters.

[wrengr, 2016/11/02 23:52:26]

Acknowledged.

+# figure this out, but no matter); thus, even though the ``run`` methods
+# don't need them (and don't want them) they must still accept a bunch
+# of arguments. Another thing to bear in mind is that whatever objects
 # ``CrashWrapperPipeline.run`` yields must be JSON-serializable. The base
 # class handles most of that for us, so the force of this constraint is
 # that all the arguments to the constructors for those classes must be
-# JSON-serializable. Thus, we cannot actually pass a Findit object to
-# the constructor, but rather must pass only the ``client_id`` (or whatever
+# JSON-serializable. Thus, we cannot actually pass a Findit object to the
+# constructor, but rather must pass only the ``client_id`` (or whatever
 # JSON dict) and then reconstruct the Findit object from that. Moreover,
 # the ``run`` method and the ``finalized`` method will be run in different
 # processes, so we will actually end up reconstructing the Findit object
@@ -91,8 +95,8 @@ class CrashBasePipeline(BasePipeline):
 
 
 class CrashAnalysisPipeline(CrashBasePipeline):
-  def finalized(self): # pragma: no cover
-    if self.was_aborted:
+  def finalized(self):
+    if self.was_aborted: # pragma: no cover
       self._PutAbortedError()
 
   # N.B., this method must be factored out for unittest reasons; since
@@ -105,9 +109,7 @@ class CrashAnalysisPipeline(CrashBasePipeline):
     analysis.status = analysis_status.ERROR
     analysis.put()
 
-  # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self):
+  def run(self, *_args, **_kwargs):
     # TODO(wrengr): shouldn't this method somehow call _NeedsNewAnalysis
     # to guard against race conditions?
     analysis = self._findit.GetAnalysis(self._crash_identifiers)
@@ -138,21 +140,21 @@ class CrashAnalysisPipeline(CrashBasePipeline):
     analysis.result = result
     for tag_name, tag_value in tags.iteritems():
       # TODO(http://crbug.com/602702): make it possible to add arbitrary tags.
-      if hasattr(analysis, tag_name):
+      # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
+      if hasattr(analysis, tag_name): # pragma: no cover
         setattr(analysis, tag_name, tag_value)
     analysis.status = analysis_status.COMPLETED
     analysis.put()
 
 
 class PublishResultPipeline(CrashBasePipeline):
-  # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
   def finalized(self):
-    if self.was_aborted:  # pragma: no cover.
+    if self.was_aborted: # pragma: no cover.
       logging.error('Failed to publish %s analysis result for %s',
                     repr(self._crash_identifiers), self.client_id)
 
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self):
+  # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
+  def run(self, *_args, **_kwargs): # pragma: no cover

[Sharu Jiang, 2016/11/01 23:03:24]

This is a little hacky, we should at lease add doc saying we will throw out
parameters because of the yield statement. 

[stgao, 2016/11/02 01:28:27]

I don't understand this.
Sharu, do you mind elaborating a little bit? Which yield? Why the yield impacts
here?

[Sharu Jiang, 2016/11/02 17:01:35]

like the `yield` in #190 and #193
What `yield` does is to construct a pipeline and pass whatever parameters of the
construct function to run(*_args, **_kwargs). This way we can force the *_args,
**_kwargs to be the same as in construct function.

However, if we do:
pipe = PublishResultPipeline(client1, crash_identifiers1)
pipe.run(client2, crash_identifiers2, ...)

The pipe will run using (client1, crash_identifiers1) instead of using the
(client2, crash_identifiers2) passed into `run`. I think this may cause
confusion, so if we are to do something like #189-#193, we should at lease add
doc to notify users to use `yield` instead of creating a pipeline and call its
run separately.

[stgao, 2016/11/02 21:22:09]

Would you mind pointing me to the doc for the usage of calling ``run`` directly?
It's new to me and I'd like to learn more.

[wrengr, 2016/11/02 23:52:26]

I agree it's a terrible hack, but it's forced on us by the structure of
AppEngine pipelines. I'll add docstrings to reiterate the discussion from lines
57--81.

[Sharu Jiang, 2016/11/03 23:03:25]

Sorry for the late response, was busy working on my several pending cls, the doc
is in here:
https://sookocheff.com/post/appengine/pipelines/connecting-pipelines/

Saying: "The rule of thumb here is that anything you instantiate your pipeline
with (and subsequently pass to the run method) is accessible within your
pipeline. These are called immediate values and you can treat them as regular
Python values."

[Sharu Jiang, 2016/11/04 00:19:00]

Calling ``run`` directly is not a correct usage for a pipeline ( thus, we cannot
find an example in a doc), so that's why I was suggesting to add that note in
the doc to avoid that. However, if that's very obvious to all pipeline users,
then it's ok to not to bother adding that note in the doc :)

     analysis = self._findit.GetAnalysis(self._crash_identifiers)
     result = analysis.ToPublishableResult(self._crash_identifiers)
     messages_data = [json.dumps(result, sort_keys=True)]
@@ -166,7 +168,8 @@ class PublishResultPipeline(CrashBasePipeline):
                  repr(self._crash_identifiers))
 
 
-class CrashWrapperPipeline(BasePipeline):
+# TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
+class CrashWrapperPipeline(BasePipeline): # pragma: no cover
   """Fire off pipelines to (1) do the analysis and (2) publish results.
 
   The reason we have analysis and publishing as separate pipelines is
@@ -183,9 +186,7 @@ class CrashWrapperPipeline(BasePipeline):
     self._crash_identifiers = crash_identifiers
     self._client_id = client_id
 
-  # TODO(http://crbug.com/659346): write coverage tests.
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self): # pragma: no cover
+  def run(self, *_args, **_kwargs):
     run_analysis = yield CrashAnalysisPipeline(
         self._client_id, self._crash_identifiers)
     with pipeline.After(run_analysis):