How to use get_history method in fMBT

Best Python code snippet using fMBT_python

test_db.py

Source:test_db.py

...53            self.db.add_record(msg_id, rec)54        return msg_ids55    56    def test_add_record(self):57        before = self.db.get_history()58        self.load_records(5)59        after = self.db.get_history()60        self.assertEqual(len(after), len(before)+5)61        self.assertEqual(after[:-5],before)62        63    def test_drop_record(self):64        msg_id = self.load_records()[-1]65        rec = self.db.get_record(msg_id)66        self.db.drop_record(msg_id)67        self.assertRaises(KeyError,self.db.get_record, msg_id)68    69    def _round_to_millisecond(self, dt):70        """necessary because mongodb rounds microseconds"""71        micro = dt.microsecond72        extra = int(str(micro)[-3:])73        return dt - timedelta(microseconds=extra)74    75    def test_update_record(self):76        now = self._round_to_millisecond(datetime.now())77        # 78        msg_id = self.db.get_history()[-1]79        rec1 = self.db.get_record(msg_id)80        data = {'stdout': 'hello there', 'completed' : now}81        self.db.update_record(msg_id, data)82        rec2 = self.db.get_record(msg_id)83        self.assertEqual(rec2['stdout'], 'hello there')84        self.assertEqual(rec2['completed'], now)85        rec1.update(data)86        self.assertEqual(rec1, rec2)87    88    # def test_update_record_bad(self):89    #     """test updating nonexistant records"""90    #     msg_id = str(uuid.uuid4())91    #     data = {'stdout': 'hello there'}92    #     self.assertRaises(KeyError, self.db.update_record, msg_id, data)93    def test_find_records_dt(self):94        """test finding records by date"""95        hist = self.db.get_history()96        middle = self.db.get_record(hist[len(hist)//2])97        tic = middle['submitted']98        before = self.db.find_records({'submitted' : {'$lt' : tic}})99        after = self.db.find_records({'submitted' : {'$gte' : tic}})100        self.assertEqual(len(before)+len(after),len(hist))101        for b in before:102            self.assertTrue(b['submitted'] < tic)103        for a in after:104            self.assertTrue(a['submitted'] >= tic)105        same = self.db.find_records({'submitted' : tic})106        for s in same:107            self.assertTrue(s['submitted'] == tic)108    109    def test_find_records_keys(self):110        """test extracting subset of record keys"""111        found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])112        for rec in found:113            self.assertEqual(set(rec.keys()), set(['msg_id', 'submitted', 'completed']))114    115    def test_find_records_msg_id(self):116        """ensure msg_id is always in found records"""117        found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])118        for rec in found:119            self.assertTrue('msg_id' in rec.keys())120        found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted'])121        for rec in found:122            self.assertTrue('msg_id' in rec.keys())123        found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['msg_id'])124        for rec in found:125            self.assertTrue('msg_id' in rec.keys())126    127    def test_find_records_in(self):128        """test finding records with '$in','$nin' operators"""129        hist = self.db.get_history()130        even = hist[::2]131        odd = hist[1::2]132        recs = self.db.find_records({ 'msg_id' : {'$in' : even}})133        found = [ r['msg_id'] for r in recs ]134        self.assertEqual(set(even), set(found))135        recs = self.db.find_records({ 'msg_id' : {'$nin' : even}})136        found = [ r['msg_id'] for r in recs ]137        self.assertEqual(set(odd), set(found))138    139    def test_get_history(self):140        msg_ids = self.db.get_history()141        latest = datetime(1984,1,1)142        for msg_id in msg_ids:143            rec = self.db.get_record(msg_id)144            newt = rec['submitted']145            self.assertTrue(newt >= latest)146            latest = newt147        msg_id = self.load_records(1)[-1]148        self.assertEqual(self.db.get_history()[-1],msg_id)149    150    def test_datetime(self):151        """get/set timestamps with datetime objects"""152        msg_id = self.db.get_history()[-1]153        rec = self.db.get_record(msg_id)154        self.assertTrue(isinstance(rec['submitted'], datetime))155        self.db.update_record(msg_id, dict(completed=datetime.now()))156        rec = self.db.get_record(msg_id)157        self.assertTrue(isinstance(rec['completed'], datetime))158    def test_drop_matching(self):159        msg_ids = self.load_records(10)160        query = {'msg_id' : {'$in':msg_ids}}161        self.db.drop_matching_records(query)162        recs = self.db.find_records(query)163        self.assertEqual(len(recs), 0)164    165    def test_null(self):166        """test None comparison queries"""167        msg_ids = self.load_records(10)168        query = {'msg_id' : None}169        recs = self.db.find_records(query)170        self.assertEqual(len(recs), 0)171        query = {'msg_id' : {'$ne' : None}}172        recs = self.db.find_records(query)173        self.assertTrue(len(recs) >= 10)174    175    def test_pop_safe_get(self):176        """editing query results shouldn't affect record [get]"""177        msg_id = self.db.get_history()[-1]178        rec = self.db.get_record(msg_id)179        rec.pop('buffers')180        rec['garbage'] = 'hello'181        rec['header']['msg_id'] = 'fubar'182        rec2 = self.db.get_record(msg_id)183        self.assertTrue('buffers' in rec2)184        self.assertFalse('garbage' in rec2)185        self.assertEqual(rec2['header']['msg_id'], msg_id)186    187    def test_pop_safe_find(self):188        """editing query results shouldn't affect record [find]"""189        msg_id = self.db.get_history()[-1]190        rec = self.db.find_records({'msg_id' : msg_id})[0]191        rec.pop('buffers')192        rec['garbage'] = 'hello'193        rec['header']['msg_id'] = 'fubar'194        rec2 = self.db.find_records({'msg_id' : msg_id})[0]195        self.assertTrue('buffers' in rec2)196        self.assertFalse('garbage' in rec2)197        self.assertEqual(rec2['header']['msg_id'], msg_id)198    def test_pop_safe_find_keys(self):199        """editing query results shouldn't affect record [find+keys]"""200        msg_id = self.db.get_history()[-1]201        rec = self.db.find_records({'msg_id' : msg_id}, keys=['buffers', 'header'])[0]202        rec.pop('buffers')203        rec['garbage'] = 'hello'204        rec['header']['msg_id'] = 'fubar'205        rec2 = self.db.find_records({'msg_id' : msg_id})[0]206        self.assertTrue('buffers' in rec2)207        self.assertFalse('garbage' in rec2)208        self.assertEqual(rec2['header']['msg_id'], msg_id)209class TestDictBackend(TaskDBTest, TestCase):210    211    def create_db(self):212        return DictDB()213    214    def test_cull_count(self):215        self.db = self.create_db() # skip the load-records init from setUp216        self.db.record_limit = 20217        self.db.cull_fraction = 0.2218        self.load_records(20)219        self.assertEqual(len(self.db.get_history()), 20)220        self.load_records(1)221        # 0.2 * 20 = 4, 21 - 4 = 17222        self.assertEqual(len(self.db.get_history()), 17)223        self.load_records(3)224        self.assertEqual(len(self.db.get_history()), 20)225        self.load_records(1)226        self.assertEqual(len(self.db.get_history()), 17)227        228        for i in range(100):229            self.load_records(1)230            self.assertTrue(len(self.db.get_history()) >= 17)231            self.assertTrue(len(self.db.get_history()) <= 20)232    def test_cull_size(self):233        self.db = self.create_db() # skip the load-records init from setUp234        self.db.size_limit = 1000235        self.db.cull_fraction = 0.2236        self.load_records(100, buffer_size=10)237        self.assertEqual(len(self.db.get_history()), 100)238        self.load_records(1, buffer_size=0)239        self.assertEqual(len(self.db.get_history()), 101)240        self.load_records(1, buffer_size=1)241        # 0.2 * 100 = 20, 101 - 20 = 81242        self.assertEqual(len(self.db.get_history()), 81)243    244    def test_cull_size_drop(self):245        """dropping records updates tracked buffer size"""246        self.db = self.create_db() # skip the load-records init from setUp247        self.db.size_limit = 1000248        self.db.cull_fraction = 0.2249        self.load_records(100, buffer_size=10)250        self.assertEqual(len(self.db.get_history()), 100)251        self.db.drop_record(self.db.get_history()[-1])252        self.assertEqual(len(self.db.get_history()), 99)253        self.load_records(1, buffer_size=5)254        self.assertEqual(len(self.db.get_history()), 100)255        self.load_records(1, buffer_size=5)256        self.assertEqual(len(self.db.get_history()), 101)257        self.load_records(1, buffer_size=1)258        self.assertEqual(len(self.db.get_history()), 81)259    def test_cull_size_update(self):260        """updating records updates tracked buffer size"""261        self.db = self.create_db() # skip the load-records init from setUp262        self.db.size_limit = 1000263        self.db.cull_fraction = 0.2264        self.load_records(100, buffer_size=10)265        self.assertEqual(len(self.db.get_history()), 100)266        msg_id = self.db.get_history()[-1]267        self.db.update_record(msg_id, dict(result_buffers = [os.urandom(10)], buffers=[]))268        self.assertEqual(len(self.db.get_history()), 100)269        self.db.update_record(msg_id, dict(result_buffers = [os.urandom(11)], buffers=[]))270        self.assertEqual(len(self.db.get_history()), 79)271class TestSQLiteBackend(TaskDBTest, TestCase):272    @dec.skip_without('sqlite3')273    def create_db(self):274        location, fname = os.path.split(temp_db)275        log = logging.getLogger('test')276        log.setLevel(logging.CRITICAL)277        return SQLiteDB(location=location, fname=fname, log=log)278    279    def tearDown(self):280        self.db._db.close()281def teardown():282    """cleanup task db file after all tests have run"""283    try:284        os.remove(temp_db)...

indexhistoryfunc.py

Source:indexhistoryfunc.py

1from nsepy import get_history2from datetime import date3def indexhistory(indexsymbol):4    nifty_index = get_history(symbol=indexsymbol, start=date(2009,3,31), end=date(2009,3,31), index=True)5    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2010,3,31), end=date(2010,3,31), index=True))6    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2011,3,31), end=date(2011,3,31), index=True))7    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2012,3,30), end=date(2012,3,30), index=True))8    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2013,3,28), end=date(2013,3,28), index=True))9    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2014,3,31), end=date(2014,3,31), index=True))10    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2015,3,31), end=date(2015,3,31), index=True))11    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2016,3,31), end=date(2016,3,31), index=True))12    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2017,3,31), end=date(2017,3,31), index=True))13    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2018,3,28), end=date(2018,3,28), index=True))14    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2019,3,29), end=date(2019,3,29), index=True))15    nifty_index = nifty_index.append(get_history(symbol=indexsymbol, start=date(2020,3,31), end=date(2020,3,31), index=True))16    print(nifty_index)17    return nifty_index...

nsehistory.py

Source:nsehistory.py

1from nsepy import get_history2from datetime import date3#data = get_history(symbol="SBIN", start=date(2015,1,1), end=date(2015,1,31))4nifty_pe = get_history(symbol="NIFTY", start=date(2009,3,31), end=date(2009,3,31), index=True)5print(nifty_pe)6nifty_pe = get_history(symbol="NIFTY", start=date(2010,3,31), end=date(2010,3,31), index=True)7print(nifty_pe)8nifty_pe = get_history(symbol="NIFTY", start=date(2011,3,31), end=date(2011,3,31), index=True)9print(nifty_pe)10nifty_pe = get_history(symbol="NIFTY", start=date(2012,3,30), end=date(2012,3,30), index=True)11print(nifty_pe)12nifty_pe = get_history(symbol="NIFTY", start=date(2013,3,28), end=date(2013,3,28), index=True)13print(nifty_pe)14nifty_pe = get_history(symbol="NIFTY", start=date(2014,3,31), end=date(2014,3,31), index=True)15print(nifty_pe)16nifty_pe = get_history(symbol="NIFTY", start=date(2015,3,31), end=date(2015,3,31), index=True)17print(nifty_pe)18nifty_pe = get_history(symbol="NIFTY", start=date(2016,3,31), end=date(2016,3,31), index=True)19print(nifty_pe)20nifty_pe = get_history(symbol="NIFTY", start=date(2017,3,31), end=date(2017,3,31), index=True)21print(nifty_pe)22nifty_pe = get_history(symbol="NIFTY", start=date(2018,3,28), end=date(2018,3,28), index=True)23print(nifty_pe)24nifty_pe = get_history(symbol="NIFTY", start=date(2019,3,29), end=date(2019,3,29), index=True)25print(nifty_pe)26nifty_pe = get_history(symbol="NIFTY", start=date(2020,3,31), end=date(2020,3,31), index=True)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.