def ensure_cached(task_cls, expected_num_artifacts=None):
 def decorator(test_fn):
  def wrapper(self, *args, **kwargs):
   with temporary_dir() as artifact_cache:
    self.set_options_for_scope(u'cache.{}'.format(self.options_scope), write_to=[artifact_cache])
    task_cache = os.path.join(artifact_cache, task_cls.stable_name())
    os.mkdir(task_cache)
    test_fn(self, *args, **kwargs)
    num_artifacts = 0
    for (_, _, files) in os.walk(task_cache):
     num_artifacts += len(files)
    if (expected_num_artifacts is None):
     self.assertNotEqual(num_artifacts, 0)
    else:
     self.assertEqual(num_artifacts, expected_num_artifacts)
  return wrapper
 return decorator

def ensure_cached ( task_cls , expected_num_artifacts = None ) : def decorator ( test_fn ) : def wrapper ( self , * args , ** kwargs ) : with temporary_dir ( ) as artifact_cache : self . set_options_for_scope ( u'cache.{}' . format ( self . options_scope ) , write_to = [ artifact_cache ] ) task_cache = os . path . join ( artifact_cache , task_cls . stable_name ( ) ) os . mkdir ( task_cache ) test_fn ( self , * args , ** kwargs ) num_artifacts = 0 for ( _ , _ , files ) in os . walk ( task_cache ) : num_artifacts += len ( files ) if ( expected_num_artifacts is None ) : self . assertNotEqual ( num_artifacts , 0 ) else : self . assertEqual ( num_artifacts , expected_num_artifacts ) return wrapper return decorator
decorator for a task-executing unit test .