<s> import ast <EOL> import mock <EOL> import nose <EOL> import nose . tools as asserts <EOL> import json <EOL> import vcf as pyvcf <EOL> from cycledash import app , db <EOL> from common . helpers import tables , find <EOL> from workers . genotype_extractor import _extract <EOL> from test_projects_api import create_project_with_name <EOL> from test_bams_api import create_bam_with_name <EOL> from test_runs_api import create_run_with_uri <EOL> import helpers <EOL> class TestGenotypeExtractorWorker ( object ) : <EOL> def setUp ( self ) : <EOL> self . project = create_project_with_name ( '<STR_LIT>' ) <EOL> self . run = create_run_with_uri ( self . project [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> with tables ( db . engine , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) as ( con , projects , runs , genotypes ) : <EOL> genotypes . delete ( ) . execute ( ) <EOL> runs . delete ( ) . execute ( ) <EOL> projects . delete ( ) . execute ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> helpers . delete_all_records ( db ) <EOL> def test_extraction ( self ) : <EOL> _extract ( self . run [ '<STR_LIT:id>' ] ) <EOL> with tables ( db . engine , '<STR_LIT>' , '<STR_LIT>' ) as ( con , vcfs , genotypes ) : <EOL> vcf = vcfs . select ( vcfs . c . id == self . run [ '<STR_LIT:id>' ] ) . execute ( ) . fetchone ( ) <EOL> variants = genotypes . select ( <EOL> genotypes . c . vcf_id == self . run [ '<STR_LIT:id>' ] ) . execute ( ) . fetchall ( ) <EOL> variants = [ dict ( v ) for v in variants ] <EOL> expected_extant_columns = set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> extant_columns = set ( ast . literal_eval ( vcf [ '<STR_LIT>' ] ) ) <EOL> asserts . eq_ ( extant_columns , expected_extant_columns ) <EOL> asserts . eq_ ( vcf [ '<STR_LIT>' ] , <NUM_LIT:20> ) <EOL> asserts . eq_ ( vcf [ '<STR_LIT>' ] , len ( variants ) ) <EOL> with open ( '<STR_LIT>' , '<STR_LIT:r>' ) as vcffile : <EOL> vcflines = vcffile . readlines ( ) <EOL> expected_vcf_header = '<STR_LIT>' . join ( [ l for l in vcflines <EOL> if l . startswith ( '<STR_LIT:#>' ) ] ) <EOL> asserts . eq_ ( vcf [ '<STR_LIT>' ] + '<STR_LIT:\n>' , unicode ( expected_vcf_header ) ) <EOL> with open ( '<STR_LIT>' ) as vcffile : <EOL> vcfreader = pyvcf . Reader ( vcffile ) <EOL> expected_pyvcf_variants = [ r for r in vcfreader ] <EOL> assert_keys_in_variants ( expected_pyvcf_variants , variants ) <EOL> pyvcf_getters = [ lambda v : v . INFO [ '<STR_LIT>' ] , <EOL> lambda v : v . INFO [ '<STR_LIT>' ] , <EOL> lambda v : v . INFO [ '<STR_LIT>' ] , <EOL> lambda v : v . INFO [ '<STR_LIT>' ] , <EOL> lambda v : v . INFO [ '<STR_LIT>' ] ] <EOL> db_variant_getters = [ lambda v : int ( v [ '<STR_LIT>' ] ) , <EOL> lambda v : v [ '<STR_LIT>' ] , <EOL> lambda v : v [ '<STR_LIT>' ] , <EOL> lambda v : float ( v [ '<STR_LIT>' ] ) , <EOL> lambda v : float ( v [ '<STR_LIT>' ] ) ] <EOL> assert_keys_in_variants ( expected_pyvcf_variants , variants , <EOL> expected_getters = pyvcf_getters , <EOL> actual_getters = db_variant_getters ) <EOL> asserts . eq_ ( set ( vcfreader . samples ) , <EOL> set ( [ v [ '<STR_LIT>' ] for v in variants ] ) ) <EOL> def ensure_samples_equivalent ( sample_name ) : <EOL> def get_sample_data ( attr ) : <EOL> def get_from_variant ( variant ) : <EOL> sample = find ( variant . samples , lambda x : x . sample == sample_name ) <EOL> return getattr ( sample . data , attr ) <EOL> return get_from_variant <EOL> pyvcf_getters = [ get_sample_data ( '<STR_LIT>' ) , <EOL> get_sample_data ( '<STR_LIT>' ) , <EOL> get_sample_data ( '<STR_LIT>' ) , <EOL> get_sample_data ( '<STR_LIT>' ) , <EOL> get_sample_data ( '<STR_LIT>' ) , <EOL> get_sample_data ( '<STR_LIT>' ) , <EOL> lambda v : tuple ( get_sample_data ( '<STR_LIT>' ) ( v ) ) ] <EOL> db_variant_getters = [ lambda v : v [ '<STR_LIT>' ] , <EOL> lambda v : int ( v [ '<STR_LIT>' ] ) if v [ '<STR_LIT>' ] else None , <EOL> lambda v : int ( v [ '<STR_LIT>' ] ) , <EOL> lambda v : int ( v [ '<STR_LIT>' ] ) , <EOL> lambda v : int ( v [ '<STR_LIT>' ] ) , <EOL> lambda v : v [ '<STR_LIT>' ] , <EOL> lambda v : tuple ( ast . literal_eval ( v [ '<STR_LIT>' ] ) ) ] <EOL> actual_db_variants = [ v for v in variants <EOL> if v [ '<STR_LIT>' ] == sample_name ] <EOL> assert_keys_in_variants ( expected_pyvcf_variants , actual_db_variants , <EOL> expected_getters = pyvcf_getters , <EOL> actual_getters = db_variant_getters ) <EOL> ensure_samples_equivalent ( '<STR_LIT>' ) <EOL> ensure_samples_equivalent ( '<STR_LIT>' ) <EOL> def assert_keys_in_variants ( <EOL> expected_variants , actual_variants , <EOL> expected_getters = [ ] , actual_getters = [ ] ) : <EOL> """<STR_LIT>""" <EOL> def extract_expected_keys ( v ) : <EOL> base = [ v . CHROM , v . POS , v . REF , '<STR_LIT>' . join ( map ( str , v . ALT ) ) ] <EOL> extra = [ k ( v ) for k in expected_getters ] <EOL> return tuple ( base + extra ) <EOL> def extract_actual_keys ( v ) : <EOL> base = [ v [ '<STR_LIT>' ] , v [ '<STR_LIT>' ] , v [ '<STR_LIT>' ] , v [ '<STR_LIT>' ] ] <EOL> extra = [ k ( v ) for k in actual_getters ] <EOL> return tuple ( base + extra ) <EOL> expected_variants = set ( [ extract_expected_keys ( v ) for v in expected_variants ] ) <EOL> actual_variants = set ( [ extract_actual_keys ( v ) for v in actual_variants ] ) <EOL> asserts . eq_ ( expected_variants , actual_variants ) </s>
