<s> import urllib2 <EOL> import re <EOL> import tempfile <EOL> import h5py <EOL> import random <EOL> import csv <EOL> import numpy as np <EOL> import pytest <EOL> import httplib <EOL> from contextlib import closing <EOL> from params import Params <EOL> from postmethods import putAnnotation , getAnnotation <EOL> from ramon import H5AnnotationFile , getH5id , makeAnno , getId , getField , setField <EOL> import kvengine_to_test <EOL> import site_to_test <EOL> import makeunitdb <EOL> SITE_HOST = site_to_test . site <EOL> p = Params ( ) <EOL> p . token = '<STR_LIT>' <EOL> p . resolution = <NUM_LIT:0> <EOL> p . channels = [ '<STR_LIT>' ] <EOL> class Test_Ramon : <EOL> def setup_class ( self ) : <EOL> """<STR_LIT>""" <EOL> makeunitdb . createTestDB ( p . token , readonly = <NUM_LIT:0> ) <EOL> def teardown_class ( self ) : <EOL> """<STR_LIT>""" <EOL> makeunitdb . deleteTestDB ( p . token ) <EOL> def test_anno_minmal ( self ) : <EOL> """<STR_LIT>""" <EOL> tmpfile = tempfile . NamedTemporaryFile ( ) <EOL> h5fh = h5py . File ( tmpfile . name ) <EOL> idgrp = h5fh . create_group ( str ( <NUM_LIT:0> ) ) <EOL> h5fh . flush ( ) <EOL> tmpfile . seek ( <NUM_LIT:0> ) <EOL> p . annoid = putAnnotation ( p , tmpfile ) <EOL> h5ret = getAnnotation ( p ) <EOL> idgrpret = h5ret . get ( str ( p . annoid ) ) <EOL> assert idgrpret <EOL> assert ( idgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == <NUM_LIT:1> ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> mdgrpret = idgrpret [ '<STR_LIT>' ] <EOL> assert mdgrpret <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == <NUM_LIT:0.0> ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == <NUM_LIT:0> ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ : ] == '<STR_LIT>' ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ : ] == '<STR_LIT>' ) <EOL> def test_anno_full ( self ) : <EOL> """<STR_LIT>""" <EOL> tmpfile = tempfile . NamedTemporaryFile ( ) <EOL> h5fh = h5py . File ( tmpfile . name ) <EOL> idgrp = h5fh . create_group ( str ( <NUM_LIT:0> ) ) <EOL> mdgrp = idgrp . create_group ( "<STR_LIT>" ) <EOL> ann_status = random . randint ( <NUM_LIT:0> , <NUM_LIT:4> ) <EOL> ann_confidence = random . random ( ) <EOL> ann_author = '<STR_LIT>' <EOL> idgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , np . uint32 , data = <NUM_LIT:1> ) <EOL> mdgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , np . uint32 , data = ann_status ) <EOL> mdgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , np . float , data = ann_confidence ) <EOL> mdgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , dtype = h5py . special_dtype ( vlen = str ) , data = ann_author ) <EOL> h5fh . flush ( ) <EOL> tmpfile . seek ( <NUM_LIT:0> ) <EOL> p . annoid = putAnnotation ( p , tmpfile ) <EOL> h5ret = getAnnotation ( p ) <EOL> idgrpret = h5ret . get ( str ( p . annoid ) ) <EOL> assert idgrpret <EOL> assert ( idgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == <NUM_LIT:1> ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> mdgrpret = idgrpret [ '<STR_LIT>' ] <EOL> assert mdgrpret <EOL> assert ( abs ( mdgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] - ann_confidence ) < <NUM_LIT> ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == ann_status ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ : ] == ann_author ) <EOL> tmpfile . close ( ) <EOL> def test_anno_update ( self ) : <EOL> """<STR_LIT>""" <EOL> tmpfile = tempfile . NamedTemporaryFile ( ) <EOL> h5fh = h5py . File ( tmpfile . name ) <EOL> idgrp = h5fh . create_group ( str ( p . annoid ) ) <EOL> mdgrp = idgrp . create_group ( "<STR_LIT>" ) <EOL> ann_status = random . randint ( <NUM_LIT:0> , <NUM_LIT:4> ) <EOL> ann_confidence = random . random ( ) <EOL> ann_author = '<STR_LIT>' <EOL> idgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , np . uint32 , data = <NUM_LIT:1> ) <EOL> mdgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , np . uint32 , data = ann_status ) <EOL> mdgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , np . float , data = ann_confidence ) <EOL> mdgrp . create_dataset ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , dtype = h5py . special_dtype ( vlen = str ) , data = ann_author ) <EOL> h5fh . flush ( ) <EOL> tmpfile . seek ( <NUM_LIT:0> ) <EOL> p . field = '<STR_LIT>' <EOL> p . annoid = putAnnotation ( p , tmpfile ) <EOL> p . field = None <EOL> h5ret = getAnnotation ( p ) <EOL> idgrpret = h5ret . get ( str ( p . annoid ) ) <EOL> assert idgrpret <EOL> assert ( idgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == <NUM_LIT:1> ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> assert not idgrpret . get ( '<STR_LIT>' ) <EOL> mdgrpret = idgrpret [ '<STR_LIT>' ] <EOL> assert mdgrpret <EOL> assert ( abs ( mdgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] - ann_confidence ) < <NUM_LIT> ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == ann_status ) <EOL> assert ( mdgrpret [ '<STR_LIT>' ] [ : ] == ann_author ) <EOL> def test_anno_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> ( base , suffix ) = SITE_HOST . split ( "<STR_LIT:/>" , <NUM_LIT:1> ) <EOL> except : <EOL> base = SITE_HOST <EOL> suffix = None <EOL> conn = httplib . HTTPConnection ( base ) <EOL> if suffix : <EOL> conn . request ( '<STR_LIT>' , '<STR_LIT>' . format ( suffix , '<STR_LIT>' , '<STR_LIT>' , p . annoid ) ) <EOL> else : <EOL> conn . request ( '<STR_LIT>' , '<STR_LIT>' . format ( '<STR_LIT>' , '<STR_LIT>' , p . annoid ) ) <EOL> resp = conn . getresponse ( ) <EOL> content = resp . read ( ) <EOL> assert content == '<STR_LIT>' <EOL> url = "<STR_LIT>" . format ( SITE_HOST , '<STR_LIT>' , '<STR_LIT>' , p . annoid ) <EOL> with pytest . raises ( urllib2 . HTTPError ) : <EOL> urllib2 . urlopen ( url ) <EOL> def test_anno_upload ( self ) : <EOL> """<STR_LIT>""" <EOL> for anntype in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] : <EOL> annoid = <NUM_LIT:0> <EOL> f = H5AnnotationFile ( anntype , annoid ) <EOL> putid1 = putAnnotation ( p , f ) <EOL> newid = putid1 + <NUM_LIT:1> <EOL> f = H5AnnotationFile ( anntype , newid ) <EOL> putid2 = putAnnotation ( p , f ) <EOL> p . annoid = putid1 <EOL> assert ( putid1 == getId ( p ) ) <EOL> p . annoid = putid2 <EOL> assert ( putid2 == getId ( p ) ) <EOL> def test_annotation_field ( self ) : <EOL> """<STR_LIT>""" <EOL> makeAnno ( p , <NUM_LIT:1> ) <EOL> status = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:status>' , status ) <EOL> f = getField ( p , '<STR_LIT:status>' ) <EOL> assert status == int ( f . read ( ) ) <EOL> confidence = random . random ( ) <EOL> f = setField ( p , '<STR_LIT>' , confidence ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert confidence - float ( f . read ( ) ) < <NUM_LIT> <EOL> def test_synapse_field ( self ) : <EOL> """<STR_LIT>""" <EOL> makeAnno ( p , <NUM_LIT:2> ) <EOL> synapse_type = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , synapse_type ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert synapse_type == int ( f . read ( ) ) <EOL> weight = random . random ( ) <EOL> f = setField ( p , '<STR_LIT>' , weight ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert weight - float ( f . read ( ) ) < <NUM_LIT> <EOL> status = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:status>' , status ) <EOL> f = getField ( p , '<STR_LIT:status>' ) <EOL> assert status == int ( f . read ( ) ) <EOL> seeds = [ random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) ] <EOL> f = setField ( p , '<STR_LIT>' , '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in seeds ] ) ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in seeds ] ) == f . read ( ) <EOL> def test_seed_field ( self ) : <EOL> """<STR_LIT>""" <EOL> makeAnno ( p , <NUM_LIT:3> ) <EOL> parent = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , parent ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert parent == int ( f . read ( ) ) <EOL> source = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:source>' , source ) <EOL> f = getField ( p , '<STR_LIT:source>' ) <EOL> assert source == int ( f . read ( ) ) <EOL> cubelocation = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , cubelocation ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert cubelocation == int ( f . read ( ) ) <EOL> position = [ random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) ] <EOL> f = setField ( p , '<STR_LIT>' , '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in position ] ) ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in position ] ) == f . read ( ) <EOL> def test_segment_field ( self ) : <EOL> """<STR_LIT>""" <EOL> makeAnno ( p , <NUM_LIT:4> ) <EOL> parentseed = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , parentseed ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert parentseed == int ( f . read ( ) ) <EOL> segmentclass = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , segmentclass ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert segmentclass == int ( f . read ( ) ) <EOL> neuron = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , neuron ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert neuron == int ( f . read ( ) ) <EOL> status = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:status>' , status ) <EOL> f = getField ( p , '<STR_LIT:status>' ) <EOL> assert status == int ( f . read ( ) ) <EOL> synapses = [ random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) ] <EOL> f = setField ( p , '<STR_LIT>' , '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in synapses ] ) ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in synapses ] ) == f . read ( ) <EOL> def test_neuron_field ( self ) : <EOL> """<STR_LIT>""" <EOL> makeAnno ( p , <NUM_LIT:5> ) <EOL> status = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:status>' , status ) <EOL> f = getField ( p , '<STR_LIT:status>' ) <EOL> assert status == int ( f . read ( ) ) <EOL> status = [ random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) ] <EOL> f = setField ( p , '<STR_LIT>' , "<STR_LIT:U+002C>" . join ( [ str ( i ) for i in status ] ) ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert "<STR_LIT:U+002C>" . join ( [ str ( i ) for i in status ] ) == f . read ( ) <EOL> def test_organelle_field ( self ) : <EOL> """<STR_LIT>""" <EOL> makeAnno ( p , <NUM_LIT:6> ) <EOL> parentseed = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , parentseed ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert parentseed == int ( f . read ( ) ) <EOL> organelleclass = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , organelleclass ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert organelleclass == int ( f . read ( ) ) <EOL> status = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:status>' , status ) <EOL> f = getField ( p , '<STR_LIT:status>' ) <EOL> assert status == int ( f . read ( ) ) <EOL> seeds = [ random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) , random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) ] <EOL> f = setField ( p , '<STR_LIT>' , '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in seeds ] ) ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert '<STR_LIT:U+002C>' . join ( [ str ( i ) for i in seeds ] ) == f . read ( ) <EOL> def test_wrong_field ( self ) : <EOL> makeAnno ( p , <NUM_LIT:2> ) <EOL> synapse_type = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT>' , synapse_type ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert synapse_type == int ( f . read ( ) ) <EOL> weight = random . random ( ) <EOL> f = setField ( p , '<STR_LIT>' , weight ) <EOL> f = getField ( p , '<STR_LIT>' ) <EOL> assert weight - float ( f . read ( ) ) < <NUM_LIT> <EOL> status = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) <EOL> f = setField ( p , '<STR_LIT:status>' , status ) <EOL> f = getField ( p , '<STR_LIT:status>' ) <EOL> assert status == int ( f . read ( ) ) <EOL> url = "<STR_LIT>" . format ( SITE_HOST , '<STR_LIT>' , '<STR_LIT>' , p . annoid ) <EOL> with pytest . raises ( urllib2 . HTTPError ) : <EOL> req = urllib2 . Request ( url ) <EOL> f = urllib2 . urlopen ( url ) <EOL> url = "<STR_LIT>" . format ( SITE_HOST , '<STR_LIT>' , '<STR_LIT>' , p . annoid ) <EOL> with pytest . raises ( urllib2 . HTTPError ) : <EOL> req = urllib2 . Request ( url ) <EOL> f = urllib2 . urlopen ( url ) </s>
