test_pipeline.py 文件源码

def test_staff(self):
        dfw2 = self.dfw.copy()
        dfw2 = 0

        staffing = Staffing(self.dfs, self.dfw, dfw2, self.dtr, self.staffing_parameters,self.config)
        roster, s, vars =staffing.staff()

        collectors_day0 =  reduce(lambda x,y: x+y, [s.getVal(vars[i,'DSSG',datetime(2011,11,11)]) for i in range(0,self.staffing_parameters['N'])])
        collectors_day1 =  reduce(lambda x,y: x+y, [s.getVal(vars[i,'DSSG',datetime(2011,11,12)]) for i in range(0,self.staffing_parameters['N'])])
        collectors_day2 =  reduce(lambda x,y: x+y, [s.getVal(vars[i,'DSSG',datetime(2011,11,13)]) for i in range(0,self.staffing_parameters['N'])])
        collectors_day5 =  reduce(lambda x,y: x+y, [s.getVal(vars[i,'DSSG',datetime(2011,11,16)]) for i in range(0,self.staffing_parameters['N'])])
        #Need 2 people on Monday, 3 people on Tuesday, and 1 (-> 2) people on Wednesday. Zero on other days.
        self.assertEqual(collectors_day0, 2)
        self.assertEqual(collectors_day1, 3)
        self.assertEqual(collectors_day2, 2)
        self.assertEqual(collectors_day5, 0)
        self.assertEqual(roster.shape[0], 1)
        self.assertEqual( list(roster.loc['DSSG',[datetime(2011,11,11),datetime(2011,11,12),datetime(2011,11,13)]].values), [collectors_day0,collectors_day1,collectors_day2])