Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use shared_context method in Mamba

Best Python code snippet using mamba

atm.py

Source:atm.py

1import copy2import logging3from typing import TYPE_CHECKING45from errors import ErrorCode6from infra.bank_api import MockBankSystem17from model.command import MockUpdateTransactionCommand89if TYPE_CHECKING:10    from model.domain import Card, Account, CashBox11    from infra.bank_api import IBankSystem12    from typing import Callable, NoReturn13    from model.command import IUpdateTransactionCommand141516class Atm:17    def __init__(self, cash_box, bank_system=None, update_transaction=None):18        """19        Args:20            cash_box (CashBox): CashBox containing cash, not a physical one21                it must be not null, but set as optional for easier testing2223            bank_system (IBankSystem): implementation of Bank System or Mock24            update_transaction (IUpdateTransactionCommand): implementation of update transaction25        """26        self.__context = AtmContext()  # type: AtmContext27        self.__context.cash_box = cash_box28        self.__context.bank_system = bank_system() if bank_system else MockBankSystem1()29        self.__context.update_transaction_command \30            = update_transaction() if update_transaction else MockUpdateTransactionCommand()3132    """ATM ACTIONS"""33    def insert_card(self, card):34        """Insert card using `AtmWait`3536        Args:37            card (Card): Current card38        """39        self.__context.current.insert_card(card)4041    def enter_pin(self, pin):42        """Enter pin using `AtmReady`4344        Args:45            pin (str): Personal identification number46        """47        self.__context.current.enter_pin(pin)4849    def display_account_list(self):50        """Retrieve copy of accounts connected to card"""51        return self.__context.current.get_accounts()5253    def back(self):54        self.__context.current.back()5556    def select_account(self, idx):57        """Select account5859        Args:60            idx (int): Index of accounts in shared_context, start with 061        """62        self.__context.current.select_account(idx)6364    def select_deposit(self):65        """Select deposit menu"""66        self.__context.current.select_deposit()6768    def select_withdraw(self):69        """Select withdraw menu"""70        logger = logging.getLogger()71        self.__context.current.select_withdraw()7273    def put_in_cash(self, amount):74        """Put amount into selected account7576        Args:77            amount: Amount to be deposited78        """79        self.__context.current.put_cash(amount)8081    def enter_withdrawal_amount(self, amount):82        """Enter the amount to withdraw from the selected account8384        Args:85            amount (int): Amount to be withdrawn86        """87        self.__context.current.enter_withdrawal_amount(amount)8889    def take_out_cash(self, amount):90        """Withdraw the amount from selected account after vault is opened9192        Args:93            amount (int): Amount of money to withdraw94        """95        self.__context.current.take_cash(amount)9697    def select_balance(self):98        """Select balance"""99        self.__context.current.select_balance()100101    def exit(self):102        """Exit system"""103        self.__context.current.exit()104105    def take_out_card(self):106        """Take out card in exit state"""107        self.__context.current.remove_card()108109    """FOR UI IMPLEMENTATION"""110    def get_selected_account(self):111        """Get selected account object112113        * This method designed to support ui114        """115        return copy.deepcopy(self.__context.selected_account)116117    def get_inserted_card(self):118        """Get card object119120        * This method designed to support ui121        """122        card = copy.deepcopy(self.__context.card)123        return card124125    def get_user(self):126        """Get user object127128        * This method designed to support ui129        """130        user = copy.deepcopy(self.__context.card.card_holder)131        return user132133    def get_current_state_name(self):134        """Get current state135136        * This method designed to support ui137        """138        return self.__context.current.get_name()139140    def register_on_load(self, on_load_func):141        """Register on load function to be called after changing state142143        * This method designed to support ui144145        Args:146            on_load_func (function): Function to be called after changing state147        """148        self.__context.register_on_load(on_load_func)149150    def register_on_error(self, on_error_func):151        """Register on error function to be called after error152153        * The method design to support UI154155        Args:156            on_error_func (function): Function to be called after changing state157                `on_error_func` should have one parameter (e.g. Callable[[Exception], NoReturn]) to get an error message158                return type does not matter159        """160        self.__context.register_on_error(on_error_func)161162class AtmContext:163    def __init__(self):164        self.states = {165            AtmWait.get_name(): AtmWait(self),166            AtmReady.get_name(): AtmReady(self),167            AtmAuthorized.get_name(): AtmAuthorized(self),168            AtmAccountSelected.get_name(): AtmAccountSelected(self),169            AtmProcessingDeposit.get_name(): AtmProcessingDeposit(self),170            AtmPreProcessingWithdrawal.get_name(): AtmPreProcessingWithdrawal(self),171            AtmProcessingWithdrawal.get_name(): AtmProcessingWithdrawal(self),172            AtmDisplayingBalance.get_name(): AtmDisplayingBalance(self),173            AtmExit.get_name(): AtmExit(self),174        }175        # Initialize first time only176        self.cash_box = None # type: CashBox177        self.bank_system = None  # type: IBankSystem178        self.update_transaction_command = None # type: IUpdateTransactionCommand179        self.on_load_func = None # type: Callable[..., NoReturn]180        self.on_error_func = None # type: Callable[[Exception], NoReturn]181182        # Temporal variables which can be reset on user's leave183        self.current = self.states[AtmWait.get_name()]  # type: AtmState184        self.card = None  # type: Card185        self.accounts = []186        self.selected_account = None  # type: Account187        self.amount_to_be_withdrawn = 0  # type: int188    def clean_context(self):189        """Clean context190191        * It change the current state to AtmWait192        """193        self.current = self.states[AtmWait.get_name()]  # type: AtmState194        self.card = None  # type: Card195        self.accounts = []196        self.selected_account = None  # type: Account197        self.amount_to_be_withdrawn = 0  # type: int198199200    def set_state(self, state_name):201        """Set current state by state name202203        * All states needs to call this method to change itself to another204205        Args:206            state_name (str): Name of next state207        """208        self.current = self.states[state_name]209        self.current.on_load()210        balance = self.selected_account.balance if self.selected_account else 'Not Selected'211        print('[%s card=%s, balance=%s]' % (state_name, self.card, balance))212        if self.on_load_func:213            self.on_load_func()214215    def register_on_load(self, on_load_func):216        """Register function which is to be called after change state217218        Args:219            on_load_func (function): Function to be called after change state220        """221        self.on_load_func = on_load_func222223    def register_on_error(self, on_error_func):224        """Register on error function to be called after error225226        * The method design to support UI227228        Args:229            on_error_func (function): Function to be called after changing state230                `on_error_func` should have one parameter (e.g. Callable[[Exception], NoReturn]) to get an error231                return type does not matter232        """233        self.on_error_func = on_error_func234235class AtmState:236    """The default state237238    - For all method, it print message `Action is not available in the current state`239    """240241    def __init__(self, context):242        """243        Args:244            context (AtmContext): Shared context245        """246        self.shared_context = context247248    @classmethod249    def get_name(cls):250        """Return class state_name251252        - Class state_name is used for picking next state in `AtmContext`253        """254        return cls.__name__255256    def on_error(self, e):257        """call on_errr_func in context258259        Args:260            e (str): error message261        """262        if self.shared_context.on_error_func:263            self.shared_context.on_error_func(e)264265    def on_load(self):266        pass267268    def insert_card(self, card):269        """Insert card in `AtmWait`270271        Args:272            card (Card): Current card273        """274        print('Action is not available in the current state [%s]' % self.get_name())275276    def enter_pin(self, pin):277        """Enter pin number in `AtmReady`278279        Args:280            pin (str): Personal identification number281282        Raises:283            ValueError: Raised if incorrect pin is entered.284285                When raised, then it does not change to `AtmExit`286        """287        print('Action is not available in the current state [%s]' % self.get_name())288289    def get_accounts(self):290        """Get account list which is connected to card in `AtmAuthorized`291292        Returns:293            list[Account]: List of account294        """295        print('Action is not available in the current state [%s]' % self.get_name())296297    def back(self):298        print('Action is not available in the current state [%s]' % self.get_name())299300    def select_account(self, idx):301        """Select account to be used in `AtmAuthorized`302303        - When is success, then it changes to `AtmAccountSelected`304305        Args:306            idx (int): Index of accounts in shared_context, start with 0307308        Raises:309            IndexError: Raised if Idx is not in range of account list - When raised,310                then it changes to `AtmExit`311        """312        print('Action is not available in the current state [%s]' % self.get_name())313314    def select_deposit(self):315        """Select deposit menu316317        * It changes to `AtmProcessingDeposit`318        """319        print('Action is not available in the current state [%s]' % self.get_name())320321    def select_withdraw(self):322        """Select withdraw menu323324        * It changes to `AtmProcessingWithdraw`325        """326        print('Action is not available in the current state [%s]' % self.get_name())327328    def put_cash(self, amount):329        """Deposit the amount into selected account in `AtmAccountSelected`330331        Args:332            amount (int): Amount of money to deposit333        """334        print('Action is not available in the current state [%s]' % self.get_name())335336    def enter_withdrawal_amount(self, amount):337        """Enter the amount to withdraw from the selected account338339        Args:340            amount (int): Amount of money to withdraw341        """342        print('Action is not available in the current state [%s]' % self.get_name())343344    def take_cash(self, amount):345        """Withdraw the amount from selected account after vault is opened346347        Args:348            amount (int): Amount of money to withdraw349        """350        print('Action is not available in the current state [%s]' % self.get_name())351352    def select_balance(self):353        """Select display balance menu in `AtmAccountSelected`354355        * It changes to `AtmDisplayingBalance`356        """357        print('Action is not available in the current state [%s]' % self.get_name())358359    def exit(self):360        """Select display balance menu in multiple state361362        * It changes to `AtmExit`363        """364        print('Action is not available in the current state [%s]' % self.get_name())365366    def remove_card(self):367        """Remove card and remove all context variables368369        * It changes to ` AtmWait'370        """371        print('Action is not available in the current state [%s]' % self.get_name())372373374class AtmWait(AtmState):375    """The state waiting for a card (waiting for customers)376377    - Have nothing,378379    - When a card is inserted th,en it changes to the `AtmReady`380    """381382    def insert_card(self, card):383        """Insert card in `AtmWait`384385        - If successful, it changes to `AtmReady`.386387        Args:388            card (Card): Current card389        """390        print('insert card %s' % card)391        self.shared_context.card = card392        self.shared_context.set_state(AtmReady.get_name())393394395class AtmReady(AtmState):396    """The state waiting for pin397398    - Have card399400    - When a pin is entered, then it changes to the `AtmAuthorized`401402    - When a back is selected, then it changes to `AtmExit`403    """404405    def enter_pin(self, pin):406        """Enter pin number in `AtmReady`407408        Args:409            pin (str): Personal identification number410411        Raises:412            ValueError: incorrect pin is entered - When raised, it changes to `AtmExit`.413        """414        print('enter pin %s' % pin)415        # TODO: verify number from server416        try:417            if self.shared_context.bank_system.validate_pin(418                    self.shared_context.card.card_number,419                    pin420            ):421                self.shared_context.set_state(AtmAuthorized.get_name())422                return True423            else:424                raise ValueError(ErrorCode.PIN_IS_NOT_MATCHED)425        except ValueError as e:426            self.on_error(e)427            self.shared_context.set_state(AtmExit.get_name())428            logging.getLogger().warning(e)429430    def exit(self):431        self.shared_context.set_state(AtmExit.get_name())432433434class AtmAuthorized(AtmState):435    """The state waiting for selecting account436437    - Have card and pin438439    - When an account is selected, then it changes to `AtmAccountSelected`440441    - When a back-menu is selected, then it changes to `AtmReady`442    """443444    def on_load(self):445        self.get_accounts()446447    def get_accounts(self):448        """Get account list which is connected to card in `AtmAuthorized`449450        Returns:451            list[Account]: List of account452453        Raises:454            ReferenceError: Raised if cannot find accounts - When raised, it changes to `AtmExit`.455        """456        try:457            self.shared_context.accounts \458                = self.shared_context.bank_system.get_accounts(self.shared_context.card)459            if len(self.shared_context.accounts) < 1:460                raise RuntimeError(ErrorCode.CANNOT_FIND_ACCOUNT)461            print('get accounts result=%s' % self.shared_context.accounts)462        except RuntimeError as e:463            print(e)464            self.on_error(e)465            self.shared_context.set_state(AtmExit.get_name())466        return copy.deepcopy(self.shared_context.accounts)467468    def select_account(self, idx):469        """Select account to be used in `AtmAuthorized`470471        - When is success, then It changes to `AtmAccountSelected`472473        Args:474            idx (int): Index of accounts in shared_context, start with 0475476        Raises:477            IndexError: Raised if idx is not in range of account list - When raised, it does not anything478        """479        try:480            self.shared_context.selected_account \481                = self.shared_context.accounts[idx]482            self.shared_context.set_state(AtmAccountSelected.get_name())483        except IndexError as e:484            print(e)485            self.on_error(e)486            print('index starts from 0, candidates=%s' % self.shared_context.accounts)487488    def exit(self):489        self.shared_context.set_state(AtmExit.get_name())490491492class AtmAccountSelected(AtmState):493    """The state waiting for selecting transaction494495    - Have card, and selected account496497    - When a transaction is selected, then It changes to `AtmProcessing~`498499    - When a get-menu is selected, then it gives menu500501    - When a get-balance is selected, then it gives balance of selected account502    """503504    def select_deposit(self):505        """Select deposit menu506507        * It changes to `AtmProcessingDeposit`508        """509        self.shared_context.set_state(AtmProcessingDeposit.get_name())510511    def select_withdraw(self):512        """Select withdraw menu513514        * It changes to `AtmProcessingWithdraw`515        """516        self.shared_context.set_state(AtmPreProcessingWithdrawal.get_name())517518    def exit(self):519        self.shared_context.set_state(AtmExit.get_name())520521    def select_balance(self):522        """Select display balance menu in `AtmAccountSelected`523524        * It changes to `AtmDisplayingBalance`525        """526        self.shared_context.set_state(AtmDisplayingBalance.get_name())527528529    def back(self):530        """ back to `AtmAuthorized`531        """532        self.shared_context.selected_account = None533        self.shared_context.set_state(AtmAuthorized.get_name())534535class AtmProcessingDeposit(AtmState):536    """The state processing deposit transaction537538    - Have card, and selected account539540    - When customer put money, then it changes to `AtmAccountSelected`541542    - When customer put less/more money, then it spit out and is changes to543      `AtmExit` while throwing error544    """545546    def put_cash(self, amount):547        """Put amount into selected account in `AtmProcessingDeposit`548549        Args:550            amount: Amount the customer wants to deposit551        """552        print('put cash %s' % amount)553        try:554            if amount < 0:555                raise ValueError(ErrorCode.AMOUNT_MUST_BE_POSITIVE)556            # transaction557            self.shared_context.update_transaction_command.execute(558                self.shared_context.bank_system,559                self.shared_context.cash_box,560                self.shared_context.selected_account,561                + amount562            )563            self.shared_context.set_state(AtmDisplayingBalance.get_name())564        except ValueError as e:565            print(e)566            self.on_error(e)567            # in this case, assume customer withdraw the left money in the vault568            self.shared_context.set_state(AtmExit.get_name())569570    def exit(self):571        self.shared_context.set_state(AtmExit.get_name())572573    def back(self):574        """ back to `AtmAccountSelected`575576        * Customer canceled deposit577        """578        self.shared_context.set_state(AtmAccountSelected.get_name())579580class AtmPreProcessingWithdrawal(AtmState):581    """The state processing withdrawal transaction582583    - Have card, and selected account584585    - When customer enter amount to withdraw, then it check the balance586587    - When account have enough balance, then it changes to588      `AtmProcessingWithdrawal`589    """590591    def enter_withdrawal_amount(self, amount):592        """Enter amount customer want to withdraw593594        - Check current machine's cash box595596        - Check customer's account balance597598        Args:599            amount: Amount the customer want to withdraw600        """601        print('enter withdrawal amount %s' % amount)602        try:603            if amount < 0:604                raise ValueError(ErrorCode.AMOUNT_MUST_BE_POSITIVE)605            if self.shared_context.cash_box.cash < amount:606                raise ValueError(ErrorCode.CASH_BOX_DOES_NOT_HAVE_ENOUGH_CASH)607            if self.shared_context.selected_account.balance < amount:608                raise ValueError(ErrorCode.ACCOUNT_DOES_NOT_HAVE_ENOUGH_CASH)609            self.shared_context.amount_to_be_withdrawn = amount610            self.shared_context.set_state(AtmProcessingWithdrawal.get_name())611        except ValueError as e:612            print(str(e))613            self.on_error(e)614            self.shared_context.set_state(AtmExit.get_name())615616    def exit(self):617        self.shared_context.set_state(AtmExit.get_name())618619    def back(self):620        """ back to `AtmAccountSelected`621622        * Customer canceled withdrawal623        """624        self.shared_context.set_state(AtmAccountSelected.get_name())625626class AtmProcessingWithdrawal(AtmState):627    """The state processing withdrawal transaction628629    - Have card, selected account and amount_to_be_withdrawn630631    - When customer take money, then it changes to `AtmAccountSelected`632633    - When customer try to take more money than s/he got, then it changes to634      `AtmExit` while throwing error635    """636637    def take_cash(self, amount):638        """Withdraw the amount from selected account after vault is opened639640        * Customer left with out taking cash means customer take 0 cash641        Args:642            amount (int): Amount to withdraw643        """644        print('[take cash %s]' % amount)645        try:646            if amount < 0:647                raise RuntimeError(ErrorCode.AMOUNT_MUST_BE_POSITIVE)648            if amount > self.shared_context.amount_to_be_withdrawn:649                raise RuntimeError(ErrorCode.AMOUNT_MUST_BE_LOWER_THAN_AMOUNT_TO_BE_WITHDRAWN)650            # transaction651            self.shared_context.update_transaction_command.execute(652                self.shared_context.bank_system,653                self.shared_context.cash_box,654                self.shared_context.selected_account,655                - amount656            )657            self.shared_context.set_state(AtmDisplayingBalance.get_name())658        except ValueError as e:659            print(e)660            self.on_error(e)661            self.shared_context.set_state(AtmDisplayingBalance.get_name())662663    def exit(self):664        self.shared_context.selected_account.balance += self.shared_context.amount_to_be_withdrawn665        self.shared_context.set_state(AtmExit.get_name())666667    def back(self):668        """ back to `AtmPreProcessingWithdrawal`669670        * May be Customer wants to change amount to withdrawn671        """672        self.shared_context.amount_to_be_withdrawn = 0673        self.shared_context.set_state(AtmPreProcessingWithdrawal.get_name())674675class AtmDisplayingBalance(AtmState):676    """The state displaying balance677678    - Have card, and selected account679680    - When each transaction finished, then those states change to this681      state682683    - Cannot go back to former state684    """685686    def on_load(self):687        account = self.shared_context.selected_account688        print('[on load\naccount_number: %s,\naccount_holder: %s,\naccount_balance:%s]' % (689            account.account_number, account.name, account.balance))690691    def back(self):692        """Go back to accounts"""693        self.shared_context.set_state(AtmAuthorized.get_name())694695    def exit(self):696        self.shared_context.set_state(AtmExit.get_name())697698699class AtmExit(AtmState):700    """The state pull out card701702    - Have card to give back703704    - When customer take card, then it changes to `AtmWait`705    """706707    def back(self):708        """print message"""709        print('take card, then atm changes to initial state')710711    def remove_card(self):
...

experiment.py

Source:experiment.py

...215            self.context.executable = new216    #---------------------------------------------------------------------217    # Property getters/setters218    #---------------------------------------------------------------------219    def _get_shared_context(self):220        return self._shared_context221    def _set_shared_context(self, newcontext):222        subcontexts = self.context.subcontext.subcontexts223        if self._shared_context is not None:224            assert(self._shared_context in subcontexts)225            if newcontext is None:226                subcontexts.remove(self._shared_context)227                self._shared_context = None228            else:229                # Swap out the new context for the old, in-place230                ndx = subcontexts.index(self._shared_context)231                subcontexts[ndx] = newcontext232                self._shared_context = newcontext233        elif newcontext is not None:234            self._shared_context = newcontext235            subcontexts.append(newcontext)...

spell_checker.py

Source:spell_checker.py

1import json2import re3from collections import namedtuple4from functools import partial5from itertools import chain6from token import COMMENT, NAME, NEWLINE, STRING7from wordsegment import WORDS8from .websocket import register_handler9from .word_completer import is_prefix, wait_until_initialized10SpellingError = namedtuple('SpellingError',11                           ('line', 'ch', 'token', 'highlighted_token'))12Token = namedtuple('Token', ('start', 'string', 'type'))13DummyToken = namedtuple('DummyToken', ('string', ))14dummy = DummyToken('')15KEYWORDS = {'class', 'as'}16AFTER_IMPORT = {'as', 'from'}17STRING_AND_COMMENT = {STRING, COMMENT}18DELIMITER_REGEX = re.compile(r'[_\d]+')19CAMEL_REGEX = re.compile(r'([A-Z][a-z]*)')20NON_WORD = re.compile(r'[\W\d_]+')21MIN_WORD_LENGTH_TO_CHECK = 322handles = partial(register_handler, '')23@handles('AddToProjectDictionary')24async def add_to_project_dictionary(msg, send, context):25    shared_context = context.shared26    shared_context.spell_checker.project_dictionary.update(msg)27    with open(shared_context.project_dictionary_file, 'w') as f:28        json.dump(29            list(shared_context.spell_checker.project_dictionary),30            f,31            indent=4,32            sort_keys=True)33def decompose_token(token):34    """35    Split token into words.36    'something' -> ['something']37    'some_variable' -> ['some', 'variable']38    'SomeClass' -> ['some', 'class']39    'SOME_CONSTANT' -> ['some', 'constant']40    :param token: A token41    :return: A list of words in lower case42    """43    parts = DELIMITER_REGEX.split(token)44    if not ''.join(parts).isupper():45        parts = chain.from_iterable(CAMEL_REGEX.split(s) for s in parts)46    return [s.lower() for s in parts if len(s) >= MIN_WORD_LENGTH_TO_CHECK]47def highlight_spelling_errors(token, words, is_correct):48    """49    Wrap bad words in <em> tags.50    :param token: e.g. SomethingWorng51    :param words: e.g. ['something', 'worng']52    :param is_correct: e.g. [True, False]53    :return: e.g. 'Something<em>Worng</em>'54    """55    result = token56    lowercase_result = result.lower()57    index = 058    for w, i in zip(words, is_correct):59        index = lowercase_result.find(w, index)60        if not i:61            result = ''.join(62                (result[:index], '<em>', result[index:index + len(w)], '</em>',63                 result[index + len(w):]))64            lowercase_result = result.lower()65            index += 9  # length of <em></em>66    return result67class SpellChecker:68    def __init__(self, context):69        shared_context = context.shared70        shared_context.spell_checker = self71        self.project_dictionary = set()72        # load project dictionary73        if shared_context.project_dictionary_file.exists():74            with open(shared_context.project_dictionary_file, 'r') as f:75                project_dictionary = json.load(f)76            self.project_dictionary.update(project_dictionary)77    async def check_spelling(self, tokens):78        await wait_until_initialized()79        checked = set('')  # both checked tokens and words80        imported_names = set()81        spelling_errors = []82        def import_name(token):83            lower = token.lower()84            if lower in WORDS:85                return86            imported_names.add(lower)87            words = decompose_token(token)88            imported_names.update(words)89        def check_word(word, token):90            if word in checked or len(word) < 4:91                return92            if not word.islower():93                return94            if word in WORDS:95                return96            if word in self.project_dictionary:97                return98            if word in imported_names:99                return100            check_token(101                Token(102                    (token.start[0], token.start[1] + token.string.find(word)),103                    word, NAME))104        def check_token(token):105            if token.type in STRING_AND_COMMENT:106                for w in NON_WORD.split(token.string):107                    check_word(w, token)108                return109            if token.type != NAME:110                return111            s = token.string112            if len(s) <= MIN_WORD_LENGTH_TO_CHECK:113                return114            if s in checked:115                return116            words = decompose_token(token.string)117            is_correct = [(i in WORDS) for i in words]118            for i, (word, correct) in enumerate(zip(words, is_correct)):119                if correct:120                    continue121                if word in checked or is_prefix(word):122                    is_correct[i] = True123                if word in self.project_dictionary:124                    is_correct[i] = True125                if word in imported_names:126                    is_correct[i] = True127            checked.add(s)128            checked.update(words)129            if not all(is_correct):130                highlighted_token = highlight_spelling_errors(131                    token.string, words, is_correct)132                spelling_errors.append(133                    SpellingError(*token.start, token.string,134                                  highlighted_token))135        for_ = False136        def_ = False137        import_ = False138        t1, t0 = dummy, dummy139        for token in tokens:140            t1, t0 = t0, token141            t1s, t0s = t1.string, t0.string142            if t1s == 'for':143                for_ = True144            elif t0s == 'in':145                for_ = False146            if t1s == 'def':147                def_ = True148            elif t0s == ':' and t1s == ')':149                def_ = False150            if t1s == 'import':151                import_ = True152            elif t0.type == NEWLINE or t0s in AFTER_IMPORT:153                import_ = False154            # class xxx:155            if t1s in KEYWORDS:156                check_token(t0)157            # def xxx(yyy):158            elif def_:159                check_token(t0)160            # xxx = 123161            elif t0s == '=':162                check_token(t1)163            # for xxx, yyy in something:164            elif for_:165                check_token(t0)166            elif t0.type in STRING_AND_COMMENT:167                check_token(t0)168            elif import_:169                import_name(t0s)...

auth_server.py

Source:auth_server.py

1#!/usr/bin/env python32import json3from http.server import BaseHTTPRequestHandler, HTTPServer4from pathlib import Path5from urllib.request import HTTPError, Request, urlopen6from urllib.parse import urlencode, urlparse, parse_qs7from config import TOKEN_FILE as RELATIVE_TOKEN_FILE8from config import CLIENT_ID as SPOTIFY_CLIENT_ID, CLIENT_SECRET as SPOTIFY_CLIENT_SECRET9from config import REFRESHABLE_AUTH10SERVER_PORT = 819311REDIRECT_URI = f'http://localhost:{SERVER_PORT}/auth'12TOKEN_FILE = Path(__file__).parent / RELATIVE_TOKEN_FILE13def create_request_handler():14    shared_context = {15        'access_token': None,16        'code': None,17        'error': None18    }19    class AuthRequestHandler(BaseHTTPRequestHandler):20        _redirect_tpl = """21            <html>22            <script>23            if (location.href.indexOf('#') != -1)24                location.href = location.href.replace("#","?");25            </script>26            <h1>Redirecting...</h1>27            </html>28        """29        _success_tpl = """30            <html>31            <h1>Authorization successful</h1>32            <p>You can close this page now</p>33            </html>34        """35        _error_tpl = """36            <html>37            <h1>Authorization error</h1>38            <p>{error}</p>39            </html>40        """41        def do_GET(self):42            qs = urlparse(self.path).query43            qs_dict = parse_qs(qs)44            self.send_response(200)45            self.send_header("Content-Type", "text/html")46            self.end_headers()47            if not qs_dict:48                html = self._redirect_tpl49            elif 'access_token' in qs_dict:50                token = qs_dict['access_token'][0]51                shared_context['access_token'] = token52                html = self._success_tpl53            elif 'code' in qs_dict:54                token = qs_dict['code'][0]55                shared_context['code'] = token56                html = self._success_tpl57            else:58                error = qs_dict.get('error', ['unknown'])[0]59                shared_context['error'] = error60                html = self._error_tpl.format(error=error)61            self.wfile.write(html.encode('utf8'))62            self.wfile.flush()63    return AuthRequestHandler, shared_context64class AuthorizationError(Exception):65    pass66def listen_for_token(port):67    request_handler, shared_context = create_request_handler()68    httpd = HTTPServer(('localhost', port), request_handler)69    while True:70        httpd.handle_request()71        if shared_context['access_token']:72            return shared_context['access_token']73        if shared_context['code']:74            return shared_context['code']75        elif shared_context['error']:76            raise AuthorizationError(shared_context['error'])77def prompt_user_for_auth():78    if REFRESHABLE_AUTH:79        response_type = 'code'80    else:81        response_type = 'token'82    url_params = urlencode({83        'redirect_uri': REDIRECT_URI,84        'response_type': response_type,85        'client_id': SPOTIFY_CLIENT_ID,86        'scope': " ".join((87            'playlist-read-private',88            'playlist-read-collaborative',89            'user-library-read',90        ))91    })92    url = f'https://accounts.spotify.com/authorize?{url_params}'93    print(94        "",95        "Please open the following URL in a Web Browser to continue:",96        url,97        sep='\n'98    )99    return listen_for_token(port=SERVER_PORT)100def request_refresh_token(token_params):101    request_params = urlencode({102        'redirect_uri': REDIRECT_URI,103        'client_id': SPOTIFY_CLIENT_ID,104        'client_secret': SPOTIFY_CLIENT_SECRET,105        **token_params106    }).encode()107    url = f'https://accounts.spotify.com/api/token'108    with urlopen(url, data=request_params) as response:109        return json.loads(response.read().decode())110def activate_refresh_token(refresh_token):111    token_params = {112        'grant_type': 'authorization_code',113        'code': refresh_token114    }115    content = request_refresh_token(token_params)116    return content['access_token'], content['refresh_token']117def redeem_refresh_token(refresh_token):118    token_params = {119        'grant_type': 'refresh_token',120        'refresh_token': refresh_token121    }122    content = request_refresh_token(token_params)123    return content['access_token']124def save_token_to_file(token):125    token_path = Path(TOKEN_FILE).resolve()126    token_path.touch(0o600, exist_ok=True)127    token_path.write_text(token)128def get_token(restore_token=True, save_token=True):129    if restore_token and TOKEN_FILE.resolve().exists():130        token = Path(TOKEN_FILE).read_text().strip()131        if REFRESHABLE_AUTH:132            try:133                token = redeem_refresh_token(token)134            except HTTPError:135                # This will fail if the token wasn't a refresh token136                pass137        # Check that the token is valid138        req = Request(139            'https://api.spotify.com/v1/me',140            headers={'Authorization': f'Bearer {token}'}141        )142        try:143            urlopen(req)144        except HTTPError:145            pass146        else:147            return token148    token = prompt_user_for_auth()149    if REFRESHABLE_AUTH:150        token, savable_token = activate_refresh_token(token)151    else:152        savable_token = token153    if save_token:154        save_token_to_file(savable_token)155    return token156if __name__ == '__main__':...

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.