<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import os <EOL> from pants . base . build_environment import get_buildroot <EOL> from pants . base . payload import Payload <EOL> from pants . build_graph . target import Target <EOL> from pants . task . task import Task <EOL> from pants_test . tasks . task_test_base import TaskTestBase <EOL> class DummyLibrary ( Target ) : <EOL> def __init__ ( self , address , source , * args , ** kwargs ) : <EOL> payload = Payload ( ) <EOL> payload . add_fields ( { '<STR_LIT>' : self . create_sources_field ( sources = [ source ] , <EOL> sources_rel_path = address . spec_path ) } ) <EOL> self . source = source <EOL> super ( DummyLibrary , self ) . __init__ ( address = address , payload = payload , * args , ** kwargs ) <EOL> class DummyTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> _implementation_version = <NUM_LIT:0> <EOL> @ property <EOL> def incremental ( self ) : <EOL> return self . _incremental <EOL> @ property <EOL> def cache_target_dirs ( self ) : <EOL> return True <EOL> @ classmethod <EOL> def implementation_version_str ( cls ) : <EOL> return str ( cls . _implementation_version ) <EOL> def execute ( self ) : <EOL> with self . invalidated ( self . context . targets ( ) ) as invalidation : <EOL> assert len ( invalidation . invalid_vts ) == <NUM_LIT:1> <EOL> vt = invalidation . invalid_vts [ <NUM_LIT:0> ] <EOL> if vt . is_incremental : <EOL> assert os . path . isdir ( vt . previous_results_dir ) <EOL> with open ( os . path . join ( get_buildroot ( ) , vt . target . source ) , '<STR_LIT:r>' ) as infile : <EOL> outfile_name = os . path . join ( vt . results_dir , os . path . basename ( vt . target . source ) ) <EOL> with open ( outfile_name , '<STR_LIT:a>' ) as outfile : <EOL> outfile . write ( infile . read ( ) ) <EOL> vt . update ( ) <EOL> return vt <EOL> class TaskTest ( TaskTestBase ) : <EOL> _filename = '<STR_LIT:f>' <EOL> @ classmethod <EOL> def task_type ( cls ) : <EOL> return DummyTask <EOL> def assertContent ( self , vt , content ) : <EOL> with open ( os . path . join ( vt . current_results_dir , self . _filename ) , '<STR_LIT:r>' ) as f : <EOL> self . assertEquals ( f . read ( ) , content ) <EOL> def _fixture ( self , incremental ) : <EOL> target = self . make_target ( '<STR_LIT>' , target_type = DummyLibrary , source = self . _filename ) <EOL> context = self . context ( target_roots = [ target ] ) <EOL> task = self . create_task ( context ) <EOL> task . _incremental = incremental <EOL> return target , task <EOL> def _create_clean_file ( self , target , content ) : <EOL> self . create_file ( self . _filename , content ) <EOL> target . mark_invalidation_hash_dirty ( ) <EOL> def test_incremental ( self ) : <EOL> """<STR_LIT>""" <EOL> one = '<STR_LIT>' <EOL> two = '<STR_LIT>' <EOL> target , task = self . _fixture ( incremental = True ) <EOL> self . _create_clean_file ( target , one ) <EOL> vtA = task . execute ( ) <EOL> self . assertContent ( vtA , one ) <EOL> self . _create_clean_file ( target , two ) <EOL> vtB = task . execute ( ) <EOL> self . assertContent ( vtB , one + two ) <EOL> self . _create_clean_file ( target , one ) <EOL> vtC = task . execute ( ) <EOL> self . assertContent ( vtC , one + one ) <EOL> self . assertContent ( vtA , one + one ) <EOL> self . assertContent ( vtB , one + two ) <EOL> self . assertContent ( vtC , one + one ) <EOL> self . assertNotEqual ( vtA . current_results_dir , vtB . current_results_dir ) <EOL> self . assertEqual ( vtA . current_results_dir , vtC . current_results_dir ) <EOL> self . assertEqual ( vtA . results_dir , vtB . results_dir ) <EOL> self . assertEqual ( vtB . results_dir , vtC . results_dir ) <EOL> def test_non_incremental ( self ) : <EOL> """<STR_LIT>""" <EOL> one = '<STR_LIT>' <EOL> two = '<STR_LIT>' <EOL> target , task = self . _fixture ( incremental = False ) <EOL> self . _create_clean_file ( target , one ) <EOL> vtA = task . execute ( ) <EOL> self . assertContent ( vtA , one ) <EOL> self . _create_clean_file ( target , two ) <EOL> vtB = task . execute ( ) <EOL> self . assertContent ( vtA , one ) <EOL> self . assertContent ( vtB , two ) <EOL> self . assertNotEqual ( vtA . current_results_dir , vtB . current_results_dir ) <EOL> self . assertEqual ( vtA . results_dir , vtB . results_dir ) <EOL> def test_implementation_version ( self ) : <EOL> """<STR_LIT>""" <EOL> one = '<STR_LIT>' <EOL> two = '<STR_LIT>' <EOL> target , task = self . _fixture ( incremental = True ) <EOL> DummyTask . _implementation_version = <NUM_LIT:0> <EOL> self . _create_clean_file ( target , one ) <EOL> vtA = task . execute ( ) <EOL> self . assertContent ( vtA , one ) <EOL> DummyTask . _implementation_version = <NUM_LIT:1> <EOL> self . _create_clean_file ( target , two ) <EOL> vtB = task . execute ( ) <EOL> self . assertFalse ( vtA . is_incremental ) <EOL> self . assertFalse ( vtB . is_incremental ) <EOL> self . assertContent ( vtA , one ) <EOL> self . assertContent ( vtB , two ) <EOL> self . assertNotEqual ( vtA . current_results_dir , vtB . current_results_dir ) <EOL> self . assertNotEqual ( vtA . results_dir , vtB . results_dir ) </s>
