How to use show_outcome method in stestr

Best Python code snippet using stestr_python

spire_spiffe_id.py

Source:spire_spiffe_id.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3#4# Copyright (c) 2021 Patrice Congo <@congop>.5#6# This file is part of io_patricecongo.spire7# (see https://github.com/congop/io_patricecongo.spire).8#9# This program is free software: you can redistribute it and/or modify10# it under the terms of the GNU General Public License as published by11# the Free Software Foundation, either version 3 of the License, or12# (at your option) any later version.13#14# This program is distributed in the hope that it will be useful,15# but WITHOUT ANY WARRANTY; without even the implied warranty of16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the17# GNU General Public License for more details.18#19# You should have received a copy of the GNU General Public License20# along with this program. If not, see <http://www.gnu.org/licenses/>.#21# https://github.com/ansible/ansible-modules-core/blob/devel/database/postgresql/postgresql_db.py22import copy23import functools24from typing import Any, Dict25from ansible.module_utils.basic import AnsibleModule26from ansible_collections.io_patricecongo.spire.plugins.module_utils import (27    logging,28    spire_server_entry_cmd,29)30from ansible_collections.io_patricecongo.spire.plugins.module_utils.spire_server_entry_cmd import (31    Params,32    RegistrationEntry,33    SpireServerEntryShowOutcome,34)35# from . import spire_server_entry_cmd36ANSIBLE_METADATA = {37    'metadata_version': '0.0.1',38    'status': ['preview'],39    'supported_by': 'community'40}41DOCUMENTATION = '''42---43module: spire_spiffe_id44short_description: Ensure spiffe-ID is present or absent45version_added: "0.0.1"46description:47    - "It registers a spiffe-ID or removes it using spire-server CLI"48options:49    state:50        description:51            - specifies whether the entry should be present or not52        type: str53        required: False54        default: present55        choices: [absent, present]56    identity_args:57        description:58            - this specifies what constitutes the natural key.59            - It can be used to identify the entry which correspond to the given parameters without using the entry-id60            - "node=True --> a node spiffe-id is being specified"61            - "downstream=True -> a downstream server entry is being specified"62            - "node=False and downstream=False -> a workload entry is being specified"63            - other combinations of value for node and downstream are probably an input error64        type: list65        elements: str66        default: ["spiffe_id", "parent_id", "node", "downstream", "selector"]67    admin:68        description:69            - If set, the SPIFFE ID in this entry will be granted access to the Registration API70        required: false71    # data:72    #     description:73    #         - Path to a file containing registration data in JSON format (optional).74    #     required: false75    dns_name:76        description:77            - A DNS name that will be included in SVIDs issued based on this entry, where appropriate.78            - Can be used more than once79        required: false80    downstream:81        description:82            - A boolean value that, when set, indicates that the entry describes a downstream SPIRE server83        required: false84    entry_expiry:85        description:86            - An expiry, from epoch in seconds, for the resulting registration entry to be pruned from the datastore.87            - Please note that this is a data management feature and not a security feature (optional).88        required: false89    federates_with:90        description:91            - A list of trust domain SPIFFE IDs representing the trust domains this registration entry federates with.92            - A bundle for that trust domain must already exist93        required: false94    node:95        description:96            - If set, this entry will be applied to matching nodes rather than workloads97        required: false98    parent_id:99        description:100            - The SPIFFE ID of this record's parent.101    registration_uds_path:102        description: Path to the SPIRE server registration api socket /tmp/spire-registration.sock103        required: false104        default: /tmp/spire-registration.sock105    selector:106        description:107            - "A colon-delimited type:value selector used for attestation."108            - This parameter can be used more than once, to specify multiple selectors that must be satisfied.109        required: false110    spiffe_id:111        description:112            - The SPIFFE ID that this record represents and will be set to the SVID issued.113        required: true114    ttl:115        description:116            - A TTL, in seconds, for any SVID issued as a result of this record.117        required: False118        default: 3600119    spire_server_cmd:120        description:121            - Name of path of the spire-server command122        default: spire-server123author:124    - Patrice Congo (@congop)125'''126EXAMPLES = '''127- name: "Ensure spiffe id available"128      io_patricecongo.spire.spire_spiffe_id:129        state: present130        dns_name:131          - node1.local1132          - node1.local2133          - node1.local3134        downstream: no135        identity_args:136          - spiffe_id137          - parent_id138        parent_id: spiffe://example.org/myagent139        selector:140          - unix:user:etcd141          - unix:gid:1000142          - updated:yes143        spiffe_id: spiffe://example.org/myagent/etcd144        spire_server_cmd: /opt/spire/bin/spire-server145'''146RETURN = '''147state:148    description: state149    type: str150    returned: always151'''152def _module_args() -> Dict[str, Dict[str, Any]]:153    module_args: Dict[str, Dict[str, Any]] = dict(154        # name=dict(type='str', required=True),155        # new=dict(type='bool', required=False, default=False),156        # If set, the SPIFFE ID in this entry will be granted access to the Registration API157        admin=dict(type="bool", required=False),158        # Path to a file containing registration data in JSON format (optional).159        # data=dict(type="str", required=False),160        # -dns	A DNS name that will be included in SVIDs issued based on this entry, where appropriate.161        #       Can be used more than once162        dns_name=dict(type="list", elements="str", required=False),163        # -downstream	A boolean value that, when set, indicates that the entry describes a downstream SPIRE server164        downstream=dict(type="bool", required=False),165        # -entryExpiry	An expiry, from epoch in seconds, for the resulting registration entry to be pruned from166        #               the datastore. Please note that this is a data management feature and not a security167        #               feature (optional).168        entry_expiry=dict(type="str", required=False),169        # -federatesWith	A list of trust domain SPIFFE IDs representing the trust domains this registration entry170        #                   federates with. A bundle for that trust domain must already exist171        federates_with=dict(type="str", required=False),172        # -node	If set, this entry will be applied to matching nodes rather than workloads173        node=dict(type="str", required=False),174        # -parentID	The SPIFFE ID of this record's parent.175        parent_id=dict(type="str", required=False),176        # -registrationUDSPath Path to the SPIRE server registration api socket /tmp/spire-registration.sock177        registration_uds_path=dict(type="str", required=False),178        # -selector	A colon-delimited type:value selector used for attestation. This parameter can be used more179        #           than once, to specify multiple selectors that must be satisfied.180        selector=dict(type="list", elements="str", required=False),181        # -spiffeID	The SPIFFE ID that this record represents and will be set to the SVID issued.182        spiffe_id=dict(type="str", required=False),183        # -ttl	A TTL, in seconds, for any SVID issued as a result of this record.	3600184        ttl=dict(type="int", required=False),185        spire_server_cmd=dict(type=str, required=False,186                              default="spire-server"),187        state=dict(default='present', choices=['absent', 'present']),188        # this specifies what constitutes the natural key.189        # It can be used to identify the entry which correspond to the given parameters without using the entry-id190        # node=True --> a node spiffe-id is being specified191        # downstream=True -> a downstream server entry is being specified192        # node=False and downstream=False -> a workload entry is being specified193        # other combinations of value for node and downstream are probably an input error194        identity_args=dict(type="list", elements="str",195                           default=["spiffe_id", "parent_id", "node", "downstream", "selector"])196    )197    return module_args198def run_module() -> None:199    module_args = _module_args()200    result = dict(201        state='absent',202        changed=False,203    )204    module = AnsibleModule(205        argument_spec=module_args,206        supports_check_mode=True207    )208    params: Params = Params(copy.deepcopy(module.params))209    func_run_command = functools.partial(AnsibleModule.run_command, module)210    func_log = logging.CachingLogger(module.log)211    try:212        show_outcome: SpireServerEntryShowOutcome = spire_server_entry_cmd.cmd_server_entry_show(213            func_run_command,214            func_log,215            params,216        )217        if show_outcome.exec_failed():218            msg = f"""219                    Fail to execute <entry show> cmd:220                        parse_error={show_outcome.parse_error}221                        rc={show_outcome.rc}222                        stdout={show_outcome.stdout}223                        stderr={show_outcome.stderr}224                        entries={show_outcome.entries}225                    """226            raise RuntimeError(msg)227        if show_outcome.parsing_failed():228            msg = f"""229                    Fail to parse <entry show> output:230                        parse_error={show_outcome.parse_error}231                        rc={show_outcome.rc}232                        stdout={show_outcome.stdout}233                        stderr={show_outcome.stderr}234                        entries={show_outcome.entries}235                    """236            raise RuntimeError(msg)237        # identity_params = params["identity_args"]238        actual_list = spire_server_entry_cmd.entries_having_same_identity(params, show_outcome.entries)239        if len(actual_list) > 1:240            module.fail_json(msg=f'Cannot handle more than one identified corresponding entries: {actual_list}')241        actual: RegistrationEntry = RegistrationEntry() if len(actual_list) == 0 else actual_list[0]242        need_change = spire_server_entry_cmd.need_change(params, actual)243        if need_change:244            debug_msg = f"""245                need_change={need_change}246                params={params}247                actual={actual}248                actual_list={actual_list}249                entries={show_outcome.entries}250                """251            func_log(debug_msg)252        state = module.params['state']253        result["state"] = state254        result["changed"] = need_change255        if need_change:256            if not module.check_mode:257                if state == "absent":258                    # TODO actual is entry not a Param so merging is not type safe intro actual.to_params()259                    # or params.to_entry(..) because it is the command being built260                    # so command data format is needed which is entry data261                    merged = {**actual, **params}262                    spire_server_entry_cmd.cmd_server_entry_delete(func_run_command, func_log, merged)263                if state == "present":264                    if not actual:265                        spire_server_entry_cmd.cmd_server_entry_create(func_run_command, func_log, params)266                    else:267                        # TODO actual is entry not a Param so merging is not type safe268                        merged = {**actual, **params}269                        spire_server_entry_cmd.cmd_server_entry_update(func_run_command, func_log, merged)270        result["debug_msg"] = str(func_log.messages)271        module.exit_json(**result)272    except Exception as e:273        module.fail_json(msg=f"Exception while running module:{func_log.messages}", exception=str(e))274def main() -> None:275    run_module()276if __name__ == '__main__':...

ajax.py

Source:ajax.py

1'''2Created on Jul 23, 20143@author: thavr4'''5from budget.debug import get_debug_user6from django.db.models import Max, Min7from budget.models import User, Users8from budget.views import show_mock9from dateutil.parser import parse10import datetime11from django.contrib.auth.decorators import login_required12@login_required(login_url="/login/")13def search_result(request):14    ##DEBUG15    user = get_debug_user(request)16    ##DEBUG        17    search_criterium = parse_request(request)18    empty_set = []19    if search_criterium["give_no_result"]:20        return empty_set21    output_set = get_initial_transactions_set(request)22    if is_empty(output_set):23        return empty_set    24    if search_criterium["show_outcome"] and search_criterium["show_income"]:25        pass26    elif search_criterium["show_outcome"]:27        output_set = return_only_outcome(output_set)28    elif search_criterium["show_income"]:29        output_set = return_only_income(output_set)30    31    if is_empty(output_set):32        return empty_set33                34    output_set = filter_transactions_by_amount(output_set,\35                            min_amount = search_criterium["min_amount"], \36                            max_amount = search_criterium["max_amount"])37    38    if is_empty(output_set):39        return empty_set40    41    output_set = filter_transactions_by_date(output_set,\42                            before = search_criterium["before"], \43                            after = search_criterium["after"])44    45    if is_empty(output_set):46        return empty_set47    48    output_set = filter_transactions_by_users(output_set,\49                            users = search_criterium["users"])50    51    if is_empty(output_set):52        return empty_set53    54    output_set = filter_transactions_by_purpose(output_set,\55                            value=search_criterium["purpose"], \56                            exact=search_criterium["purpose_exact"])57    58    if is_empty(output_set):59        return empty_set60    61    output_set = filter_transactions_by_comment(output_set,\62                            value=search_criterium["comment"], \63                            exact=search_criterium["comment_exact"])64    65    return order_by_time(output_set)66def is_empty(input_set):67    flag = True68    for current_set in input_set:69        if len(current_set) > 0:70            flag = False71    return flag72    73@login_required(login_url="/login/")    74def get_initial_transactions_set(request):75    ##DEBUG76    user = get_debug_user(request)77    ##DEBUG78    return [user.get_all_public_family_transactions(), \79            user.get_all_privat_user_transactions()]80# -----------------------------Section of amount--------------------------------81def filter_transactions_by_amount(transactions, \82                                  min_amount = "", \83                                  max_amount = ""):84    if min_amount =="":85        min_amount = -9999999999.9986    if max_amount == "":87        max_amount = 9999999999.9988    prom_res = [transaction.filter(amount__gte=min_amount).\89                filter(amount__lte=max_amount) for transaction in transactions]    90    return prom_res91def return_only_income(transactions):92    return filter_transactions_by_amount(transactions, min_amount=0)93def return_only_outcome(transactions):94    return filter_transactions_by_amount(transactions, max_amount=0)95# -----------------------------End of Section of amount-------------------------96# -----------------------------Section of date----------------------------------97def filter_transactions_by_date(transactions, \98                                  before = "", \99                                  after = ""):100    if before == "" and after=="":101        return transactions102    elif after == "":103        return [transaction.filter(transactionstime__lte = before) \104                             for transaction in transactions]105    elif before == "":106        return [transaction.filter(transactionstime__gte = after)  \107                             for transaction in transactions]108    else:109        return [transaction.filter(transactionstime__gte = after).\110            filter(transactionstime__lte = before) \111                             for transaction in transactions]112# -----------------------------End of Section of date---------------------------113# -----------------------------Section of text search---------------------------114def filter_transactions_by_text_parameter(transactions, field, \115                                          value="", exact=False):116    if value == "":117        return transactions118    if exact:119        if field == "purpose":120            return [transaction.filter(purpose__iexact=value) \121                             for transaction in transactions]122        elif field == "comment":123            return [transaction.filter(comment__iexact=value) \124                             for transaction in transactions]125        else:126            raise KeyError127    else:128        if field == "purpose":129            return [transaction.filter(purpose__icontains = value) \130                             for transaction in transactions]131        elif field == "comment":132            return [transaction.filter(comment__icontains=value) \133                             for transaction in transactions]134        else:135            raise KeyError136# -----------------------------End section of text search-----------------------137# -----------------------------Section of purpose-------------------------------138def filter_transactions_by_purpose(transactions,  \139                                          value="", exact=False):140    141    return filter_transactions_by_text_parameter(transactions, "purpose", \142                                                 value=value, exact=exact)143# -----------------------------End of Section of purpose------------------------144# -----------------------------Section of comment-------------------------------145def filter_transactions_by_comment(transactions,  \146                                          value="//", exact=False):147    148    return filter_transactions_by_text_parameter(transactions, "comment", \149                                                 value=value, exact=exact)150# -----------------------------End of Section of comment------------------------151# -----------------------------Section of users---------------------------------152def filter_transactions_by_users(transactions, users = []):153    154    # NOTE!!! users is ALREADY list of Users objects155    if users == []:156        return transactions157    return [transaction.filter(user__in = users) \158                             for transaction in transactions]159# -----------------------------End section of users---------------------------------160# -----------------------------Section of auxilaries----------------------------161def get_border_value_of_transactions(user, border = "max", \162                                         criterium = "transactionstime"):163        tr = user.get_all_public_family_transactions()164        if border == "max":165            tr1 = tr.aggregate(Max(criterium))166            return tr1[criterium+"__max"]167        elif border == "min":168            tr1 = tr.aggregate(Min(criterium))169            return tr1[criterium+"__min"]170        171        raise KeyError172    173def get_max_value_of_transaction(user, criterium = "transactionstime"):174        return get_border_value_of_transactions(user, border = "max", \175                                                     criterium = criterium)176    177def get_min_value_of_transaction(user, criterium = "transactionstime"):178        return get_border_value_of_transactions(user, border = "min", \179                                                     criterium = criterium)180@login_required(login_url="/login/")        181def parse_request(request):182    ##DEBUG183    user = get_debug_user(request)184    ##DEBUG185    output = {}186    try:187        after = request.GET["begin"]188    except KeyError:189        after = get_min_value_of_transaction(user, "transactionstime")190    191    if after != "":192        try:193            after = parse(after)194        except:195            after = ""196        197    output["after"] = after198    199    try:200        before = request.GET["end"]201    except KeyError:202        before = get_max_value_of_transaction(user, "transactionstime")203        204    if before != "":205        try:206            before = parse(before)207        except:208            before = ""209    output["before"] = before210    211    try:212        max_amount = request.GET["maxamount"]213    except KeyError:214        max_amount = get_max_value_of_transaction(user, "amount")215    try:216        float(max_amount)217    except ValueError:218        max_amount = ""219    except TypeError:220        max_amount = ""221    output["max_amount"] = max_amount222    223    try:224        min_amount = request.GET["minamount"]225    except KeyError:226        min_amount = get_min_value_of_transaction(user, "amount")227    try:228        float(min_amount)229    except ValueError:230        min_amount = ""231    except TypeError:232        min_amount = ""233    output["min_amount"] = min_amount234    235#     try:236#         show_private = request.GET["private"]237#         output["show_private"] = True238#     except KeyError:239#         output["show_private"] = False240#         241#     try:242#         show_private = request.GET["public"]243#         output["show_public"] = True244#     except KeyError:245#         output["show_public"] = False246    247    try:248        show_income = request.GET["income"]249        output["show_income"] = True250    except KeyError:251        output["show_income"] = False  252        253    try:254        give_no_result = request.GET["nosearch"]255        output["give_no_result"] = True256    except KeyError:257        output["give_no_result"] = False  258    259    try:260        show_outcome = request.GET["outcome"]261        output["show_outcome"] = True262    except KeyError:263        output["show_outcome"] = False264    265    try:266        users = request.GET.getlist('users')267        output["users"] = []268        for user_email in users:269            try:270                user_auth = User.objects.get(email = user_email)271                user = Users.objects.get(user = user_auth)272                output["users"].append(user)273            except User.DoesNotExist:274                pass275            except Users.DoesNotExist:276                pass277    except KeyError:278        output["users"] = []        279        280    output["purpose_exact"] = define_exact("purposecomparation", request)281    output["comment_exact"] = define_exact("commentcomparation", request)282    output["purpose"] = define_text("purpose", request)283    output["comment"] = define_text("comment", request)284    285    return output 286def define_exact(parameter, request):287    try:288        exact = request.GET[parameter]289        if exact == "equals":290            return True291        else:292            return False293    except KeyError:294        return False295    296def define_text(parameter, request):297    try:298        text = request.GET[parameter]299        return text300    except KeyError:301        return ""    302    303def order_by_time(transactions, asc=False):304    if asc:305        param = "transactionstime"306    else:307        param = "-transactionstime"308    return [transaction.order_by(param)  \309                             for transaction in transactions]310    311    312    313    314            315        ...

quick_draw.py

Source:quick_draw.py

...14PLAYER_PIXELS = {15  1 : (0,1,2,3,4),16  2 : (5,6,7,8,9)17}18def show_outcome(player, winner):19    # Turn them all off20    cpx.pixels.fill(0)21    22    # Set pixel color23    if winner:24        color = 0x00FF0025    else:26        color = 0xFF000027    # Show which player won/lost28    for p in PLAYER_PIXELS[player]:29        cpx.pixels[p] = color30    31    # Play a little tune32    if winner:33        cpx.play_tone(800, 0.2)34        cpx.play_tone(900, 0.2)35        cpx.play_tone(1400, 0.2)36        cpx.play_tone(1100, 0.2)37    else:38        cpx.play_tone(200, 1)39    # Sit here forever40    while True:41        pass      42# Seed the random function with noise43a4 = AnalogIn(board.A4)44a5 = AnalogIn(board.A5)45a6 = AnalogIn(board.A6)46a7 = AnalogIn(board.A7)47seed  = a4.value48seed += a5.value49seed += a6.value50seed += a7.value51random.seed(seed)52# Wait a random amount of time53count_time = random.randrange(SHORTEST_DELAY, LONGEST_DELAY+1)54start_time = time.monotonic()55while time.monotonic() - start_time < count_time:56    # Check if player draws too soon57    if cpx.button_a:58        show_outcome(1, False)59    if cpx.button_b:60        show_outcome(2, False)    61# Turn on all the NeoPixels62cpx.pixels.fill(0xFFFFFF)63# Check for player draws64while True:65    if cpx.button_a:66        show_outcome(1, True)67    if cpx.button_b:...

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.