KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use state_obj method in fMBT

Best Python code snippet using fMBT_python

old_settings.py

Source:old_settings.py

1#- * -coding: utf - 8 - * -2"""3@author: â Ryan McConnell ââ  â§4"""5from time import time6from .Core.Settings import EnhanceSettings7from .utilities import chain_iter8from .Core.ItemStore import ItemStoreItem, ItemStore9from .common import GEAR_DB_MANAGER, GearItemStore10MEMORY_FRAG_COST = 174000011P_CRON_STONE_COST = 200000012P_CLEANSE_COST = 10000013def convert_0002(state_obj):14    P_NUM_FS = 'num_fs'15    P_CRON_STONE_COST = 'cost_cron'16    P_CLEANSE_COST = 'cost_cleanse'17    P_FAIL_STACKERS = 'fail_stackers'18    P_COST_CONC_W = 'cost_conc_w'19    P_COST_BS_W = 'cost_bs_w'20    P_R_ENHANCE_ME = 'r_enhance_me'21    P_FS_EXCEPTIONS = 'fs_exceptions'22    P_COST_CONC_A = 'cost_conc_a'23    P_R_FAIL_STACKERS = 'r_fail_stackers'24    P_COST_BS_A = 'cost_bs_a'25    P_ENHANCE_ME= 'enhance_me'26    P_FS_COUNTS = 'fail_stackers_count'27    P_COST_MEME = 'cost_meme'28    item_shop = {'items':{29        ItemStore.P_MEMORY_FRAG: state_obj.pop(P_COST_MEME),30        ItemStore.P_CONC_WEAPON: state_obj.pop(P_COST_CONC_W),31        ItemStore.P_CONC_ARMOR: state_obj.pop(P_COST_CONC_A),32        ItemStore.P_BLACK_STONE_WEAPON: state_obj.pop(P_COST_BS_W),33        ItemStore.P_BLACK_STONE_ARMOR: state_obj.pop(P_COST_BS_A)34    }35    }36    state_obj[EnhanceSettings.P_ITEM_STORE] = item_shop37    return state_obj38def convert_0010(state_obj):39    P_NUM_FS = 'num_fs'40    P_CRON_STONE_COST = 'cost_cron'41    P_CLEANSE_COST = 'cost_cleanse'42    P_FAIL_STACKERS = 'fail_stackers'43    P_COST_CONC_W = 'cost_conc_w'44    P_COST_BS_W = 'cost_bs_w'45    P_R_ENHANCE_ME = 'r_enhance_me'46    P_FS_EXCEPTIONS = 'fs_exceptions'47    P_COST_CONC_A = 'cost_conc_a'48    P_R_FAIL_STACKERS = 'r_fail_stackers'49    P_COST_BS_A = 'cost_bs_a'50    P_ENHANCE_ME= 'enhance_me'51    P_FS_COUNTS = 'fail_stackers_count'52    P_COST_MEME = 'cost_meme'53    P_fail_stackers = state_obj[P_FAIL_STACKERS]54    P_r_fail_stackers = state_obj[P_R_FAIL_STACKERS]55    P_enhanceme = state_obj[P_ENHANCE_ME]56    P_r_enhanceme = state_obj[P_R_ENHANCE_ME]57    for gear_obj in chain_iter(P_fail_stackers, P_r_fail_stackers, P_enhanceme, P_r_enhanceme):58        cost = gear_obj.pop('cost')59        gear_obj['base_item_cost'] = cost60    return state_obj61def convert_0011(state_obj):62    item_store = state_obj['item_store']63    items = item_store['items']64    P_BLACK_STONE_ARMOR = items['BLACK_STONE_ARMOR']65    P_BLACK_STONE_WEAPON = items['BLACK_STONE_WEAPON']66    P_CONC_ARMOR = items['CONC_ARMOR']67    P_CONC_WEAPON = items['CONC_WEAPON']68    P_MEMORY_FRAG = items['MEMORY_FRAG']69    P_DRAGON_SCALE = items['DRAGON_SCALE']70    hour_from_now = time() + 360071    state_obj[EnhanceSettings.P_ITEM_STORE] = {'items':{72        ItemStore.P_BLACK_STONE_ARMOR: ItemStoreItem('BLACK_STONE_ARMOR', [P_BLACK_STONE_ARMOR], expires=hour_from_now).get_state_json(),73        ItemStore.P_BLACK_STONE_WEAPON: ItemStoreItem('BLACK_STONE_WEAPON', [P_BLACK_STONE_WEAPON], expires=hour_from_now).get_state_json(),74        ItemStore.P_CONC_ARMOR: ItemStoreItem('CONC_ARMOR', [P_CONC_ARMOR], expires=hour_from_now).get_state_json(),75        ItemStore.P_CONC_WEAPON: ItemStoreItem('CONC_WEAPON', [P_CONC_WEAPON], expires=hour_from_now).get_state_json(),76        ItemStore.P_MEMORY_FRAG: ItemStoreItem('MEMORY_FRAG', [P_MEMORY_FRAG], expires=hour_from_now).get_state_json(),77        ItemStore.P_DRAGON_SCALE: ItemStoreItem('DRAGON_SCALE', [P_DRAGON_SCALE], expires=hour_from_now).get_state_json()78    }79    }80    return state_obj81def convert_0012(state_obj):82    P_VALKS = 'valks'83    valks = state_obj[P_VALKS]84    new_valk = {}85    for v in valks:86        if v in new_valk:87            new_valk[v] += 188        else:89            new_valk[v] = 190    state_obj[P_VALKS] = new_valk91    return state_obj92def convert_0013(state_obj):93    P_VALKS = 'valks'94    item_store = state_obj['item_store']95    items = item_store['items']96    new_store = GearItemStore(gear_db=GEAR_DB_MANAGER).get_state_json()97    for key, v in new_store['items'].items():98        if key in items:99            new_store[key] = items[key]100    state_obj['item_store'] = new_store101    return state_obj102def convert_0014(state_obj):103    fail_stackers = state_obj.pop('fail_stackers')104    r_fail_stackers = state_obj.pop('r_fail_stackers')105    fail_stackers_2 = []106    state_obj['fail_stackers_2'] = fail_stackers_2107    r_fail_stackers_2 = []108    state_obj['r_fail_stackers_2'] = r_fail_stackers_2109    for_profit_gear = []110    state_obj['for_profit_gear'] = for_profit_gear111    r_for_profit_gear = []112    state_obj['r_for_profit_gear'] = r_for_profit_gear113    new_fail_stackers = []114    new_r_fail_stackers = []115    state_obj['fail_stackers'] = new_fail_stackers116    state_obj['r_fail_stackers'] = new_r_fail_stackers117    for gear_obj in fail_stackers:118        if gear_obj['enhance_lvl'] == '15':119            new_fail_stackers.append(gear_obj)120        elif gear_obj['procurement_cost'] == 0 and gear_obj['sale_balance'] == 0 and gear_obj['fail_sale_balance'] == 0:121            fail_stackers_2.append(gear_obj)122        else:123            for_profit_gear.append(gear_obj)124    for gear_obj in r_fail_stackers:125        if gear_obj['enhance_lvl'] == '15':126            new_r_fail_stackers.append(gear_obj)127        elif gear_obj['procurement_cost'] == 0 and gear_obj['sale_balance'] == 0 and gear_obj['fail_sale_balance'] == 0:128            r_fail_stackers_2.append(gear_obj)129        else:130            r_for_profit_gear.append(gear_obj)131    state_obj['fs_genome'] = [0, 23, 6, 6, 13]132    return state_obj133def convert_0015(state_obj):134    P_GENOME_FS = state_obj.pop('fs_genome')135    fsl_sec_gidx = P_GENOME_FS[0]136    genome = P_GENOME_FS[1:]137    state_obj['fs_genome'] = [{138        'genome': genome,139        'gear_dx': fsl_sec_gidx,140        'num_fs': state_obj['num_fs']141    }]142    return state_obj143def convert_0016(state_obj):144    P_GENOME_FS = state_obj['fs_genome']145    fss = state_obj['fail_stackers_2']146    def convert_fsl(fsl_state):147        gear_dx = fsl_state.pop('gear_dx')148        if gear_dx is None:149            this_id = None150        else:151            gs_obj = fss[gear_dx]152            if 'id' in gs_obj:153                this_id = gs_obj['id']154            else:155                this_id = id(gs_obj)156                gs_obj['id'] = this_id157        fsl_state['gear_id'] = this_id158    list(map(convert_fsl, P_GENOME_FS))159    if 'fsl_l' in state_obj:160        for toplvl in state_obj['fsl_l']:161            fslsl = toplvl[2]162            for fsl_spec in fslsl:163                if type(fsl_spec) == dict:164                    convert_fsl(fsl_spec)165    return state_obj166def convert_0017(state_obj):167    item_store = state_obj['item_store']168    item_store['custom_prices']= {}169    return state_obj170def convert_0018(state_obj):171    item_store = state_obj['item_store']172    item_store['custom_prices'] ={}173    item_store['custom_gear_prices'] = {}174    return state_obj175class ConversionError(Exception):176    pass177class ConversionManager(object):178    def __init__(self, state_obj):179        self.state_obj = state_obj180        self.converters = {181            '0.0.0.2': (convert_0002, '0.0.1.0'),182            '0.0.1.0': (convert_0010, '0.0.1.1'),183            '0.0.1.1': (convert_0011, '0.0.1.2'),184            '0.0.1.2': (convert_0012, '0.0.1.3'),185            '0.0.1.3': (convert_0013, '0.0.1.4'),186            '0.0.1.4': (convert_0014, '0.0.1.5'),187            '0.0.1.5': (convert_0015, '0.0.1.6'),188            '0.0.1.6': (convert_0016, '0.0.1.7'),189            '0.0.1.7': (convert_0017, '0.0.1.8'),190            '0.0.1.8': (convert_0018, '0.0.1.9')191        }192    def add_converter(self, target_ver, conversion_func, out_ver):193        self.converters[target_ver] = (conversion_func, out_ver)194    def convert(self, input_version, target_ver=None):195        state_obj = self.state_obj196        ver = input_version197        while (ver != target_ver) and (ver in self.converters):198            try:199                convert_func, ver = self.converters[ver]200            except KeyError:201                raise ConversionError202            convert_func(state_obj)...

state.py

Source:state.py

1#!/usr/bin/env python2"""State Repository implementation"""3__author__ = 'Michael Meisinger'4from pyon.core import bootstrap5from pyon.core.exception import NotFound, BadRequest, Conflict6from pyon.datastore.datastore import DataStore7from pyon.util.containers import get_ion_ts8from pyon.util.log import log9from interface.objects import ProcessState10class StateRepository(object):11    """12    Class that uses a data store to provide a persistent state repository for ION processes.13    """14    def __init__(self, datastore_manager=None, container=None):15        self.container = container or bootstrap.container_instance16        # Get an instance of datastore configured as directory.17        # May be persistent or mock, forced clean, with indexes18        datastore_manager = datastore_manager or self.container.datastore_manager19        self.state_store = datastore_manager.get_datastore("state", DataStore.DS_PROFILE.STATE)20    def start(self):21        pass22    def stop(self):23        self.close()24    def close(self):25        """26        Pass-through method to close the underlying datastore.27        """28        self.state_store.close()29    def put_state(self, key, state, state_obj=None):30        """31        Persist a private process state using the given key (typically a process id).32        The state vector is an object (e.g. a dict) that may contain any python type that33        is JSON-able. This means no custom objects are allowed in here.34        WARNING: If multiple threads/greenlets persist state concurrently, e.g. based35        on message processing and time, the calls to this method need to be protected36        by an exclusive lock (semaphore).37        @retval the ProcessState object as written38        """39        log.debug("Store persistent state for key=%s", key)40        if not isinstance(state, dict):41            raise BadRequest("state must be type dict, not %s" % type(state))42        if state_obj is not None:43            if not isinstance(state_obj, ProcessState):44                raise BadRequest("Argument state_obj is not ProcessState object")45            state_obj.state = state46            state_obj.ts = get_ion_ts()47            try:48                id, rev = self.state_store.update(state_obj)49                state_obj._rev = rev50                return state_obj51            except Conflict as ce:52                log.info("Process %s state update conflict - retry.")53        try:54            state_obj = self.state_store.read(key)55            state_obj.state = state56            state_obj.ts = get_ion_ts()57            id, rev = self.state_store.update(state_obj)58            state_obj._rev = rev59        except NotFound as nf:60            state_obj = ProcessState(state=state, ts=get_ion_ts())61            id, rev = self.state_store.create(state_obj, object_id=key)62            state_obj._id = id63            state_obj._rev = rev64        return state_obj65    def get_state(self, key):66        """67        Returns the state vector for given key (typically a process id).68        The state vector is a previously persisted object (e.g. a dict).69        In case no state was found, NotFound is raised.70        @retval a tuple with state vector and ProcessState object71        """72        log.debug("Retrieving persistent state for key=%s", key)73        state_obj = self.state_store.read(key)74        return state_obj.state, state_obj75class StatefulProcessMixin(object):76    """77    Mixin class for stateful processes.78    Need to avoid __init__79    """80    def _set_state(self, key, value):81        """82        Sets a key-value in the process's state vector. Marks the state as changed83        if the value has actually changed.84        """85        if not hasattr(self, "_proc_state"):86            self._proc_state = {}87            self._proc_state_changed = False88        old_state = self._proc_state.get(key, None)89        if old_state != value:90            self._proc_state[key] = value91            self._proc_state_changed = True92            log.debug("Process state updated. pid=%s, key=%s, value=%s", self.id, key, value)93    def _get_state(self, key, default=None):94        """95        Returns the value for a key from the process's state vector. If there is no96        value or the value is None, the default is returned.97        """98        if not hasattr(self, "_proc_state"):99            return None100        state = self._proc_state.get(key, None)101        return state if state is not None else default102    def _get_state_vector(self):103        """104        Returns the entire process state vector as a dict.105        Note: direct changes to the state vector will not automatically be detected.106        """107        if not hasattr(self, "_proc_state"):108            self._proc_state = {}109        return self._proc_state110    def _mark_changed(self):111        """112        Marks the process state vector as changed. The container will flush the113        state change to the repository when appropriate.114        """115        if not hasattr(self, "_proc_state"):116            self._proc_state = {}117        self._proc_state_changed = True118    def _flush_state(self):119        """120        Immediately pushes the state to the state repository. This call blocks121        until the write has completed122        """123        pass124    def _load_state(self):125        """126        Loads the process's state vector again from the state repository. Calling this127        operation should not be necessary in normal circumstances.128        """...

report_controller.py

Source:report_controller.py

1# -*- coding: utf-8 -*-2# from odoo.addons.web.http import Controller, route, request3from odoo.addons.report.controllers.main import ReportController4from odoo import http5import simplejson6import logging7_logger = logging.getLogger(__name__)8class FalReportController(ReportController):9    @http.route(['/report/download'], type='http', auth="user")10    def report_download(self, data, token):11        # picking_obj = http.request.env['stock.picking']12        # mo_obj = http.request.env['mrp.production']13        # payslip_obj = http.request.env['hr.payslip']14        # expense_obj = http.request.env['hr.expense.expense']15        requestcontent = simplejson.loads(data)16        url, type = requestcontent[0], requestcontent[1]17        # url = u'/report/pdf/sale.report_saleorder/37'18        # type = u'qweb-pdf'19        response = ReportController().report_download(data, token)20        if type == 'qweb-pdf':21            reportname = url.split('/report/pdf/')[1].split('?')[0]22            # reportname = u'sale.report_saleorder/37'23            try:24                reportname, _docids = reportname.split('/')25                assert _docids26                # reportname = u'sale.report_saleorder'27                # docids = 3728            except ValueError:29                reportname = reportname.split('.')[1]30            filename = reportname31            if reportname in [32                'ssl_sale_ext.report_saleorder_document_ssl',33                'ssl_purchase_ext.report_purchases_document_ssl',34                'ssl_account_ext.report_invoices_document_ssl',35                'ssl_sale_ext.report_saleorder_document_pipeline',36            ]:37                for docids in _docids.split(','):38                    # Sales Order39                    s = 'ssl_sale_ext.report_saleorder_document_ssl'40                    if reportname == s:41                        sale_obj = http.request.env['sale.order']42                        object = sale_obj.browse(int(docids))43                        state_obj = object.state.encode('utf-8', 'ignore')44                        if state_obj in ['draft', 'sent', 'cancel']:45                            state_obj = 'Quotation'46                        elif state_obj in ['sale', 'done']:47                            state_obj = 'Sale Order'48                        filename = object.name.encode(49                            'utf-8', 'ignore') + '-' + state_obj50                    # Purchase Order51                    p = 'ssl_purchase_ext.report_purchases_document_ssl'52                    if reportname == p:53                        purchase_obj = http.request.env['purchase.order']54                        object = purchase_obj.browse(int(docids))55                        state_obj = object.state.encode('utf-8', 'ignore')56                        if state_obj in ['draft', 'sent', 'cancel']:57                            state_obj = 'Request for Quotation'58                        elif state_obj in ['purchase', 'done']:59                            state_obj = 'Purchase Order'60                        elif state_obj in ['to approve']:61                            state_obj = 'To Approve'62                        filename = object.name.encode(63                            'utf-8', 'ignore') + '-' + state_obj64                    # Account Invoice65                    i = 'ssl_account_ext.report_invoices_document_ssl'66                    if reportname == i:67                        invoice_obj = http.request.env['account.invoice']68                        object = invoice_obj.browse(int(docids))69                        state_obj = object.state.encode('utf-8', 'ignore')70                        type_obj = object.type.encode('utf-8', 'ignore')71                        if state_obj in [72                            'draft',73                            'proforma',74                            'proforma2',75                            'cancelled'76                        ] and type_obj == 'out_invoice':77                            state_obj = 'Proforma Invoice'78                        elif type_obj == 'in_invoice':79                            state_obj = 'Vendor Bill'80                        elif state_obj in ['open', 'paid'] and \81                                type_obj == 'out_invoice':82                            state_obj = 'Invoice'83                        if object.number:84                            filename = object.number.encode(85                                'utf-8', 'ignore') + '-' + state_obj86                        else:87                            filename = object.fal_draft_number.encode(88                                'utf-8', 'ignore') + '-' + state_obj89                    # Wishlist90                    w = 'ssl_sale_ext.report_saleorder_document_pipeline'91                    if reportname == w:92                        crm_obj = http.request.env['crm.lead']93                        object = crm_obj.browse(int(docids))94                        filename = object.fal_wishlist_number.encode(95                            'utf-8', 'ignore') + '-' + 'Wishlist'96                        # filename = SO001297            filename = filename.replace(",", " ")98            response.headers.set(99                'Content-Disposition',100                'attachment; filename=%s.pdf' % filename)...

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.