How to use test_address method in Nose

Best Python code snippet using nose

testgriddaemons.py

Source:testgriddaemons.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3#4# --- BEGIN_HEADER ---5#6# testgriddaemons- Set of unit tests for grid daemon helper functions7# Copyright (C) 2010-2020  The MiG Project lead by Brian Vinter8#9# This file is part of MiG.10#11# MiG is free software: you can redistribute it and/or modify12# it under the terms of the GNU General Public License as published by13# the Free Software Foundation; either version 2 of the License, or14# (at your option) any later version.15#16# MiG is distributed in the hope that it will be useful,17# but WITHOUT ANY WARRANTY; without even the implied warranty of18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the19# GNU General Public License for more details.20#21# You should have received a copy of the GNU General Public License22# along with this program; if not, write to the Free Software23# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.24#25# -- END_HEADER ---26#27"""Unit tests for grid daemon helper functions"""28from __future__ import print_function29import sys30import time31import logging32from mig.shared.griddaemons.ratelimits import default_max_user_hits, \33    expire_rate_limit, hit_rate_limit, update_rate_limit34from mig.shared.griddaemons.sessions import active_sessions, \35    clear_sessions, get_active_session, get_open_sessions, \36    track_open_session, track_close_session, track_close_expired_sessions37# TODO: Add unit test for validate_auth_attempt ?38if __name__ == "__main__":39    from mig.shared.conf import get_configuration_object40    conf = get_configuration_object()41    logging.basicConfig(filename=None, level=logging.INFO,42                        format="%(asctime)s %(levelname)s %(message)s")43    conf.logger = logging44    test_proto, test_address, test_port, test_id, test_session_id = \45        'DUMMY', '127.0.0.42', 42000, \46        'user@some-domain.org', 'DUMMY_SESSION_ID'47    test_pw = "T3stp4ss"48    invalid_id = 'root'49    print("Running unit test on rate limit functions")50    print("Force expire all")51    expired = expire_rate_limit(conf, test_proto, fail_cache=0)52    print("Expired: %s" % expired)53    this_pw = test_pw54    print("Emulate rate limit")55    for i in range(default_max_user_hits-1):56        hit = hit_rate_limit(conf, test_proto, test_address, test_id)57        print("Blocked: %s" % hit)58        update_rate_limit(conf, test_proto, test_address, test_id, False,59                          this_pw)60        print("Updated fail for %s:%s from %s" % \61              (test_id, this_pw, test_address))62        this_pw += 'x'63        time.sleep(1)64    hit = hit_rate_limit(conf, test_proto, test_address, test_id)65    print("Blocked: %s" % hit)66    print("Check with original user and password again")67    update_rate_limit(conf, test_proto, test_address, test_id, False, test_pw)68    hit = hit_rate_limit(conf, test_proto, test_address, test_id)69    print("Blocked: %s" % hit)70    print("Check with original user and new password again to hit limit")71    update_rate_limit(conf, test_proto, test_address, test_id, False, this_pw)72    hit = hit_rate_limit(conf, test_proto, test_address, test_id)73    print("Blocked: %s" % hit)74    other_proto, other_address = "BOGUS", '127.10.20.30'75    other_id, other_pw = 'other@some.org', "0th3rP4ss"76    print("Update with other proto")77    update_rate_limit(conf, other_proto, test_address, test_id, False, test_pw)78    print("Update with other address")79    update_rate_limit(conf, test_proto, other_address, test_id, False, test_pw)80    print("Update with other user")81    update_rate_limit(conf, test_proto, test_address, other_id, False, test_pw)82    print("Check with same user from other address")83    hit = hit_rate_limit(conf, test_proto, other_address, test_id)84    print("Blocked: %s" % hit)85    print("Check with other user from same address")86    hit = hit_rate_limit(conf, test_proto, test_address, other_id)87    print("Blocked: %s" % hit)88    time.sleep(2)89    print("Force expire some entries")90    expired = expire_rate_limit(conf, test_proto,91                                fail_cache=default_max_user_hits)92    print("Expired: %s" % expired)93    print("Test reset on success")94    hit = hit_rate_limit(conf, test_proto, test_address, test_id)95    print("Blocked: %s" % hit)96    update_rate_limit(conf, test_proto, test_address, test_id, True, test_pw)97    print("Updated success for %s from %s" % (test_id, test_address))98    hit = hit_rate_limit(conf, test_proto, test_address, test_id)99    print("Blocked: %s" % hit)100    print("Check with same user from other address")101    hit = hit_rate_limit(conf, test_proto, other_address, test_id)102    print("Blocked: %s" % hit)103    print("Check with other user from same address")104    hit = hit_rate_limit(conf, test_proto, test_address, other_id)105    print("Blocked: %s" % hit)106    print("Check with invalid user from same address")107    hit = hit_rate_limit(conf, test_proto, test_address, invalid_id)108    print("Blocked: %s" % hit)109    print("Test active session counting")110    active_count = active_sessions(conf, test_proto, test_id)111    print("Open sessions: %d" % active_count)112    print("Clear sessions")113    clear_sessions(conf, test_proto)114    active_count = active_sessions(conf, test_proto, test_id)115    print("Open sessions: %d" % active_count)116    print("Track open session")117    track_open_session(conf, test_proto, test_id, test_address, test_port)118    active_count = active_sessions(conf, test_proto, test_id)119    print("Open sessions: %d" % active_count)120    print("Track open session")121    track_open_session(conf, test_proto, test_id, test_address, test_port+1)122    active_count = active_sessions(conf, test_proto, test_id)123    print("Open sessions: %d" % active_count)124    print("Track close session")125    track_close_session(conf, test_proto, test_id, test_address, test_port, )126    active_count = active_sessions(conf, test_proto, test_id)127    print("Open sessions: %d" % active_count)128    print("Track close session")129    track_close_session(conf, test_proto, test_id, test_address, test_port+1, )130    active_count = active_sessions(conf, test_proto, test_id)131    print("Open sessions: %d" % active_count)132    print("Test session tracking functions")133    expected_session_keys = ['ip_addr',134                             'tcp_port',135                             'session_id',136                             'authorized',137                             'client_id',138                             'timestamp']139    print("Track open session #1")140    open_session = track_open_session(conf,141                                      test_proto,142                                      test_id,143                                      test_address,144                                      test_port,145                                      test_session_id,146                                      authorized=True)147    active_count = active_sessions(conf, test_proto, test_id)148    if active_count != 1:149        print("ERROR: Excpected 1 active session(s) for: %s, found: %d" \150            % (test_id, active_count))151        sys.exit(1)152    active_count = active_sessions(conf, test_proto, test_id+"_1")153    if active_count != 0:154        print("ERROR: Excpected 0 active session(s) for: %s, found: %d" \155            % (test_id+"_1", active_count))156        sys.exit(1)157    if isinstance(open_session, dict):158        if open_session.keys() == expected_session_keys:159            print("OK")160        else:161            print("ERROR: Invalid session dictionary: '%s'" \162                % (open_session))163            sys.exit(1)164    else:165        print("ERROR: Expected dictionary: %s" % type(open_session))166        sys.exit(1)167    print("Track open session #2")168    open_session = track_open_session(conf,169                                      test_proto,170                                      test_id,171                                      test_address,172                                      test_port+1,173                                      test_session_id+"_1",174                                      authorized=True)175    active_count = active_sessions(conf, test_proto, test_id)176    if active_count != 2:177        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \178            % (test_id, active_count))179        sys.exit(1)180    active_count = active_sessions(conf, test_proto, test_id+"_1")181    if active_count != 0:182        print("ERROR: Excpected 0 active session(s) for: %s, found: %d" \183            % (test_id+"_1", active_count))184        sys.exit(1)185    if isinstance(open_session, dict):186        if open_session.keys() == expected_session_keys:187            print("OK")188        else:189            print("ERROR: Invalid session dictionary: '%s'" \190                % (open_session))191            sys.exit(1)192    else:193        print("ERROR: Expected dictionary: %s" % type(open_session))194        sys.exit(1)195    print("Track open session #3")196    open_session = track_open_session(conf,197                                      test_proto,198                                      test_id+"_1",199                                      test_address,200                                      test_port+2,201                                      test_session_id+"_2",202                                      authorized=True)203    active_count = active_sessions(conf, test_proto, test_id)204    if active_count != 2:205        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \206            % (test_id, active_count))207        sys.exit(1)208    active_count = active_sessions(conf, test_proto, test_id+"_1")209    if active_count != 1:210        print("ERROR: Excpected 1 active session(s) for: %s, found: %d" \211            % (test_id+"_1", active_count))212        sys.exit(1)213    if isinstance(open_session, dict):214        if open_session.keys() == expected_session_keys:215            print("OK")216        else:217            print("ERROR: Invalid session dictionary: '%s'" \218                % (open_session))219            sys.exit(1)220    else:221        print("ERROR: Expected dictionary: %s" % type(open_session))222        sys.exit(1)223    print("Track open session #4")224    open_session = track_open_session(conf,225                                      test_proto,226                                      test_id+"_1",227                                      test_address,228                                      test_port+3,229                                      test_session_id+"_3",230                                      authorized=True)231    active_count = active_sessions(conf, test_proto, test_id)232    if active_count != 2:233        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \234            % (test_id, active_count))235        sys.exit(1)236    active_count = active_sessions(conf, test_proto, test_id+"_1")237    if active_count != 2:238        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \239            % (test_id+"_1", active_count))240        sys.exit(1)241    if isinstance(open_session, dict):242        if open_session.keys() == expected_session_keys:243            print("OK")244        else:245            print("ERROR: Invalid session dictionary: '%s'" \246                % (open_session))247            sys.exit(1)248    else:249        print("ERROR: Expected dictionary: %s" % type(open_session))250    print("Track get open sessions #1")251    cur_open_sessions = get_open_sessions(conf, 'INVALID')252    if isinstance(cur_open_sessions, dict):253        if not cur_open_sessions:254            print("OK")255        else:256            print("ERROR: Excpected empty dictionary: %s" \257                % cur_open_sessions)258            sys.exit(1)259    else:260        print("ERROR: Expected dictionary: %s" % type(cur_open_sessions))261        sys.exit(1)262    print("Track get open sessions #2")263    cur_open_sessions = get_open_sessions(conf, test_proto, 'INVALID')264    if isinstance(cur_open_sessions, dict):265        if not cur_open_sessions:266            print("OK")267        else:268            print("ERROR: Excpected empty dictionary: %s" \269                % cur_open_sessions)270            sys.exit(1)271    else:272        print("ERROR: Expected dictionary: %s" % type(cur_open_sessions))273        sys.exit(1)274    print("Track get open sessions #3")275    cur_open_sessions = get_open_sessions(conf, test_proto)276    if isinstance(cur_open_sessions, dict):277        if len(cur_open_sessions.keys()) != 4:278            print("ERROR: Expected dictionary #keys: 4" \279                + ", found: %s, %s" % (len(cur_open_sessions.keys()),280                                       cur_open_sessions.keys()))281            sys.exit(1)282        status = True283        for (key, val) in cur_open_sessions.iteritems():284            if not isinstance(val, dict) \285                    or val.keys() != expected_session_keys:286                status = False287                print("ERROR: Invalid session dictionary: '%s'" \288                    % (val))289                sys.exit(1)290        if status:291            print("OK")292    else:293        print("ERROR: Expected dictionary: %s" % type(cur_open_sessions))294        sys.exit(1)295    print("Track get open sessions #4")296    cur_open_sessions = get_open_sessions(conf,297                                          test_proto,298                                          client_id=test_id)299    if isinstance(cur_open_sessions, dict):300        if len(cur_open_sessions.keys()) != 2:301            print("ERROR: Expected dictionary #keys: 2" \302                + ", found: %s, %s" % (len(cur_open_sessions.keys()),303                                       cur_open_sessions.keys()))304            sys.exit(1)305        status = True306        for (key, val) in cur_open_sessions.iteritems():307            if not isinstance(val, dict) \308                    or val.keys() != expected_session_keys:309                status = False310                print("ERROR: Invalid session dictionary: '%s'" \311                    % (val))312                sys.exit(1)313        if status:314            print("OK")315    else:316        print("ERROR: Expected dictionary: %s" % type(cur_open_sessions))317        sys.exit(1)318    print("Track get active session #1")319    active_session = get_active_session(conf,320                                        'INVALID',321                                        test_id,322                                        test_session_id)323    if isinstance(active_session, dict):324        if not active_session:325            print("OK")326        else:327            print("ERROR: Excpected empty dictionary: %s" \328                % active_session)329            sys.exit(1)330    else:331        print("ERROR: Expected dictionary: %s" % type(active_session))332        sys.exit(1)333    print("Track get active session #2")334    active_session = get_active_session(conf,335                                        test_proto,336                                        'INVALID',337                                        test_session_id)338    if isinstance(active_session, dict):339        if not active_session:340            print("OK")341        else:342            print("ERROR: Excpected empty dictionary: %s" \343                % active_session)344            sys.exit(1)345    else:346        print("ERROR: Expected dictionary: %s" % type(active_session))347        sys.exit(1)348    print("Track get active session #3")349    active_session = get_active_session(conf,350                                        test_proto,351                                        test_id,352                                        'INVALID')353    if isinstance(active_session, dict):354        if not active_session:355            print("OK")356        else:357            print("ERROR: Excpected empty dictionary: %s" \358                % active_session)359            sys.exit(1)360    else:361        print("ERROR: Expected dictionary: %s" % type(active_session))362        sys.exit(1)363    print("Track get active session #4")364    active_session = get_active_session(conf,365                                        test_proto,366                                        test_id,367                                        test_session_id)368    if isinstance(active_session, dict):369        if active_session.keys() == expected_session_keys:370            print("OK")371        else:372            print("ERROR: Invalid session dictionary: '%s'" \373                % (active_session))374            sys.exit(1)375    else:376        print("ERROR: Expected dictionary: %s" % type(active_session))377        sys.exit(1)378    print("Track close session #1")379    close_session = track_close_session(conf,380                                        'INVALID',381                                        test_id,382                                        test_address,383                                        test_port,384                                        session_id=test_session_id)385    active_count = active_sessions(conf, test_proto, test_id)386    if active_count != 2:387        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \388            % (test_id, active_count))389        sys.exit(1)390    active_count = active_sessions(conf, test_proto, test_id+"_1")391    if active_count != 2:392        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \393            % (test_id+"_1", active_count))394        sys.exit(1)395    if isinstance(close_session, dict):396        if not close_session:397            print("OK")398        else:399            print("ERROR: Excpected empty dictionary: %s" \400                % close_session)401            sys.exit(1)402    else:403        print("ERROR: Expected dictionary: %s" % type(close_session))404        sys.exit(1)405    print("Track close session #2")406    close_session = track_close_session(conf,407                                        test_proto,408                                        'INVALID',409                                        test_address,410                                        test_port,411                                        session_id=test_session_id)412    active_count = active_sessions(conf, test_proto, test_id)413    if active_count != 2:414        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \415            % (test_id, active_count))416        sys.exit(1)417    active_count = active_sessions(conf, test_proto, test_id+"_1")418    if active_count != 2:419        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \420            % (test_id+"_1", active_count))421        sys.exit(1)422    if isinstance(close_session, dict):423        if not close_session:424            print("OK")425        else:426            print("ERROR: Excpected empty dictionary: %s" \427                % close_session)428            sys.exit(1)429    else:430        print("ERROR: Expected dictionary: %s" % type(close_session))431        sys.exit(1)432    print("Track close session #3")433    close_session = track_close_session(conf,434                                        test_proto,435                                        test_id,436                                        test_address,437                                        test_port,438                                        session_id=None)439    active_count = active_sessions(conf, test_proto, test_id)440    if active_count != 2:441        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \442            % (test_id, active_count))443        sys.exit(1)444    active_count = active_sessions(conf, test_proto, test_id+"_1")445    if active_count != 2:446        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \447            % (test_id+"_1", active_count))448        sys.exit(1)449    if isinstance(close_session, dict):450        if not close_session:451            print("OK")452        else:453            print("ERROR: Excpected empty dictionary: %s" \454                % close_session)455            sys.exit(1)456    else:457        print("ERROR: Expected dictionary: %s" % type(close_session))458        sys.exit(1)459    print("Track close session #4")460    close_session = track_close_session(conf,461                                        test_proto,462                                        test_id,463                                        test_address,464                                        test_port,465                                        session_id=test_session_id)466    active_count = active_sessions(conf, test_proto, test_id)467    if active_count != 1:468        print("ERROR: Excpected 1 active session(s) for: %s, found: %d" \469            % (test_id, active_count))470        sys.exit(1)471    active_count = active_sessions(conf, test_proto, test_id+"_1")472    if active_count != 2:473        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \474            % (test_id+"_1", active_count))475        sys.exit(1)476    if isinstance(close_session, dict):477        if close_session.keys() == expected_session_keys:478            print("OK")479        else:480            print("ERROR: Invalid session dictionary: '%s'" \481                % (close_session))482            sys.exit(1)483    else:484        print("ERROR: Expected dictionary: %s" % type(close_session))485        sys.exit(1)486    print("Track close expired sessions #1")487    expired_sessions = track_close_expired_sessions(conf,488                                                    'INVALID')489    active_count = active_sessions(conf, test_proto, test_id)490    if active_count != 1:491        print("ERROR: Excpected 1 active session(s) for: %s, found: %d" \492            % (test_id, active_count))493        sys.exit(1)494    active_count = active_sessions(conf, test_proto, test_id+"_1")495    if active_count != 2:496        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \497            % (test_id+"_1", active_count))498        sys.exit(1)499    if isinstance(expired_sessions, dict):500        if not expired_sessions:501            print("OK")502        else:503            print("ERROR: Excpected empty dictionary: %s" \504                % expired_sessions)505            sys.exit(1)506    else:507        print("ERROR: Expected dictionary: %s" % type(expired_sessions))508        sys.exit(1)509    print("Track close expired sessions #2")510    expired_sessions = track_close_expired_sessions(conf,511                                                    test_proto,512                                                    'INVALID')513    active_count = active_sessions(conf, test_proto, test_id)514    if active_count != 1:515        print("ERROR: Excpected 1 active session(s) for: %s, found: %d" \516            % (test_id, active_count))517        sys.exit(1)518    active_count = active_sessions(conf, test_proto, test_id+"_1")519    if active_count != 2:520        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \521            % (test_id+"_1", active_count))522        sys.exit(1)523    if isinstance(expired_sessions, dict):524        if not expired_sessions:525            print("OK")526        else:527            print("ERROR: Excpected empty dictionary: %s" \528                % expired_sessions)529            sys.exit(1)530    else:531        print("ERROR: Expected dictionary: %s" % type(expired_sessions))532        sys.exit(1)533    print("Track close expired sessions #3")534    expired_sessions = track_close_expired_sessions(conf,535                                                    test_proto,536                                                    client_id=test_id)537    active_count = active_sessions(conf, test_proto, test_id)538    if active_count != 0:539        print("ERROR: Excpected 0 active session(s) for: %s, found: %d" \540            % (test_id, active_count))541        sys.exit(1)542    active_count = active_sessions(conf, test_proto, test_id+"_1")543    if active_count != 2:544        print("ERROR: Excpected 2 active session(s) for: %s, found: %d" \545            % (test_id+"_1", active_count))546        sys.exit(1)547    if isinstance(expired_sessions, dict):548        if len(expired_sessions.keys()) == 1:549            status = True550            for (key, val) in expired_sessions.iteritems():551                if not isinstance(val, dict) \552                        or val.keys() != expected_session_keys:553                    status = False554                    print("ERROR: Invalid session dictionary: '%s'" \555                        % (val))556                    sys.exit(1)557            if status:558                print("OK")559        else:560            print("ERROR: Expected 1 expired session, found: %s" \561                % len(expired_sessions.keys()))562            sys.exit(1)563    else:564        print("ERROR: Expected dictionary: %s" % type(expired_sessions))565        sys.exit(1)566    print("Track close expired sessions #4")567    expired_sessions = track_close_expired_sessions(conf,568                                                    test_proto)569    active_count = active_sessions(conf, test_proto, test_id)570    if active_count != 0:571        print("ERROR: Excpected 0 active session(s) for: %s, found: %d" \572            % (test_id, active_count))573        sys.exit(1)574    active_count = active_sessions(conf, test_proto, test_id+"_1")575    if active_count != 0:576        print("ERROR: Excpected 0 active session(s) for: %s, found: %d" \577            % (test_id+"_1", active_count))578        sys.exit(1)579    if isinstance(expired_sessions, dict):580        if len(expired_sessions.keys()) == 2:581            status = True582            for (key, val) in expired_sessions.iteritems():583                if not isinstance(val, dict) \584                        or val.keys() != expected_session_keys:585                    status = False586                    print("ERROR: Invalid session dictionary: '%s'" \587                        % (val))588                    sys.exit(1)589            if status:590                print("OK")591        else:592            print("ERROR: Expected 2 expired session, found: %s" \593                % len(expired_sessions.keys()))594            sys.exit(1)595    else:596        print("ERROR: Expected dictionary: %s" % type(expired_sessions))...

address_tests.py

Source:address_tests.py

1#***2#*  GroceryDirect - Address Tests3#*  4#*  Author: Melanie Cornelius, mseryn5#*  Written for CS 425-02 Fall 2016 Final Project6#*7#*  Addresses:8#*  have:9#*      -- street10#*      -- apt (optional)11#*      -- city12#*      -- state13#*      -- zip code14#*      -- type15#*  get:16#*      -- street17#*      -- apt (if exists)18#*      -- city19#*      -- state20#*      -- zip code21#*      -- full address string?22#*      -- type23#*  modify:24#*      -- street25#*      -- apt26#*      -- city27#*      -- state28#*      -- zip code29#*      -- type30#***31import address32import person33# Default strings for tests34TEST_STREET         = "123 Test St."35TEST_APT_NO         = "Test Apt No. 1"36TEST_CITY           = "Test City"37TEST_STATE_CODE     = "AZ"38TEST_ZIP_CODE       = 1234539TEST_TYPE_STRING    = "shipping"40# Default strings for test person41FNAME       = "FirstName"42LNAME       = "LastName"43MIDDLE_INIT = "M" 44USERNAME    = "testaccount"45PASSWORD    = "testpassword"46# Test objects (avoid cluttering db)47CUSTOMER = person.check_credentials(USERNAME, PASSWORD)48if not CUSTOMER:49        CUSTOMER  = person.Person.new_person(USERNAME, PASSWORD, FNAME, LNAME, "customer", \50                           middle_initial = MIDDLE_INIT)51#**************************************************************************************************52#**  ADDRESS: PERSON53#**************************************************************************************************54def test_address_get_null_person():55    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \56        TEST_TYPE_STRING)57    assert(test_address.get_person().get_id() == None), "get_person did not correctly return \58        null for default for address"59def test_address_add_person():60    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \61        TEST_TYPE_STRING)62    test_address.add_person(CUSTOMER)63    assert(test_address.get_person().get_id() == CUSTOMER.get_id()), "get and add person did \64        not correctly function for address"65#**************************************************************************************************66#**  ADDRESS: DEFAULT FLAG67#**************************************************************************************************68def test_address_get_default_flag():69    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \70        TEST_TYPE_STRING)71    assert(test_address.get_default_flag() == False), \72        "default flag did not return False for unmodified address"73def test_address_set_default():74    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \75        TEST_TYPE_STRING)76    test_address.set_default_flag(True)77    assert(test_address.get_default_flag() == True), \78        "set default flag failed to modify flag to True for address without person"79def test_address_remove_default():80    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \81        TEST_TYPE_STRING)82    test_address.set_default_flag(True)83    test_address.set_default_flag(False)84    assert(test_address.get_default_flag() == False), \85        "set default flag failed to modify flag to False for address without person"86#**************************************************************************************************87#**  ADDRESS: STREET88#**************************************************************************************************89def test_address_get_street():90    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \91        TEST_TYPE_STRING)92    assert(test_address.get_street() == TEST_STREET), \93        "get_street() did not return street name for address"94def test_address_modify_street():95    new_street_string = "321 New Test Street"96    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \97        TEST_TYPE_STRING)98    test_address.modify_street(new_street_string)99    assert(test_address.get_street() == new_street_string), \100        "modify_street() did not modify and return street name for address"101#**************************************************************************************************102#**  ADDRESS: APARTPEMNT NUMBER103#**************************************************************************************************104def test_address_get_apt():105    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \106        TEST_TYPE_STRING, apt_no = TEST_APT_NO)107    assert(test_address.get_apartment_no() == TEST_APT_NO), \108        "get_apartment_no() did not return apartment no. line for address"109def test_address_get_apt_none():110    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \111        TEST_TYPE_STRING)112    assert(test_address.get_apartment_no() == ""), \113        "get_apartment_no() did not return None for nonexistant apartment no. line for address"114def test_address_modify_apt():115    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \116        TEST_TYPE_STRING)117    test_address.modify_apartment_no(TEST_APT_NO)118    assert(test_address.get_apartment_no() == TEST_APT_NO), \119        "modify_apartment_no() did not modify and return apartment no. line for address"120#**************************************************************************************************121#**  ADDRESS: APARTPEMNT NUMBER122#**************************************************************************************************123def test_address_get_city():124    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \125        TEST_TYPE_STRING)126    assert(test_address.get_city() == TEST_CITY), "get_city() did not return city for address"127def test_address_modify_city():128    new_city_string = "New Test City"129    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \130        TEST_TYPE_STRING)131    test_address.modify_city(new_city_string)132    assert(test_address.get_city() == new_city_string), \133        "modify_city() did not modify and return city for address"134#**************************************************************************************************135#**  ADDRESS: ZIP_CODE136#**************************************************************************************************137def test_address_get_zip_code():138    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \139        TEST_TYPE_STRING)140    assert(test_address.get_zip_code() == TEST_ZIP_CODE), \141        "get_zip_code() did not return zipcode for address"142def test_address_modify_zip_code():143    new_zipcode = 99999144    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \145        TEST_TYPE_STRING)146    test_address.modify_zip_code(new_zipcode)147    assert(test_address.get_zip_code() == new_zipcode), \148        "modify_zip_code() did not modify and return zipcode for address"149def test_address_modify_zip_invalid_type():150    new_zipcode = "invalid string"151    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \152        TEST_TYPE_STRING)153    test_address.modify_zip_code(new_zipcode)154    assert(test_address.get_zip_code() == TEST_ZIP_CODE), \155        "modify_zip_code() did not appropriatly handle invalid zip code for address"156def test_address_modify_zip_invalid_number():157    new_zipcode = 999999158    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \159        TEST_TYPE_STRING)160    test_address.modify_zip_code(new_zipcode)161    assert(test_address.get_zip_code() == TEST_ZIP_CODE), \162        "modify_zip_code() did not appropriatly handle invalid numeric zip code for address"163#**************************************************************************************************164#**  ADDRESS: STATE165#**************************************************************************************************166def test_address_get_state():167    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \168        TEST_TYPE_STRING)169    assert(test_address.get_state() == TEST_STATE_CODE), \170        "get_state() did not return state for address"171def test_address_modify_state():172    new_state_code = "MN"173    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \174        TEST_TYPE_STRING)175    test_address.modify_state(new_state_code)176    assert(test_address.get_state() == new_state_code), \177        "modify_state() did not modify and return state for address"178def test_address_modify_state_invalid():179    new_state_code = "LL"180    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \181        TEST_TYPE_STRING)182    test_address.modify_state(new_state_code)183    assert(test_address.get_state() == TEST_STATE_CODE), \184        "modify_state() did not appropriatly handle invalid state code for address"185#**************************************************************************************************186#**  ADDRESS: TYPE187#**************************************************************************************************188def test_address_get_type():189    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \190        TEST_TYPE_STRING)191    assert(test_address.get_type() == TEST_TYPE_STRING), \192        "get_type() did not return type for address"193def test_address_modify_type_to_warehouse():194    new_type_string = "warehouse"195    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \196        TEST_TYPE_STRING)197    test_address.modify_type(new_type_string)198    assert(test_address.get_type() == new_type_string), \199        "modify_type() did not modify and return type for change to %s type address" \200        %(new_type_string)201def test_address_modify_type_to_billing():202    new_type_string = "billing"203    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \204        TEST_TYPE_STRING)205    test_address.modify_type(new_type_string)206    assert(test_address.get_type() == new_type_string), \207        "modify_type() did not modify and return type for change to %s type address" \208        %(new_type_string)209def test_address_modify_type_to_supplier():210    new_type_string = "supplier"211    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \212        TEST_TYPE_STRING)213    test_address.modify_type(new_type_string)214    assert(test_address.get_type() == new_type_string), \215        "modify_type() did not modify and return type for change to %s type address" \216        %(new_type_string)217def test_address_modify_type_invalid():218    new_type_string = "invalid string"219    test_address = address.Address.new_address(TEST_STREET, TEST_CITY, TEST_STATE_CODE, TEST_ZIP_CODE, \220        TEST_TYPE_STRING)221    test_address.modify_type(new_type_string)222    assert(test_address.get_type() == TEST_TYPE_STRING), \...

test_services.py

Source:test_services.py

1"""Test KNX services."""2import pytest3from xknx.telegram.apci import GroupValueResponse, GroupValueWrite4from homeassistant.const import STATE_OFF, STATE_ON5from homeassistant.core import HomeAssistant6from .conftest import KNXTestKit7from tests.common import async_capture_events8@pytest.mark.parametrize(9    "service_payload,expected_telegrams,expected_apci",10    [11        # send DPT 1 telegram12        (13            {"address": "1/2/3", "payload": True, "response": True},14            [("1/2/3", True)],15            GroupValueResponse,16        ),17        (18            {"address": "1/2/3", "payload": True, "response": False},19            [("1/2/3", True)],20            GroupValueWrite,21        ),22        # send DPT 5 telegram23        (24            {"address": "1/2/3", "payload": [99], "response": True},25            [("1/2/3", (99,))],26            GroupValueResponse,27        ),28        (29            {"address": "1/2/3", "payload": [99], "response": False},30            [("1/2/3", (99,))],31            GroupValueWrite,32        ),33        # send DPT 5 percent telegram34        (35            {"address": "1/2/3", "payload": 99, "type": "percent", "response": True},36            [("1/2/3", (0xFC,))],37            GroupValueResponse,38        ),39        (40            {"address": "1/2/3", "payload": 99, "type": "percent", "response": False},41            [("1/2/3", (0xFC,))],42            GroupValueWrite,43        ),44        # send temperature DPT 9 telegram45        (46            {47                "address": "1/2/3",48                "payload": 21.0,49                "type": "temperature",50                "response": True,51            },52            [("1/2/3", (0x0C, 0x1A))],53            GroupValueResponse,54        ),55        (56            {57                "address": "1/2/3",58                "payload": 21.0,59                "type": "temperature",60                "response": False,61            },62            [("1/2/3", (0x0C, 0x1A))],63            GroupValueWrite,64        ),65        # send multiple telegrams66        (67            {68                "address": ["1/2/3", "2/2/2", "3/3/3"],69                "payload": 99,70                "type": "percent",71                "response": True,72            },73            [74                ("1/2/3", (0xFC,)),75                ("2/2/2", (0xFC,)),76                ("3/3/3", (0xFC,)),77            ],78            GroupValueResponse,79        ),80        (81            {82                "address": ["1/2/3", "2/2/2", "3/3/3"],83                "payload": 99,84                "type": "percent",85                "response": False,86            },87            [88                ("1/2/3", (0xFC,)),89                ("2/2/2", (0xFC,)),90                ("3/3/3", (0xFC,)),91            ],92            GroupValueWrite,93        ),94    ],95)96async def test_send(97    hass: HomeAssistant,98    knx: KNXTestKit,99    service_payload,100    expected_telegrams,101    expected_apci,102):103    """Test `knx.send` service."""104    await knx.setup_integration({})105    await hass.services.async_call(106        "knx",107        "send",108        service_payload,109        blocking=True,110    )111    for expected_response in expected_telegrams:112        group_address, payload = expected_response113        await knx.assert_telegram(group_address, payload, expected_apci)114async def test_read(hass: HomeAssistant, knx: KNXTestKit):115    """Test `knx.read` service."""116    await knx.setup_integration({})117    # send read telegram118    await hass.services.async_call("knx", "read", {"address": "1/1/1"}, blocking=True)119    await knx.assert_read("1/1/1")120    # send multiple read telegrams121    await hass.services.async_call(122        "knx",123        "read",124        {"address": ["1/1/1", "2/2/2", "3/3/3"]},125        blocking=True,126    )127    await knx.assert_read("1/1/1")128    await knx.assert_read("2/2/2")129    await knx.assert_read("3/3/3")130async def test_event_register(hass: HomeAssistant, knx: KNXTestKit):131    """Test `knx.event_register` service."""132    events = async_capture_events(hass, "knx_event")133    test_address = "1/2/3"134    await knx.setup_integration({})135    # no event registered136    await knx.receive_write(test_address, True)137    await hass.async_block_till_done()138    assert len(events) == 0139    # register event with `type`140    await hass.services.async_call(141        "knx",142        "event_register",143        {"address": test_address, "type": "2byte_unsigned"},144        blocking=True,145    )146    await knx.receive_write(test_address, (0x04, 0xD2))147    await hass.async_block_till_done()148    assert len(events) == 1149    typed_event = events.pop()150    assert typed_event.data["data"] == (0x04, 0xD2)151    assert typed_event.data["value"] == 1234152    # remove event registration - no event added153    await hass.services.async_call(154        "knx",155        "event_register",156        {"address": test_address, "remove": True},157        blocking=True,158    )159    await knx.receive_write(test_address, True)160    await hass.async_block_till_done()161    assert len(events) == 0162    # register event without `type`163    await hass.services.async_call(164        "knx", "event_register", {"address": test_address}, blocking=True165    )166    await knx.receive_write(test_address, True)167    await knx.receive_write(test_address, False)168    await hass.async_block_till_done()169    assert len(events) == 2170    untyped_event_2 = events.pop()171    assert untyped_event_2.data["data"] is False172    assert untyped_event_2.data["value"] is None173    untyped_event_1 = events.pop()174    assert untyped_event_1.data["data"] is True175    assert untyped_event_1.data["value"] is None176async def test_exposure_register(hass: HomeAssistant, knx: KNXTestKit):177    """Test `knx.exposure_register` service."""178    test_address = "1/2/3"179    test_entity = "fake.entity"180    test_attribute = "fake_attribute"181    await knx.setup_integration({})182    # no exposure registered183    hass.states.async_set(test_entity, STATE_ON, {})184    await knx.assert_no_telegram()185    # register exposure186    await hass.services.async_call(187        "knx",188        "exposure_register",189        {"address": test_address, "entity_id": test_entity, "type": "binary"},190        blocking=True,191    )192    hass.states.async_set(test_entity, STATE_OFF, {})193    await knx.assert_write(test_address, False)194    # register exposure195    await hass.services.async_call(196        "knx",197        "exposure_register",198        {"address": test_address, "remove": True},199        blocking=True,200    )201    hass.states.async_set(test_entity, STATE_ON, {})202    await knx.assert_no_telegram()203    # register exposure for attribute with default204    await hass.services.async_call(205        "knx",206        "exposure_register",207        {208            "address": test_address,209            "entity_id": test_entity,210            "attribute": test_attribute,211            "type": "percentU8",212            "default": 0,213        },214        blocking=True,215    )216    # no attribute on first change wouldn't work because no attribute change since last test217    hass.states.async_set(test_entity, STATE_ON, {test_attribute: 30})218    await knx.assert_write(test_address, (30,))219    hass.states.async_set(test_entity, STATE_OFF, {})220    await knx.assert_write(test_address, (0,))221    # don't send same value sequentially222    hass.states.async_set(test_entity, STATE_ON, {test_attribute: 25})223    hass.states.async_set(test_entity, STATE_ON, {test_attribute: 25})224    hass.states.async_set(test_entity, STATE_ON, {test_attribute: 25, "unrelated": 2})225    hass.states.async_set(test_entity, STATE_OFF, {test_attribute: 25})226    await knx.assert_telegram_count(1)...

test_utils.py

Source:test_utils.py

...48        except ValueError:49            pass50        else:51            self.fail("Nonsense test name should throw ValueError")52    def test_test_address(self):53        # test addresses are specified as54        #     package.module:class.method55        #     /path/to/file.py:class.method56        # converted into 3-tuples (file, module, callable)57        # all terms optional58        test_address = util.test_address59        absfile = util.absfile60        class Foo:61            def bar(self):62                pass63        def baz():64            pass65        f = Foo()66        class FooTC(unittest.TestCase):67            def test_one(self):68                pass69            def test_two(self):70                pass71        class CustomTestType(type):72            pass73        class CustomTC(unittest.TestCase):74            __metaclass__ = CustomTestType75            def test_one(self):76                pass77            def test_two(self):78                pass79        foo_funct = case.FunctionTestCase(baz)80        foo_functu = unittest.FunctionTestCase(baz)81        foo_mtc = case.MethodTestCase(unbound_method(Foo, Foo.bar))82        me = util.src(absfile(__file__))83        self.assertEqual(test_address(baz),84                         (me, __name__, 'baz'))85        assert test_address(Foo) == (me, __name__, 'Foo')86        assert test_address(unbound_method(Foo, Foo.bar)) == (me, __name__,87                                                              'Foo.bar')88        assert test_address(f) == (me, __name__, 'Foo')89        assert test_address(f.bar) == (me, __name__, 'Foo.bar')90        assert test_address(nose) == (91            util.src(absfile(nose.__file__)), 'nose', None)92        # test passing the actual test callable, as the93        # missed test plugin must do94        self.assertEqual(test_address(FooTC('test_one')),95                         (me, __name__, 'FooTC.test_one'))96        self.assertEqual(test_address(CustomTC('test_one')),97                         (me, __name__, 'CustomTC.test_one'))98        self.assertEqual(test_address(foo_funct),99                         (me, __name__, 'baz'))100        self.assertEqual(test_address(foo_functu),101                         (me, __name__, 'baz'))102        self.assertEqual(test_address(foo_mtc),103                         (me, __name__, 'Foo.bar'))104        # verify that we fail on an invalid input type105        self.assertRaises(TypeError, test_address, 1)106        self.assertRaises(TypeError, test_address, "foo")107    def test_isclass_detects_classes(self):108        class TC(unittest.TestCase):109            pass110        class TC_Classic:111            pass112        class TC_object(object):113            pass114        # issue153 -- was not detecting custom typed classes...115        class TCType(type):116            pass...

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.