<s> from django . db import models <EOL> from django . test import TestCase <EOL> from django . utils import unittest <EOL> from google . appengine . api . datastore import Key <EOL> try : <EOL> from djangoappengine . mapreduce . input_readers import DjangoModelInputReader , DjangoModelIterator <EOL> import mapreduce <EOL> from mapreduce import input_readers <EOL> from mapreduce . lib import key_range <EOL> from mapreduce import model <EOL> except ImportError : <EOL> mapreduce = None <EOL> class TestModel ( models . Model ) : <EOL> test_property = models . IntegerField ( default = <NUM_LIT:0> ) <EOL> def __unicode__ ( self ) : <EOL> return str ( self . test_property ) <EOL> ENTITY_KIND = '<STR_LIT>' % ( TestModel . __module__ , TestModel . __name__ ) <EOL> def key ( entity_id , kind = TestModel ) : <EOL> return Key . from_path ( kind . _meta . db_table , entity_id ) <EOL> @ unittest . skipUnless ( mapreduce , '<STR_LIT>' ) <EOL> class DjangoModelInputReaderTest ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testValidate_Passes ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> "<STR_LIT>" : ENTITY_KIND , <EOL> } <EOL> mapper_spec = model . MapperSpec ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> params , <NUM_LIT:1> ) <EOL> DjangoModelInputReader . validate ( mapper_spec ) <EOL> def testValidate_NoEntityFails ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> mapper_spec = model . MapperSpec ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> params , <NUM_LIT:1> ) <EOL> self . assertRaises ( input_readers . BadReaderParamsError , <EOL> DjangoModelInputReader . validate , <EOL> mapper_spec ) <EOL> def testValidate_BadEntityKind ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> "<STR_LIT>" : "<STR_LIT:foo>" , <EOL> } <EOL> mapper_spec = model . MapperSpec ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> params , <NUM_LIT:1> ) <EOL> self . assertRaises ( input_readers . BadReaderParamsError , <EOL> DjangoModelInputReader . validate , <EOL> mapper_spec ) <EOL> def testValidate_BadNamespace ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> "<STR_LIT>" : ENTITY_KIND , <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> } <EOL> mapper_spec = model . MapperSpec ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> params , <NUM_LIT:1> ) <EOL> self . assertRaises ( input_readers . BadReaderParamsError , <EOL> DjangoModelInputReader . validate , <EOL> mapper_spec ) <EOL> def testGeneratorWithKeyRange ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_entities = [ ] <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:100> ) : <EOL> entity = TestModel ( test_property = i ) <EOL> entity . save ( ) <EOL> expected_entities . append ( entity ) <EOL> params = { <EOL> "<STR_LIT>" : ENTITY_KIND , <EOL> } <EOL> mapper_spec = model . MapperSpec ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> params , <NUM_LIT:1> ) <EOL> input_ranges = DjangoModelInputReader . split_input ( mapper_spec ) <EOL> entities = [ ] <EOL> for query_range in input_ranges : <EOL> for entity in query_range : <EOL> entities . append ( entity ) <EOL> self . assertEquals ( <NUM_LIT:100> , len ( entities ) ) <EOL> self . assertEquals ( expected_entities , entities ) <EOL> @ unittest . skipUnless ( mapreduce , '<STR_LIT>' ) <EOL> class DjangoModelIteratorTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> expected_entities = [ ] <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:100> ) : <EOL> entity = TestModel ( test_property = i ) <EOL> entity . save ( ) <EOL> expected_entities . append ( entity ) <EOL> self . expected_entities = expected_entities <EOL> def testCursors ( self ) : <EOL> qs = model . QuerySpec ( TestModel , model_class_path = ENTITY_KIND ) <EOL> kr = key_range . KeyRange ( key_start = key ( <NUM_LIT:1> ) , key_end = key ( <NUM_LIT> ) , direction = "<STR_LIT>" ) <EOL> json = { '<STR_LIT>' : kr . to_json ( ) , '<STR_LIT>' : qs . to_json ( ) , '<STR_LIT>' : None } <EOL> entities = [ ] <EOL> while True : <EOL> model_iter = DjangoModelIterator . from_json ( json ) <EOL> c = False <EOL> count = <NUM_LIT:0> <EOL> for entity in model_iter : <EOL> count += <NUM_LIT:1> <EOL> entities . append ( entity ) <EOL> if count == <NUM_LIT:10> : <EOL> c = True <EOL> break <EOL> if c : <EOL> json = model_iter . to_json ( ) <EOL> else : <EOL> break <EOL> self . assertEquals ( <NUM_LIT:100> , len ( entities ) ) <EOL> self . assertEquals ( self . expected_entities , entities ) </s>
