<s> from datetime import datetime <EOL> import random <EOL> import string <EOL> from django . test import TestCase <EOL> from casexml . apps . case . tests . util import delete_all_xforms <EOL> from casexml . apps . stock . utils import get_current_ledger_transactions , get_current_ledger_state <EOL> from corehq . apps . commtrack . models import SQLProduct <EOL> from casexml . apps . stock . const import REPORT_TYPE_BALANCE <EOL> from casexml . apps . stock . models import StockReport , StockTransaction <EOL> from corehq . apps . commtrack . processing import StockProcessingResult <EOL> from corehq . apps . domain . shortcuts import create_domain <EOL> from corehq . form_processor . interfaces . processor import FormProcessorInterface <EOL> from corehq . form_processor . parsers . ledgers . helpers import StockReportHelper , StockTransactionHelper <EOL> from couchforms . models import XFormInstance <EOL> DOMAIN_MAX_LENGTH = <NUM_LIT> <EOL> class StockReportDomainTest ( TestCase ) : <EOL> def _get_name_for_domain ( self ) : <EOL> return '<STR_LIT>' . join ( <EOL> random . choice ( string . ascii_lowercase ) <EOL> for _ in range ( DOMAIN_MAX_LENGTH ) <EOL> ) <EOL> def create_report ( self , transactions = None , tag = None , date = None ) : <EOL> form = XFormInstance ( domain = self . domain ) <EOL> form . save ( ) <EOL> report = StockReportHelper . make_from_form ( <EOL> form , <EOL> date or datetime . utcnow ( ) , <EOL> tag or REPORT_TYPE_BALANCE , <EOL> transactions or [ ] , <EOL> ) <EOL> return report , form <EOL> def _create_models_for_stock_report_helper ( self , form , stock_report_helper ) : <EOL> processing_result = StockProcessingResult ( form , stock_report_helpers = [ stock_report_helper ] ) <EOL> processing_result . populate_models ( ) <EOL> processing_result . commit ( ) <EOL> def setUp ( self ) : <EOL> self . case_ids = { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:30> , '<STR_LIT>' : <NUM_LIT:50> } <EOL> self . section_ids = { '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : <NUM_LIT:9> } <EOL> self . product_ids = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : <NUM_LIT:5> } <EOL> self . domain = self . _get_name_for_domain ( ) <EOL> self . ledger_processor = FormProcessorInterface ( domain = self . domain ) . ledger_processor <EOL> create_domain ( self . domain ) <EOL> SQLProduct . objects . bulk_create ( [ <EOL> SQLProduct ( product_id = id ) for id in self . product_ids <EOL> ] ) <EOL> transactions_flat = [ ] <EOL> self . transactions = { } <EOL> for case , c_bal in self . case_ids . items ( ) : <EOL> for section , s_bal in self . section_ids . items ( ) : <EOL> for product , p_bal in self . product_ids . items ( ) : <EOL> bal = c_bal + s_bal + p_bal <EOL> transactions_flat . append ( <EOL> StockTransactionHelper ( <EOL> case_id = case , <EOL> section_id = section , <EOL> product_id = product , <EOL> action = '<STR_LIT>' , <EOL> quantity = bal <EOL> ) <EOL> ) <EOL> self . transactions . setdefault ( case , { } ) . setdefault ( section , { } ) [ product ] = bal <EOL> self . new_stock_report , self . form = self . create_report ( transactions_flat ) <EOL> self . _create_models_for_stock_report_helper ( self . form , self . new_stock_report ) <EOL> def tearDown ( self ) : <EOL> delete_all_xforms ( ) <EOL> StockReport . objects . all ( ) . delete ( ) <EOL> StockTransaction . objects . all ( ) . delete ( ) <EOL> def test_stock_report ( self ) : <EOL> self . create_report ( ) <EOL> filtered_stock_report = StockReport . objects . filter ( domain = self . domain ) <EOL> self . assertEquals ( filtered_stock_report . count ( ) , <NUM_LIT:1> ) <EOL> stock_report = filtered_stock_report . get ( ) <EOL> self . assertEquals ( stock_report . form_id , self . form . _id ) <EOL> self . assertEquals ( stock_report . domain , self . domain ) <EOL> def test_get_current_ledger_transactions ( self ) : <EOL> for case in self . case_ids : <EOL> transactions = get_current_ledger_transactions ( case ) <EOL> for section , products in transactions . items ( ) : <EOL> for product , trans in products . items ( ) : <EOL> self . assertEqual ( trans . stock_on_hand , self . transactions [ case ] [ section ] [ product ] ) <EOL> def _validate_case_data ( self , data , expected ) : <EOL> for section , products in data . items ( ) : <EOL> for product , trans in products . items ( ) : <EOL> self . assertEqual ( trans . stock_on_hand , expected [ section ] [ product ] ) <EOL> def _test_get_current_ledger_transactions ( self , tester_fn ) : <EOL> tester_fn ( self . transactions ) <EOL> date = datetime . utcnow ( ) <EOL> report , _ = self . create_report ( [ <EOL> StockTransactionHelper ( <EOL> case_id = '<STR_LIT>' , <EOL> section_id = '<STR_LIT>' , <EOL> product_id = '<STR_LIT>' , <EOL> action = '<STR_LIT>' , <EOL> quantity = <NUM_LIT> ) <EOL> ] , date = date ) <EOL> self . _create_models_for_stock_report_helper ( self . form , report ) <EOL> report , _ = self . create_report ( [ <EOL> StockTransactionHelper ( <EOL> case_id = '<STR_LIT>' , <EOL> section_id = '<STR_LIT>' , <EOL> product_id = '<STR_LIT>' , <EOL> action = '<STR_LIT>' , <EOL> quantity = <NUM_LIT:1> ) <EOL> ] , date = date ) <EOL> self . _create_models_for_stock_report_helper ( self . form , report ) <EOL> new_trans = self . transactions . copy ( ) <EOL> new_trans [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> tester_fn ( new_trans ) <EOL> def test_get_current_ledger_transactions_1 ( self ) : <EOL> def test_transactions ( expected ) : <EOL> for case in self . case_ids : <EOL> transactions = get_current_ledger_transactions ( case ) <EOL> self . _validate_case_data ( transactions , expected [ case ] ) <EOL> self . _test_get_current_ledger_transactions ( test_transactions ) <EOL> self . assertEqual ( { } , get_current_ledger_transactions ( '<STR_LIT>' ) ) <EOL> def test_get_current_ledger_state ( self ) : <EOL> def test_transactions ( expected ) : <EOL> state = get_current_ledger_state ( self . case_ids . keys ( ) ) <EOL> for case , sections in state . items ( ) : <EOL> self . _validate_case_data ( sections , expected [ case ] ) <EOL> self . _test_get_current_ledger_transactions ( test_transactions ) <EOL> self . assertEqual ( { } , get_current_ledger_state ( [ ] ) ) </s>
