<s> """<STR_LIT>""" <EOL> import json <EOL> from django . db import transaction <EOL> from . . core . auth import User <EOL> from . . tags . models import Tag <EOL> from . models import Case , CaseVersion , CaseStep , Suite , SuiteCase <EOL> class Importer ( object ) : <EOL> """<STR_LIT>""" <EOL> @ transaction . commit_on_success <EOL> def import_data ( self , productversion , case_data , force_dupes = False ) : <EOL> """<STR_LIT>""" <EOL> result = ImportResult ( ) <EOL> suite_importer = None <EOL> if "<STR_LIT>" in case_data : <EOL> suite_importer = SuiteImporter ( productversion . product ) <EOL> suite_importer . add_dicts ( case_data [ "<STR_LIT>" ] ) <EOL> if "<STR_LIT>" in case_data : <EOL> case_importer = CaseImporter ( productversion , suite_importer ) <EOL> result . append ( case_importer . import_cases ( <EOL> case_data [ "<STR_LIT>" ] , <EOL> force_dupes = force_dupes ) ) <EOL> if suite_importer : <EOL> result . append ( suite_importer . import_suites ( ) ) <EOL> return result <EOL> class CaseImporter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , productversion , suite_importer = None ) : <EOL> """<STR_LIT>""" <EOL> self . productversion = productversion <EOL> self . suite_importer = ( <EOL> suite_importer or SuiteImporter ( productversion . product ) <EOL> ) <EOL> self . tag_importer = TagImporter ( self . productversion . product ) <EOL> self . user_cache = UserCache ( ) <EOL> def import_cases ( self , case_dict_list , force_dupes = False ) : <EOL> """<STR_LIT>""" <EOL> result = ImportResult ( ) <EOL> for new_case in case_dict_list : <EOL> if not "<STR_LIT:name>" in new_case : <EOL> result . warn ( <EOL> ImportResult . SKIP_CASE_NO_NAME , <EOL> new_case , <EOL> ) <EOL> continue <EOL> if not force_dupes and CaseVersion . objects . filter ( <EOL> name = new_case [ "<STR_LIT:name>" ] , <EOL> productversion = self . productversion , <EOL> ) . exists ( ) : <EOL> result . warn ( <EOL> ImportResult . SKIP_CASE_NAME_CONFLICT , <EOL> new_case , <EOL> ) <EOL> continue <EOL> user = None <EOL> if "<STR_LIT>" in new_case : <EOL> try : <EOL> email = new_case [ "<STR_LIT>" ] <EOL> user = self . user_cache . get_user ( email ) <EOL> except User . DoesNotExist : <EOL> result . warn ( <EOL> ImportResult . WARN_USER_NOT_FOUND , <EOL> email , <EOL> ) <EOL> sid = transaction . savepoint ( ) <EOL> case = Case . objects . create ( <EOL> product = self . productversion . product , <EOL> idprefix = new_case . get ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ) <EOL> caseversion = CaseVersion . objects . create ( <EOL> productversion = self . productversion , <EOL> case = case , <EOL> name = new_case [ "<STR_LIT:name>" ] , <EOL> description = new_case . get ( "<STR_LIT:description>" , "<STR_LIT>" ) , <EOL> user = user , <EOL> ) <EOL> if "<STR_LIT>" in new_case : <EOL> try : <EOL> self . import_steps ( caseversion , new_case [ "<STR_LIT>" ] ) <EOL> except ValueError as e : <EOL> result . warn ( <EOL> e . args [ <NUM_LIT:0> ] , <EOL> new_case , <EOL> ) <EOL> transaction . savepoint_rollback ( sid ) <EOL> continue <EOL> else : <EOL> result . warn ( <EOL> ImportResult . WARN_NO_STEPS , <EOL> caseversion , <EOL> ) <EOL> if "<STR_LIT>" in new_case : <EOL> self . tag_importer . add_names ( caseversion , new_case [ "<STR_LIT>" ] ) <EOL> if "<STR_LIT>" in new_case : <EOL> self . suite_importer . add_names ( case , new_case [ "<STR_LIT>" ] ) <EOL> result . num_cases += <NUM_LIT:1> <EOL> transaction . savepoint_commit ( sid ) <EOL> self . tag_importer . import_tags ( ) <EOL> result . append ( self . suite_importer . import_suites ( ) ) <EOL> return result <EOL> def import_steps ( self , caseversion , step_data ) : <EOL> """<STR_LIT>""" <EOL> for step_num , new_step in enumerate ( step_data ) : <EOL> try : <EOL> casestep = CaseStep . objects . create ( <EOL> caseversion = caseversion , <EOL> number = step_num + <NUM_LIT:1> , <EOL> instruction = new_step [ "<STR_LIT>" ] , <EOL> expected = new_step . get ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ) <EOL> except KeyError : <EOL> raise ValueError ( ImportResult . SKIP_STEP_NO_INSTRUCTION ) <EOL> class UserCache ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cache = { } <EOL> def get_user ( self , email ) : <EOL> """<STR_LIT>""" <EOL> if email in self . cache : <EOL> return self . cache [ email ] <EOL> else : <EOL> try : <EOL> user = User . objects . get ( email = email ) <EOL> self . cache [ email ] = user <EOL> except User . DoesNotExist as e : <EOL> self . cache [ email ] = None <EOL> raise e <EOL> return self . cache [ email ] <EOL> class TagImporter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , product ) : <EOL> """<STR_LIT>""" <EOL> self . product = product <EOL> self . map = { } <EOL> def add_names ( self , caseversion , tag_names ) : <EOL> """<STR_LIT>""" <EOL> for tag_name in tag_names : <EOL> caseversions = self . map . setdefault ( tag_name , [ ] ) <EOL> caseversions . append ( caseversion ) <EOL> def import_tags ( self ) : <EOL> """<STR_LIT>""" <EOL> for tag_name , caseversions in self . map . items ( ) : <EOL> existing_tags = Tag . objects . filter ( <EOL> name = tag_name , <EOL> product__in = [ None , self . product ] , <EOL> ) . order_by ( "<STR_LIT>" ) <EOL> if list ( existing_tags ) : <EOL> tag = existing_tags [ <NUM_LIT:0> ] <EOL> else : <EOL> tag = Tag . objects . create ( <EOL> name = tag_name , <EOL> product = self . product , <EOL> ) <EOL> tag . caseversions . add ( * caseversions ) <EOL> self . map . clear ( ) <EOL> class SuiteImporter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , product ) : <EOL> """<STR_LIT>""" <EOL> self . product = product <EOL> self . map = { } <EOL> self . result = ImportResult ( ) <EOL> def add_names ( self , case , suite_names ) : <EOL> """<STR_LIT>""" <EOL> for suite_name in suite_names : <EOL> suite = self . map . setdefault ( suite_name , { } ) <EOL> cases = suite . setdefault ( "<STR_LIT>" , [ ] ) <EOL> cases . append ( case ) <EOL> def add_dicts ( self , suite_dicts ) : <EOL> """<STR_LIT>""" <EOL> for suite_dict in suite_dicts : <EOL> try : <EOL> suite = self . map . setdefault ( suite_dict [ "<STR_LIT:name>" ] , { } ) <EOL> suite . setdefault ( <EOL> "<STR_LIT:description>" , <EOL> suite_dict . get ( "<STR_LIT:description>" , "<STR_LIT>" ) , <EOL> ) <EOL> except KeyError : <EOL> self . result . warn ( <EOL> ImportResult . SKIP_SUITE_NO_NAME , <EOL> suite_dict , <EOL> ) <EOL> def import_suites ( self ) : <EOL> """<STR_LIT>""" <EOL> for suite_name , suite_data in self . map . items ( ) : <EOL> suite , created = Suite . objects . get_or_create ( <EOL> name = suite_name , <EOL> product = self . product , <EOL> defaults = { "<STR_LIT:description>" : suite_data . get ( "<STR_LIT:description>" , "<STR_LIT>" ) } , <EOL> ) <EOL> if created : <EOL> self . result . num_suites += <NUM_LIT:1> <EOL> if "<STR_LIT>" in suite_data : <EOL> for case in suite_data [ "<STR_LIT>" ] : <EOL> SuiteCase . objects . create ( case = case , suite = suite ) <EOL> self . map . clear ( ) <EOL> return self . result <EOL> class ImportResult ( object ) : <EOL> """<STR_LIT>""" <EOL> SKIP_SUITE_NO_NAME = "<STR_LIT>" <EOL> SKIP_CASE_NO_NAME = "<STR_LIT>" <EOL> SKIP_STEP_NO_INSTRUCTION = "<STR_LIT>" <EOL> SKIP_CASE_NAME_CONFLICT = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> WARN_NO_STEPS = "<STR_LIT>" <EOL> WARN_USER_NOT_FOUND = "<STR_LIT>" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . num_cases = <NUM_LIT:0> <EOL> self . num_suites = <NUM_LIT:0> <EOL> self . warnings = [ ] <EOL> def warn ( self , reason , item ) : <EOL> """<STR_LIT>""" <EOL> self . warnings . append ( { "<STR_LIT>" : reason , "<STR_LIT>" : item } ) <EOL> def append ( self , result ) : <EOL> """<STR_LIT>""" <EOL> self . num_cases += result . num_cases <EOL> self . num_suites += result . num_suites <EOL> self . warnings . extend ( result . warnings ) <EOL> def get_as_list ( self ) : <EOL> """<STR_LIT>""" <EOL> result_list = [ <EOL> "<STR_LIT>" . format ( x [ "<STR_LIT>" ] , json . dumps ( x [ "<STR_LIT>" ] , indent = <NUM_LIT:4> ) ) <EOL> for x in self . warnings <EOL> ] <EOL> result_list . append ( "<STR_LIT>" . format ( self . num_cases ) ) <EOL> result_list . append ( "<STR_LIT>" . format ( self . num_suites ) ) <EOL> return result_list </s>
