<s> from __future__ import absolute_import <EOL> from collections import defaultdict <EOL> from changes . config import db <EOL> from changes . constants import Result , Status <EOL> from changes . models import Build , Job , TestCase , Source <EOL> def first ( key , iterable ) : <EOL> for x in iterable : <EOL> if key ( x ) : <EOL> return x <EOL> return None <EOL> def find_failure_origins ( build , test_failures ) : <EOL> """<STR_LIT>""" <EOL> project = build . project <EOL> if not test_failures : <EOL> return { } <EOL> last_pass = Build . query . join ( <EOL> Source , Source . id == Build . source_id , <EOL> ) . filter ( <EOL> Build . project == project , <EOL> Build . date_created <= build . date_created , <EOL> Build . status == Status . finished , <EOL> Build . result == Result . passed , <EOL> Build . id != build . id , <EOL> Source . patch == None , <EOL> ) . order_by ( Build . date_created . desc ( ) ) . first ( ) <EOL> if last_pass is None : <EOL> return { } <EOL> previous_runs = Build . query . join ( <EOL> Source , Source . id == build . source_id , <EOL> ) . filter ( <EOL> Build . project == project , <EOL> Build . date_created <= build . date_created , <EOL> Build . date_created >= last_pass . date_created , <EOL> Build . status == Status . finished , <EOL> Build . result . in_ ( [ Result . failed , Result . passed ] ) , <EOL> Build . id != build . id , <EOL> Build . id != last_pass . id , <EOL> Source . patch == None , <EOL> ) . order_by ( Build . date_created . desc ( ) ) [ : <NUM_LIT:100> ] <EOL> if not previous_runs : <EOL> return { } <EOL> queryset = db . session . query ( <EOL> TestCase . name_sha , Job . build_id , <EOL> ) . join ( <EOL> Job , Job . id == TestCase . job_id , <EOL> ) . filter ( <EOL> Job . build_id . in_ ( b . id for b in previous_runs ) , <EOL> Job . status == Status . finished , <EOL> Job . result == Result . failed , <EOL> TestCase . result == Result . failed , <EOL> TestCase . name_sha . in_ ( t . name_sha for t in test_failures ) , <EOL> ) . group_by ( <EOL> TestCase . name_sha , Job . build_id <EOL> ) <EOL> previous_test_failures = defaultdict ( set ) <EOL> for name_sha , build_id in queryset : <EOL> previous_test_failures [ build_id ] . add ( name_sha ) <EOL> failures_at_build = dict ( ) <EOL> searching = set ( t for t in test_failures ) <EOL> last_checked_run = build <EOL> for p_build in previous_runs : <EOL> p_build_failures = previous_test_failures [ p_build . id ] <EOL> for f_test in list ( searching ) : <EOL> if f_test . name_sha not in p_build_failures : <EOL> failures_at_build [ f_test ] = last_checked_run <EOL> searching . remove ( f_test ) <EOL> last_checked_run = p_build <EOL> for f_test in searching : <EOL> failures_at_build [ f_test ] = last_checked_run <EOL> return failures_at_build </s>
