<s> """<STR_LIT>""" <EOL> import datetime <EOL> from django . core . exceptions import ValidationError , ObjectDoesNotExist <EOL> from django . db import connection , transaction , models <EOL> from django . db . models import Q , Count , Max <EOL> from model_utils import Choices <EOL> from . . mtmodel import MTModel , TeamModel , DraftStatusModel <EOL> from . . core . auth import User <EOL> from . . core . models import ProductVersion <EOL> from . . environments . models import Environment , HasEnvironmentsModel <EOL> from . . library . models import CaseVersion , Suite , CaseStep <EOL> class Run ( MTModel , TeamModel , DraftStatusModel , HasEnvironmentsModel ) : <EOL> """<STR_LIT>""" <EOL> productversion = models . ForeignKey ( ProductVersion , related_name = "<STR_LIT>" ) <EOL> name = models . CharField ( max_length = <NUM_LIT:200> ) <EOL> description = models . TextField ( blank = True ) <EOL> start = models . DateField ( default = datetime . date . today ) <EOL> end = models . DateField ( blank = True , null = True ) <EOL> build = models . TextField ( null = True , blank = True ) <EOL> is_series = models . BooleanField ( default = False ) <EOL> series = models . ForeignKey ( "<STR_LIT>" , null = True , blank = True ) <EOL> caseversions = models . ManyToManyField ( <EOL> CaseVersion , through = "<STR_LIT>" , related_name = "<STR_LIT>" ) <EOL> suites = models . ManyToManyField ( <EOL> Suite , through = "<STR_LIT>" , related_name = "<STR_LIT>" ) <EOL> def __unicode__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . name <EOL> def clean ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . end is not None and self . start > self . end : <EOL> raise ValidationError ( "<STR_LIT>" ) <EOL> class Meta : <EOL> permissions = [ ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> @ property <EOL> def parent ( self ) : <EOL> return self . productversion <EOL> @ classmethod <EOL> def cascade_envs_to ( cls , objs , adding ) : <EOL> if adding : <EOL> return { } <EOL> return { RunCaseVersion : RunCaseVersion . objects . filter ( run__in = objs ) } <EOL> def clone ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( <EOL> "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> overrides = kwargs . setdefault ( "<STR_LIT>" , { } ) <EOL> overrides [ "<STR_LIT:status>" ] = self . STATUS . draft <EOL> overrides . setdefault ( "<STR_LIT:name>" , "<STR_LIT>" . format ( self . name ) ) <EOL> return super ( Run , self ) . clone ( * args , ** kwargs ) <EOL> def clone_for_series ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> build = kwargs . pop ( "<STR_LIT>" , None ) <EOL> kwargs . setdefault ( <EOL> "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> overrides = kwargs . setdefault ( "<STR_LIT>" , { } ) <EOL> overrides . setdefault ( "<STR_LIT:name>" , "<STR_LIT>" . format ( <EOL> self . name , build ) ) <EOL> overrides [ "<STR_LIT:status>" ] = self . STATUS . draft <EOL> overrides . setdefault ( "<STR_LIT>" , False ) <EOL> overrides . setdefault ( "<STR_LIT>" , build ) <EOL> overrides . setdefault ( "<STR_LIT>" , self ) <EOL> overrides . setdefault ( <EOL> "<STR_LIT:start>" , <EOL> datetime . date . today ( ) . strftime ( "<STR_LIT>" ) , <EOL> ) <EOL> return super ( Run , self ) . clone ( * args , ** kwargs ) <EOL> def activate ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . status == self . STATUS . draft : <EOL> self . update_case_versions ( ) <EOL> super ( Run , self ) . activate ( * args , ** kwargs ) <EOL> def refresh ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . status == self . STATUS . active : <EOL> self . update_case_versions ( ) <EOL> def update_case_versions ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_series : <EOL> self . _lock_case_versions ( ) <EOL> @ transaction . commit_on_success <EOL> def _lock_case_versions ( self ) : <EOL> """<STR_LIT>""" <EOL> run_env_ids = self . environments . values_list ( "<STR_LIT:id>" , flat = True ) <EOL> if len ( run_env_ids ) : <EOL> cursor = connection . cursor ( ) <EOL> sql = """<STR_LIT>""" . format ( self . id , "<STR_LIT:U+002C>" . join ( map ( str , run_env_ids ) ) ) <EOL> cursor . execute ( sql ) <EOL> cv_list = [ x [ <NUM_LIT:0> ] for x in cursor . fetchall ( ) ] <EOL> else : <EOL> cv_list = [ ] <EOL> self . _delete_runcaseversions ( cv_list ) <EOL> dups = self . runcaseversions . values ( "<STR_LIT>" ) . annotate ( <EOL> num_records = Count ( "<STR_LIT>" ) ) . filter ( num_records__gt = <NUM_LIT:1> ) <EOL> if len ( dups ) > <NUM_LIT:0> : <EOL> for dup in dups : <EOL> rcv_to_save = self . runcaseversions . annotate ( <EOL> latest_result = Max ( "<STR_LIT>" ) ) . filter ( <EOL> caseversion = dup [ "<STR_LIT>" ] ) . order_by ( <EOL> "<STR_LIT>" ) [ <NUM_LIT:0> ] <EOL> self . runcaseversions . filter ( <EOL> caseversion = dup [ "<STR_LIT>" ] ) . exclude ( <EOL> id = rcv_to_save . id ) . delete ( ) <EOL> existing_rcv_map = { } <EOL> for map_item in self . runcaseversions . values ( "<STR_LIT:id>" , "<STR_LIT>" ) : <EOL> existing_rcv_map [ map_item [ "<STR_LIT>" ] ] = map_item [ "<STR_LIT:id>" ] <EOL> rcv_to_update = [ ] <EOL> rcv_proxies_to_create = [ ] <EOL> order = <NUM_LIT:1> <EOL> for cv in cv_list : <EOL> if cv in existing_rcv_map : <EOL> rcv_to_update . append ( { "<STR_LIT>" : cv , "<STR_LIT>" : order } ) <EOL> else : <EOL> kwargs = { <EOL> "<STR_LIT>" : self . id , <EOL> "<STR_LIT>" : cv , <EOL> "<STR_LIT>" : order <EOL> } <EOL> rcv_proxies_to_create . append ( RunCaseVersion ( ** kwargs ) ) <EOL> order += <NUM_LIT:1> <EOL> for rcv in rcv_to_update : <EOL> self . runcaseversions . filter ( <EOL> caseversion = rcv [ "<STR_LIT>" ] ) . update ( order = rcv [ "<STR_LIT>" ] ) <EOL> self . _bulk_insert_new_runcaseversions ( rcv_proxies_to_create ) <EOL> self . _bulk_update_runcaseversion_environments_for_lock ( ) <EOL> self . _lock_caseversions_complete ( ) <EOL> def _delete_runcaseversions ( self , cv_list ) : <EOL> """<STR_LIT>""" <EOL> self . runcaseversions . exclude ( caseversion__in = cv_list ) . delete ( <EOL> permanent = True ) <EOL> def _bulk_insert_new_runcaseversions ( self , rcv_proxies ) : <EOL> """<STR_LIT>""" <EOL> self . runcaseversions . bulk_create ( rcv_proxies ) <EOL> def _bulk_update_runcaseversion_environments_for_lock ( self ) : <EOL> """<STR_LIT>""" <EOL> final_rcvs = RunCaseVersion . objects . filter ( run = self ) . select_related ( <EOL> "<STR_LIT>" ) . prefetch_related ( "<STR_LIT>" ) <EOL> final_rcv_ids = [ x . id for x in final_rcvs ] <EOL> prev_rcv_envs_set = set ( RunCaseVersion . environments . through . objects . filter ( <EOL> runcaseversion_id__in = final_rcv_ids ) . values_list ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> needed_rcv_envs_tuples = [ ] <EOL> run_env_ids = set ( <EOL> self . environments . values_list ( "<STR_LIT:id>" , flat = True ) ) <EOL> for rcv in final_rcvs : <EOL> case_env_ids = set ( [ x . id for x in rcv . caseversion . environments . all ( ) ] ) <EOL> for env in run_env_ids . intersection ( case_env_ids ) : <EOL> needed_rcv_envs_tuples . append ( ( rcv . id , env ) ) <EOL> needed_rcv_envs_set = set ( needed_rcv_envs_tuples ) <EOL> delete_rcv_envs = prev_rcv_envs_set - needed_rcv_envs_set <EOL> if len ( delete_rcv_envs ) : <EOL> delquery = Q ( ) <EOL> for combo in delete_rcv_envs : <EOL> delquery = delquery | Q ( <EOL> ** { "<STR_LIT>" : combo [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : combo [ <NUM_LIT:1> ] } ) <EOL> RunCaseVersion . environments . through . objects . filter ( delquery ) . delete ( ) <EOL> needed_rcv_envs_set = needed_rcv_envs_set - prev_rcv_envs_set <EOL> needed_rcv_envs = [ RunCaseVersion . environments . through ( <EOL> runcaseversion_id = needed [ <NUM_LIT:0> ] , <EOL> environment_id = needed [ <NUM_LIT:1> ] ) for needed in needed_rcv_envs_set ] <EOL> RunCaseVersion . environments . through . objects . bulk_create ( needed_rcv_envs ) <EOL> def _lock_caseversions_complete ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def result_summary ( self ) : <EOL> """<STR_LIT>""" <EOL> return result_summary ( Result . objects . filter ( runcaseversion__run = self ) ) <EOL> def completion ( self ) : <EOL> """<STR_LIT>""" <EOL> total = RunCaseVersion . environments . through . objects . filter ( <EOL> runcaseversion__run = self , <EOL> runcaseversion__deleted_on = None , <EOL> ) . count ( ) <EOL> skipped = Result . objects . filter ( <EOL> runcaseversion__run = self , <EOL> is_latest = True , <EOL> status = Result . STATUS . skipped ) . count ( ) <EOL> completed = Result . objects . filter ( <EOL> status__in = Result . COMPLETED_STATES , <EOL> is_latest = True , <EOL> runcaseversion__run = self ) . values ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) . distinct ( ) . count ( ) <EOL> try : <EOL> return float ( completed ) / ( total - skipped ) <EOL> except ZeroDivisionError : <EOL> return <NUM_LIT:0> <EOL> def completion_single_env ( self , env_id ) : <EOL> """<STR_LIT>""" <EOL> total = RunCaseVersion . objects . filter ( <EOL> environments = env_id , <EOL> run = self ) . count ( ) <EOL> skipped = Result . objects . filter ( <EOL> runcaseversion__run = self , <EOL> environment = env_id , <EOL> is_latest = True , <EOL> status = Result . STATUS . skipped ) . count ( ) <EOL> completed = Result . objects . filter ( <EOL> status__in = Result . COMPLETED_STATES , <EOL> is_latest = True , <EOL> environment = env_id , <EOL> runcaseversion__run = self ) . values ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) . distinct ( ) . count ( ) <EOL> try : <EOL> return float ( completed ) / ( total - skipped ) <EOL> except ZeroDivisionError : <EOL> return <NUM_LIT:0.0> <EOL> def _environment_intersection ( run , caseversion ) : <EOL> """<STR_LIT>""" <EOL> run_env_ids = set ( <EOL> run . environments . values_list ( "<STR_LIT:id>" , flat = True ) ) <EOL> case_env_ids = set ( <EOL> caseversion . environments . values_list ( "<STR_LIT:id>" , flat = True ) ) <EOL> return run_env_ids . intersection ( case_env_ids ) <EOL> class RunCaseVersion ( HasEnvironmentsModel , MTModel ) : <EOL> """<STR_LIT>""" <EOL> run = models . ForeignKey ( Run , related_name = "<STR_LIT>" ) <EOL> caseversion = models . ForeignKey ( CaseVersion , related_name = "<STR_LIT>" ) <EOL> order = models . IntegerField ( default = <NUM_LIT:0> , db_index = True ) <EOL> def __unicode__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( self . caseversion , self . run ) <EOL> def bug_urls ( self ) : <EOL> """<STR_LIT>""" <EOL> return set ( <EOL> StepResult . objects . filter ( <EOL> result__runcaseversion = self ) . exclude ( <EOL> bug_url = "<STR_LIT>" ) . values_list ( "<STR_LIT>" , flat = True ) . distinct ( ) <EOL> ) <EOL> class Meta : <EOL> ordering = [ "<STR_LIT>" ] <EOL> permissions = [ <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ] <EOL> def save ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> adding = False <EOL> if self . id is None : <EOL> adding = True <EOL> inherit_envs = kwargs . pop ( "<STR_LIT>" , True ) <EOL> ret = super ( RunCaseVersion , self ) . save ( * args , ** kwargs ) <EOL> if adding and inherit_envs : <EOL> self . environments . add ( <EOL> * _environment_intersection ( self . run , self . caseversion ) ) <EOL> return ret <EOL> def result_summary ( self ) : <EOL> """<STR_LIT>""" <EOL> return result_summary ( self . results . values ( ) ) <EOL> def completion ( self ) : <EOL> """<STR_LIT>""" <EOL> total = self . environments . count ( ) <EOL> skipped = self . results . filter ( <EOL> is_latest = True , <EOL> status = Result . STATUS . skipped ) . count ( ) <EOL> completed = self . results . filter ( <EOL> is_latest = True , <EOL> status__in = Result . COMPLETED_STATES ) . values ( <EOL> "<STR_LIT>" ) . distinct ( ) . count ( ) <EOL> try : <EOL> return float ( completed ) / ( total - skipped ) <EOL> except ZeroDivisionError : <EOL> return <NUM_LIT:0> <EOL> def testers ( self ) : <EOL> """<STR_LIT>""" <EOL> return User . objects . filter ( <EOL> pk__in = self . results . values_list ( "<STR_LIT>" , flat = True ) . distinct ( ) ) <EOL> def start ( self , environment = None , user = None ) : <EOL> """<STR_LIT>""" <EOL> envs = [ environment ] <EOL> try : <EOL> latest = self . results . get ( <EOL> is_latest = True , <EOL> tester = user , <EOL> environment = environment , <EOL> ) <EOL> if latest . status == Result . STATUS . skipped : <EOL> envs = self . environments . all ( ) <EOL> except ObjectDoesNotExist : <EOL> pass <EOL> for env in envs : <EOL> Result . objects . create ( <EOL> runcaseversion = self , <EOL> tester = user , <EOL> environment = env , <EOL> status = Result . STATUS . started , <EOL> user = user <EOL> ) <EOL> def get_result_method ( self , status ) : <EOL> """<STR_LIT>""" <EOL> status_methods = { <EOL> "<STR_LIT>" : self . result_pass , <EOL> "<STR_LIT>" : self . result_fail , <EOL> "<STR_LIT>" : self . result_invalid , <EOL> "<STR_LIT>" : self . result_block , <EOL> "<STR_LIT>" : self . result_skip , <EOL> } <EOL> return status_methods [ status ] <EOL> def result_pass ( self , environment = None , user = None ) : <EOL> """<STR_LIT>""" <EOL> return Result . objects . create ( <EOL> runcaseversion = self , <EOL> tester = user , <EOL> environment = environment , <EOL> status = Result . STATUS . passed , <EOL> user = user <EOL> ) <EOL> def result_skip ( self , environment = None , user = None ) : <EOL> """<STR_LIT>""" <EOL> envs = self . environments . all ( ) <EOL> created = [ ] <EOL> for env in envs : <EOL> created . append ( Result . objects . create ( <EOL> runcaseversion = self , <EOL> tester = user , <EOL> environment = env , <EOL> status = Result . STATUS . skipped , <EOL> user = user <EOL> ) ) <EOL> return created <EOL> def result_invalid ( self , environment = None , comment = "<STR_LIT>" , user = None ) : <EOL> """<STR_LIT>""" <EOL> return Result . objects . create ( <EOL> runcaseversion = self , <EOL> tester = user , <EOL> environment = environment , <EOL> status = Result . STATUS . invalidated , <EOL> comment = comment , <EOL> user = user , <EOL> ) <EOL> def result_block ( self , environment = None , comment = "<STR_LIT>" , user = None ) : <EOL> """<STR_LIT>""" <EOL> return Result . objects . create ( <EOL> runcaseversion = self , <EOL> tester = user , <EOL> environment = environment , <EOL> status = Result . STATUS . blocked , <EOL> comment = comment , <EOL> user = user , <EOL> ) <EOL> def result_fail ( self , environment = None , comment = "<STR_LIT>" , stepnumber = None , bug = "<STR_LIT>" , user = None ) : <EOL> """<STR_LIT>""" <EOL> result = Result . objects . create ( <EOL> runcaseversion = self , <EOL> tester = user , <EOL> environment = environment , <EOL> status = Result . STATUS . failed , <EOL> comment = comment , <EOL> user = user , <EOL> ) <EOL> if stepnumber is not None : <EOL> try : <EOL> step = self . caseversion . steps . get ( <EOL> number = stepnumber ) <EOL> except CaseStep . DoesNotExist : <EOL> pass <EOL> else : <EOL> stepresult = StepResult ( result = result , step = step ) <EOL> stepresult . status = StepResult . STATUS . failed <EOL> stepresult . bug_url = bug <EOL> stepresult . save ( user = user ) <EOL> self . save ( force_update = True , user = user ) <EOL> class RunSuite ( MTModel ) : <EOL> """<STR_LIT>""" <EOL> run = models . ForeignKey ( Run , related_name = "<STR_LIT>" ) <EOL> suite = models . ForeignKey ( Suite , related_name = "<STR_LIT>" ) <EOL> order = models . IntegerField ( default = <NUM_LIT:0> , db_index = True ) <EOL> def __unicode__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( self . suite , self . run ) <EOL> class Meta : <EOL> ordering = [ "<STR_LIT>" ] <EOL> class Result ( MTModel ) : <EOL> """<STR_LIT>""" <EOL> STATUS = Choices ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> REVIEW = Choices ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> ALL_STATES = STATUS . _full <EOL> PENDING_STATES = [ STATUS . assigned , STATUS . started ] <EOL> COMPLETED_STATES = [ STATUS . passed , STATUS . failed , STATUS . invalidated , <EOL> STATUS . blocked ] <EOL> FAILED_STATES = [ STATUS . failed , STATUS . blocked ] <EOL> tester = models . ForeignKey ( User , related_name = "<STR_LIT>" ) <EOL> runcaseversion = models . ForeignKey ( <EOL> RunCaseVersion , related_name = "<STR_LIT>" ) <EOL> environment = models . ForeignKey ( Environment , related_name = "<STR_LIT>" ) <EOL> status = models . CharField ( <EOL> max_length = <NUM_LIT:50> , db_index = True , choices = STATUS , default = STATUS . assigned ) <EOL> comment = models . TextField ( blank = True ) <EOL> is_latest = models . BooleanField ( default = True ) <EOL> review = models . CharField ( <EOL> max_length = <NUM_LIT:50> , db_index = True , choices = REVIEW , default = REVIEW . pending ) <EOL> reviewed_by = models . ForeignKey ( <EOL> User , related_name = "<STR_LIT>" , blank = True , null = True ) <EOL> def __unicode__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> self . runcaseversion , self . tester , self . environment , self . status ) <EOL> class Meta : <EOL> permissions = [ ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> def bug_urls ( self ) : <EOL> """<STR_LIT>""" <EOL> return set ( <EOL> self . stepresults . exclude ( <EOL> bug_url = "<STR_LIT>" ) . values_list ( "<STR_LIT>" , flat = True ) . distinct ( ) <EOL> ) <EOL> def save ( self , * args , ** kwargs ) : <EOL> if self . pk is None : <EOL> self . set_latest ( ) <EOL> super ( Result , self ) . save ( * args , ** kwargs ) <EOL> def set_latest ( self ) : <EOL> """<STR_LIT>""" <EOL> Result . objects . filter ( <EOL> tester = self . tester , <EOL> runcaseversion = self . runcaseversion , <EOL> environment = self . environment , <EOL> is_latest = True , <EOL> ) . exclude ( pk = self . pk ) . update ( is_latest = False ) <EOL> self . is_latest = True <EOL> class StepResult ( MTModel ) : <EOL> """<STR_LIT>""" <EOL> STATUS = Choices ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> result = models . ForeignKey ( Result , related_name = "<STR_LIT>" ) <EOL> step = models . ForeignKey ( CaseStep , related_name = "<STR_LIT>" ) <EOL> status = models . CharField ( <EOL> max_length = <NUM_LIT:50> , db_index = True , choices = STATUS , default = STATUS . passed ) <EOL> bug_url = models . URLField ( db_index = True , blank = True ) <EOL> def __unicode__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( self . result , self . step , self . status ) <EOL> def result_summary ( results ) : <EOL> """<STR_LIT>""" <EOL> states = Result . COMPLETED_STATES <EOL> result_ids = results . filter ( is_latest = True ) . values_list ( "<STR_LIT:id>" , flat = True ) <EOL> if not result_ids : <EOL> return dict ( ( s , <NUM_LIT:0> ) for s in states ) <EOL> cols = [ "<STR_LIT>" ] * len ( states ) <EOL> sql = "<STR_LIT>" . format ( <EOL> "<STR_LIT:U+002C>" . join ( cols ) , Result . _meta . db_table , "<STR_LIT:U+002C>" . join ( map ( str , result_ids ) ) <EOL> ) <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( sql , states ) <EOL> return dict ( zip ( states , cursor . fetchone ( ) ) ) </s>
