<s> from __future__ import absolute_import <EOL> from collections import defaultdict <EOL> from flask_restful . reqparse import RequestParser <EOL> from itertools import groupby <EOL> from sqlalchemy . orm import contains_eager , joinedload , subqueryload_all <EOL> from typing import List <EOL> from uuid import UUID <EOL> from changes . api . base import APIView <EOL> from changes . api . serializer . models . testcase import TestCaseWithOriginCrumbler <EOL> from changes . config import db <EOL> from changes . constants import Result , Status <EOL> from changes . lib import build_type <EOL> from changes . models . build import Build , BuildPriority <EOL> from changes . models . buildseen import BuildSeen <EOL> from changes . models . event import Event <EOL> from changes . models . failurereason import FailureReason <EOL> from changes . models . job import Job <EOL> from changes . models . jobstep import JobStep <EOL> from changes . models . source import Source <EOL> from changes . models . test import TestCase <EOL> from changes . models . user import User <EOL> from changes . utils . originfinder import find_failure_origins <EOL> def find_changed_tests ( current_build , previous_build , limit = <NUM_LIT> ) : <EOL> current_job_ids = [ j . id . hex for j in current_build . jobs ] <EOL> previous_job_ids = [ j . id . hex for j in previous_build . jobs ] <EOL> if not ( current_job_ids and previous_job_ids ) : <EOL> return [ ] <EOL> current_job_clause = '<STR_LIT:U+002CU+0020>' . join ( <EOL> '<STR_LIT>' % i for i in range ( len ( current_job_ids ) ) <EOL> ) <EOL> previous_job_clause = '<STR_LIT:U+002CU+0020>' . join ( <EOL> '<STR_LIT>' % i for i in range ( len ( previous_job_ids ) ) <EOL> ) <EOL> params = { } <EOL> for idx , job_id in enumerate ( current_job_ids ) : <EOL> params [ '<STR_LIT>' % idx ] = job_id <EOL> for idx , job_id in enumerate ( previous_job_ids ) : <EOL> params [ '<STR_LIT>' % idx ] = job_id <EOL> query = """<STR_LIT>""" % { <EOL> '<STR_LIT>' : current_job_clause , <EOL> '<STR_LIT>' : previous_job_clause <EOL> } <EOL> total = db . session . query ( <EOL> '<STR_LIT:count>' <EOL> ) . from_statement ( <EOL> '<STR_LIT>' % ( query , ) <EOL> ) . params ( ** params ) . scalar ( ) <EOL> if not total : <EOL> return { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> results = db . session . query ( <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> ) . from_statement ( <EOL> '<STR_LIT>' % ( query , limit ) <EOL> ) . params ( ** params ) <EOL> all_test_ids = set ( ) <EOL> for c_id , p_id in results : <EOL> if c_id : <EOL> all_test_ids . add ( c_id ) <EOL> else : <EOL> all_test_ids . add ( p_id ) <EOL> test_map = dict ( <EOL> ( t . id , t ) for t in TestCase . query . filter ( <EOL> TestCase . id . in_ ( all_test_ids ) , <EOL> ) . options ( <EOL> joinedload ( '<STR_LIT>' , innerjoin = True ) , <EOL> ) <EOL> ) <EOL> diff = [ ] <EOL> for c_id , p_id in results : <EOL> if p_id : <EOL> diff . append ( ( '<STR_LIT:->' , test_map [ UUID ( p_id ) ] ) ) <EOL> else : <EOL> diff . append ( ( '<STR_LIT:+>' , test_map [ UUID ( c_id ) ] ) ) <EOL> return { <EOL> '<STR_LIT>' : total , <EOL> '<STR_LIT>' : sorted ( diff , key = lambda x : ( x [ <NUM_LIT:1> ] . package , x [ <NUM_LIT:1> ] . name ) ) , <EOL> } <EOL> def get_failure_reasons ( build ) : <EOL> from changes . buildfailures import registry <EOL> rows = FailureReason . query . join ( <EOL> JobStep , JobStep . id == FailureReason . step_id , <EOL> ) . filter ( <EOL> FailureReason . build_id == build . id , <EOL> JobStep . replacement_id . is_ ( None ) , <EOL> ) <EOL> failure_reasons = [ ] <EOL> for row in rows : <EOL> failure_reasons . append ( { <EOL> '<STR_LIT:id>' : row . reason , <EOL> '<STR_LIT>' : registry [ row . reason ] . get_html_label ( build ) , <EOL> '<STR_LIT>' : row . step_id , <EOL> '<STR_LIT>' : row . job_id , <EOL> '<STR_LIT:data>' : dict ( row . data or { } ) , <EOL> } ) <EOL> return failure_reasons <EOL> def get_parents_last_builds ( build ) : <EOL> if build . source . patch : <EOL> parents = [ build . source . patch . parent_revision_sha ] <EOL> elif build . source . revision : <EOL> parents = build . source . revision . parents <EOL> if parents : <EOL> parent_builds = list ( Build . query . filter ( <EOL> Build . project == build . project , <EOL> Build . status == Status . finished , <EOL> Build . id != build . id , <EOL> ) . join ( <EOL> Source , Build . source_id == Source . id , <EOL> ) . options ( <EOL> contains_eager ( '<STR_LIT:source>' ) . joinedload ( '<STR_LIT>' ) , <EOL> ) . filter ( <EOL> Source . revision_sha . in_ ( parents ) , <EOL> * build_type . get_any_commit_build_filters ( ) <EOL> ) . order_by ( Build . date_created . desc ( ) ) ) <EOL> if parent_builds : <EOL> return [ <EOL> list ( builds ) [ <NUM_LIT:0> ] <EOL> for sha , builds in groupby ( <EOL> parent_builds , <EOL> lambda rev : rev . source . revision_sha <EOL> ) <EOL> ] <EOL> return [ ] <EOL> class BuildDetailsAPIView ( APIView ) : <EOL> post_parser = RequestParser ( ) <EOL> post_parser . add_argument ( '<STR_LIT>' , choices = BuildPriority . _member_names_ ) <EOL> def get ( self , build_id ) : <EOL> build = Build . query . options ( <EOL> joinedload ( '<STR_LIT>' , innerjoin = True ) , <EOL> joinedload ( '<STR_LIT>' ) , <EOL> joinedload ( '<STR_LIT:source>' ) . joinedload ( '<STR_LIT>' ) , <EOL> subqueryload_all ( '<STR_LIT>' ) , <EOL> ) . get ( build_id ) <EOL> if build is None : <EOL> return '<STR_LIT>' , <NUM_LIT> <EOL> try : <EOL> most_recent_run = Build . query . filter ( <EOL> Build . project == build . project , <EOL> Build . date_created < build . date_created , <EOL> Build . status == Status . finished , <EOL> Build . id != build . id , <EOL> ) . join ( <EOL> Source , Build . source_id == Source . id , <EOL> ) . filter ( <EOL> * build_type . get_any_commit_build_filters ( ) <EOL> ) . options ( <EOL> contains_eager ( '<STR_LIT:source>' ) . joinedload ( '<STR_LIT>' ) , <EOL> joinedload ( '<STR_LIT>' ) , <EOL> ) . order_by ( Build . date_created . desc ( ) ) [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> most_recent_run = None <EOL> jobs = list ( Job . query . filter ( <EOL> Job . build_id == build . id , <EOL> ) ) <EOL> if not jobs : <EOL> test_failures = [ ] <EOL> num_test_failures = <NUM_LIT:0> <EOL> else : <EOL> test_failures = TestCase . query . options ( <EOL> joinedload ( '<STR_LIT>' , innerjoin = True ) , <EOL> ) . filter ( <EOL> TestCase . job_id . in_ ( [ j . id for j in jobs ] ) , <EOL> TestCase . result == Result . failed , <EOL> ) . order_by ( TestCase . name . asc ( ) ) <EOL> num_test_failures = test_failures . count ( ) <EOL> test_failures = test_failures [ : <NUM_LIT> ] <EOL> failures_by_job = defaultdict ( list ) <EOL> for failure in test_failures : <EOL> failures_by_job [ failure . job ] . append ( failure ) <EOL> failure_origins = find_failure_origins ( <EOL> build , test_failures ) <EOL> for test_failure in test_failures : <EOL> test_failure . origin = failure_origins . get ( test_failure ) <EOL> if most_recent_run and build . status == Status . finished : <EOL> changed_tests = find_changed_tests ( build , most_recent_run ) <EOL> else : <EOL> changed_tests = [ ] <EOL> seen_by = list ( User . query . join ( <EOL> BuildSeen , BuildSeen . user_id == User . id , <EOL> ) . filter ( <EOL> BuildSeen . build_id == build . id , <EOL> ) ) <EOL> extended_serializers = { <EOL> TestCase : TestCaseWithOriginCrumbler ( ) , <EOL> } <EOL> event_list = list ( Event . query . filter ( <EOL> Event . item_id == build . id , <EOL> ) . order_by ( Event . date_created . desc ( ) ) ) <EOL> context = self . serialize ( build ) <EOL> context . update ( { <EOL> '<STR_LIT>' : jobs , <EOL> '<STR_LIT>' : seen_by , <EOL> '<STR_LIT>' : event_list , <EOL> '<STR_LIT>' : get_failure_reasons ( build ) , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : num_test_failures , <EOL> '<STR_LIT>' : self . serialize ( test_failures , extended_serializers ) , <EOL> } , <EOL> '<STR_LIT>' : self . serialize ( changed_tests , extended_serializers ) , <EOL> '<STR_LIT>' : self . serialize ( get_parents_last_builds ( build ) ) , <EOL> } ) <EOL> return self . respond ( context ) <EOL> def post ( self , build_id ) : <EOL> build = Build . query . options ( <EOL> joinedload ( '<STR_LIT>' , innerjoin = True ) , <EOL> joinedload ( '<STR_LIT>' ) , <EOL> joinedload ( '<STR_LIT:source>' ) . joinedload ( '<STR_LIT>' ) , <EOL> ) . get ( build_id ) <EOL> if build is None : <EOL> return '<STR_LIT>' , <NUM_LIT> <EOL> args = self . post_parser . parse_args ( ) <EOL> if args . priority is not None : <EOL> build . priority = BuildPriority [ args . priority ] <EOL> db . session . add ( build ) <EOL> context = self . serialize ( build ) <EOL> return self . respond ( context , serialize = False ) </s>
