<s> from __future__ import absolute_import <EOL> from freight import vcs <EOL> from freight . config import db , queue <EOL> from freight . models import LogChunk , TaskStatus <EOL> from freight . testutils import TransactionTestCase <EOL> from freight . utils . workspace import Workspace <EOL> class ExecuteTaskTestCase ( TransactionTestCase ) : <EOL> def test_simple ( self ) : <EOL> user = self . create_user ( ) <EOL> repo = self . create_repo ( ) <EOL> app = self . create_app ( repository = repo ) <EOL> self . create_taskconfig ( app = app ) <EOL> task = self . create_task ( app = app , user = user ) <EOL> deploy = self . create_deploy ( app = app , task = task ) <EOL> db . session . commit ( ) <EOL> workspace = Workspace ( <EOL> path = repo . get_path ( ) , <EOL> ) <EOL> vcs_backend = vcs . get ( <EOL> repo . vcs , <EOL> url = repo . url , <EOL> workspace = workspace , <EOL> ) <EOL> if vcs_backend . exists ( ) : <EOL> vcs_backend . update ( ) <EOL> else : <EOL> vcs_backend . clone ( ) <EOL> queue . apply ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : deploy . id } ) <EOL> db . session . expire_all ( ) <EOL> assert task . date_started is not None <EOL> assert task . date_finished is not None <EOL> assert task . status == TaskStatus . finished <EOL> logchunks = list ( LogChunk . query . filter ( <EOL> LogChunk . task_id == task . id , <EOL> ) . order_by ( LogChunk . offset . asc ( ) ) ) <EOL> assert len ( logchunks ) >= <NUM_LIT:1> <EOL> all_text = '<STR_LIT>' . join ( c . text for c in logchunks ) <EOL> assert "<STR_LIT>" in all_text </s>
