KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run stestr automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
