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

Copied to Clipboard!

How to use is_error method in Slash

Best Python code snippet using slash

test_group_calls_function.py

Source:test_group_calls_function.py

1import unittest2from smscountrysdk.group_calls import GroupCallsApi3from urllib2 import URLError4import json5import datetime6import random, string7class GroupCallsTestCase(unittest.TestCase):8    def setUp(self):9        self.authKey = "q0VA5j2G8v1RpchQqVfs"10        self.authToken = "MHpmHoHxDGPbBij3SZSC7OGYHHHrRX507G6vqgrX"11        self.api = GroupCallsApi(self.authKey, self.authToken)12        self.name_group = ''.join(random.choice(13            string.lowercase) for i in range(10))14        self.lst_participants = [15            {"Name": "tiendang", "Number": "+84905708052"},16            {"Name": "hoangvo", "Number": "+84985022669"},17            {"Name": "hoangvo12", "Number": " 1234567890 "}]18        self.end = datetime.datetime.now()19        self.start = self.end + datetime.timedelta(days=-1)20        self.number = "+84905%s" % random.randint(100000, 999999)21        self.groupCallUUID = "68769E1C6D934B9B8F35711D77769E"22        self.participantId = ""23        self.fileFormat = "mp3"24        self.recordingUUID = ""25    def test_create_group_call(self):26        print "====Begin Call Test Function Create New Group Call====\n"27        is_error = False28        try:29            resp = self.api.create_group_call(Name=self.name_group, Participants=self.lst_participants, WelcomeSound="http://yourdomain/welcomsoundurl", WaitSound="http://yourdomain/waitsound",30                                              StartGropCallOnEnter="+84905708052", EndGroupCallOnExit="+84985022669", AnswerUrl="http://yourdomain/answer")31            data_json = json.loads(resp)32            print 'response : ', data_json33            self.assertEqual(data_json["Success"], 'True')34            # self.assertEqual(data_json["Message"], "Group Created")35        except Exception, e:36            print e37            is_error = True38        self.assertFalse(is_error, "Call API Create New Group Call Errors")39        print "====End Test Function Create New Group Call Success====\n"40    def test_get_groupcalls(self):41        print "====Begin Call Test Function Get Group Calls====\n"42        is_error = False43        try:44            resp = self.api.get_groupcalls(FromDate=None, ToDate=None, Offset=None, Limit=10)45            data_json = json.loads(resp)46            print 'response : ', data_json47            self.assertEqual(data_json["Success"], 'True')48            # self.assertEqual(data_json["Message"], "Group Created")49        except Exception, e:50            print e51            is_error = True52        self.assertFalse(is_error, "Call API Get Group Calls Errors")53        print "====End Test Function Get Group Calls Success====\n"54    def test_get_groupcall_detail(self):55        print "====Begin Call Test Function Get Group Call Detail====\n"56        is_error = False57        try:58            self.assertNotEqual(self.groupCallUUID, "")59            resp = self.api.get_groupcall_detail(self.groupCallUUID)60            data_json = json.loads(resp)61            print 'response : ', data_json62            self.assertEqual(data_json["Success"], 'True')63            # self.assertEqual(data_json["Message"], "Group Created")64        except Exception, e:65            print e66            is_error = True67        self.assertFalse(is_error, "Call API Get Group Call Detail Errors")68        print "====End Test Function Get Group Call Detail Success====\n"69    def test_get_all_participant_from_groupcall(self):70        print "====Begin Call Test Function Get All Participant From GroupCalls====\n"71        is_error = False72        try:73            self.assertNotEqual(self.groupCallUUID, "")74            75            resp = self.api.get_all_participant_from_groupcall(self.groupCallUUID)76            data_json = json.loads(resp)77            print 'response : ', data_json78            self.assertEqual(data_json["Success"], 'True')79            # self.assertEqual(data_json["Message"], "Group Created")80        except Exception, e:81            print e82            is_error = True83        self.assertFalse(is_error, "Call API Get All Participant From GroupCalls Errors")84        print "====End Test Function Get All Participant From GroupCalls Success====\n"85    def test_get_participant_from_groupcall(self):86        print "====Begin Call Test Function Get Participant Detail From GroupCalls====\n"87        is_error = False88        try:89            self.assertNotEqual(self.groupCallUUID, "")90            self.assertNotEqual(self.participantId, "")91            92            resp = self.api.get_participant_from_groupcall(self.groupCallUUID, self.participantId)93            data_json = json.loads(resp)94            print 'response : ', data_json95            self.assertEqual(data_json["Success"], 'True')96            # self.assertEqual(data_json["Message"], "Group Created")97        except Exception, e:98            print e99            is_error = True100        self.assertFalse(is_error, "Call API Get Participant Detail From GroupCalls Errors")101        print "====End Test Function Get Participant Detail From GroupCalls Success====\n"102    def test_play_sound_into_groupcall(self):103        print "====Begin Call Test Function Play Sound Into GroupCalls====\n"104        is_error = False105        try:106            self.assertNotEqual(self.groupCallUUID, "")107            resp = self.api.play_sound_into_groupcall(self.groupCallUUID, FileUrl=None)108            data_json = json.loads(resp)109            print 'response : ', data_json110            self.assertEqual(data_json["Success"], 'True')111            # self.assertEqual(data_json["Message"], "Group Created")112        except Exception, e:113            print e114            is_error = True115        self.assertFalse(is_error, "Call API Play Sound Into GroupCalls Errors")116        print "====End Test Function Play Sound Into GroupCalls Success====\n"117    def test_play_sound_into_participant_groupcall(self):118        print "====Begin Call Test Function Play Sound Into Participant GroupCalls====\n"119        is_error = False120        try:121            self.assertNotEqual(self.groupCallUUID, "")122            self.assertNotEqual(self.participantId, "")123                        124            resp = self.api.play_sound_into_participant_groupcall(self.groupCallUUID, self.participantId, FileUrl=None)125            data_json = json.loads(resp)126            print 'response : ', data_json127            self.assertEqual(data_json["Success"], 'True')128            # self.assertEqual(data_json["Message"], "Group Created")129        except Exception, e:130            print e131            is_error = True132        self.assertFalse(is_error, "Call API Play Sound Into Participant GroupCalls Errors")133        print "====End Test Function Play Sound Into Participant GroupCalls Success====\n"134    def test_mute_all_participant_in_groupcall(self):135        print "====Begin Call Test Function Mute All Participant Call In GroupCalls====\n"136        is_error = False137        try:138            self.assertNotEqual(self.groupCallUUID, "")139                        140            resp = self.api.mute_all_participant_in_groupcall(self.groupCallUUID)141            data_json = json.loads(resp)142            print 'response : ', data_json143            self.assertEqual(data_json["Success"], 'True')144            # self.assertEqual(data_json["Message"], "Group Created")145        except Exception, e:146            print e147            is_error = True148        self.assertFalse(is_error, "Call API Mute All Participant Call In GroupCalls Errors")149        print "====End Test Function Mute All Participant Call In GroupCalls Success====\n"150    def test_mute_participant_in_groupcall(self):151        print "====Begin Call Test Function Mute Participant Call In GroupCalls====\n"152        is_error = False153        try:154            self.assertNotEqual(self.groupCallUUID, "")155            self.assertNotEqual(self.participantId, "")156                        157            resp = self.api.mute_participant_in_groupcall(GroupCallUUID=self.groupCallUUID, ParticipantId=self.participantId)158            data_json = json.loads(resp)159            print 'response : ', data_json160            self.assertEqual(data_json["Success"], 'True')161            # self.assertEqual(data_json["Message"], "Group Created")162        except Exception, e:163            print e164            is_error = True165        self.assertFalse(is_error, "Call API Mute Participant Call In GroupCalls Errors")166        print "====End Test Function Mute Participant Call In GroupCalls Success====\n"167    def test_unmute_all_participant_in_groupcall(self):168        print "====Begin Call Test Function UnMute All Participant Call In GroupCalls====\n"169        is_error = False170        try:171            self.assertNotEqual(self.groupCallUUID, "")172                        173            resp = self.api.unmute_all_participant_in_groupcall(GroupCallUUID=self.groupCallUUID)174            data_json = json.loads(resp)175            print 'response : ', data_json176            self.assertEqual(data_json["Success"], 'True')177            # self.assertEqual(data_json["Message"], "Group Created")178        except Exception, e:179            print e180            is_error = True181        self.assertFalse(is_error, "Call API UnMute All Participant Call In GroupCalls Errors")182        print "====End Test Function UnMute All Participant Call In GroupCalls Success====\n"183    def test_unmute_participant_in_groupcall(self):184        print "====Begin Call Test Function UnMute Participant Call In GroupCalls====\n"185        is_error = False186        try:187            self.assertNotEqual(self.groupCallUUID, "")188            self.assertNotEqual(self.participantId, "")189                        190            resp = self.api.unmute_participant_in_groupcall(GroupCallUUID=self.groupCallUUID, ParticipantId=self.participantId)191            data_json = json.loads(resp)192            print 'response : ', data_json193            self.assertEqual(data_json["Success"], 'True')194            # self.assertEqual(data_json["Message"], "Group Created")195        except Exception, e:196            print e197            is_error = True198        self.assertFalse(is_error, "Call API UnMute Participant Call In GroupCalls Errors")199        print "====End Test Function UnMute Participant Call In GroupCalls Success====\n"200    def test_start_recording_groupcall(self):201        print "====Begin Call Test Function Start Recording GroupCalls====\n"202        is_error = False203        try:204            self.assertNotEqual(self.groupCallUUID, "")205                        206            resp = self.api.start_recording_groupcall(GroupCallUUID=self.groupCallUUID, FileFormat=self.fileFormat)207            data_json = json.loads(resp)208            print 'response : ', data_json209            self.assertEqual(data_json["Success"], 'True')210            # self.assertEqual(data_json["Message"], "Group Created")211        except Exception, e:212            print e213            is_error = True214        self.assertFalse(is_error, "Call API Start Recording GroupCalls Errors")215        print "====End Test Function Start Recording GroupCalls Success====\n"216    def test_stop_all_recording_groupcall(self):217        print "====Begin Call Test Function Stop All Recording GroupCalls====\n"218        is_error = False219        try:220            self.assertNotEqual(self.groupCallUUID, "")221                        222            resp = self.api.stop_all_recording_groupcall(GroupCallUUID=self.groupCallUUID)223            data_json = json.loads(resp)224            print 'response : ', data_json225            self.assertEqual(data_json["Success"], 'True')226            # self.assertEqual(data_json["Message"], "Group Created")227        except Exception, e:228            print e229            is_error = True230        self.assertFalse(is_error, "Call API Stop All Recording GroupCalls Errors")231        print "====End Test Function Stop All Recording GroupCalls Success====\n"232    def test_get_all_recording_detail_of_groupcall(self):233        print "====Begin Call Test Function Get All Recording Details GroupCalls====\n"234        is_error = False235        try:236            self.assertNotEqual(self.groupCallUUID, "")237                        238            resp = self.api.get_all_recording_detail_of_groupcall(GroupCallUUID=self.groupCallUUID)239            data_json = json.loads(resp)240            print 'response : ', data_json241            self.assertEqual(data_json["Success"], 'True')242            # self.assertEqual(data_json["Message"], "Group Created")243        except Exception, e:244            print e245            is_error = True246        self.assertFalse(is_error, "Call API Get All Recording Details GroupCalls Errors")247        print "====End Test Function Get All Recording Details GroupCalls Success====\n"248    def test_get_recording_detail_of_groupcall(self):249        print "====Begin Call Test Function Get Recording Details GroupCalls====\n"250        is_error = False251        try:252            self.assertNotEqual(self.groupCallUUID, "")253            self.assertNotEqual(self.recordingUUID, "")254                        255            resp = self.api.get_recording_detail_of_groupcall(GroupCallUUID=self.groupCallUUID, RecordingUUID=slef.recordingUUID)256            data_json = json.loads(resp)257            print 'response : ', data_json258            self.assertEqual(data_json["Success"], 'True')259            # self.assertEqual(data_json["Message"], "Group Created")260        except Exception, e:261            print e262            is_error = True263        self.assertFalse(is_error, "Call API Get Recording Details GroupCalls Errors")264        print "====End Test Function Get Recording Details GroupCalls Success====\n"265    def test_stop_recording_groupcall(self):266        print "====Begin Call Test Function Stop Recording Details GroupCalls====\n"267        is_error = False268        try:269            self.assertNotEqual(self.groupCallUUID, "")270            self.assertNotEqual(self.recordingUUID, "")271                        272            resp = self.api.stop_recording_groupcall(GroupCallUUID=self.groupCallUUID, RecordingUUID=slef.recordingUUID)273            data_json = json.loads(resp)274            print 'response : ', data_json275            self.assertEqual(data_json["Success"], 'True')276            # self.assertEqual(data_json["Message"], "Group Created")277        except Exception, e:278            print e279            is_error = True280        self.assertFalse(is_error, "Call API Stop Recording Details GroupCalls Errors")281        print "====End Test Function Stop Recording Details GroupCalls Success====\n"282    def test_delete_recording_of_groupcall(self):283        print "====Begin Call Test Function Delete Recording Of GroupCalls====\n"284        is_error = False285        try:286            self.assertNotEqual(self.groupCallUUID, "")287            self.assertNotEqual(self.recordingUUID, "")288                        289            resp = self.api.delete_recording_of_groupcall(GroupCallUUID=self.groupCallUUID, RecordingUUID=slef.recordingUUID)290            data_json = json.loads(resp)291            print 'response : ', data_json292            self.assertEqual(data_json["Success"], 'True')293            # self.assertEqual(data_json["Message"], "Group Created")294        except Exception, e:295            print e296            is_error = True297        self.assertFalse(is_error, "Call API Delete Recording Of GroupCalls Errors")298        print "====End Test Function Delete Recording Of GroupCalls Success====\n"299    def test_delete_all_recording_of_groupcall(self):300        print "====Begin Call Test Function Delete All Recording Of GroupCalls====\n"301        is_error = False302        try:303            self.assertNotEqual(self.groupCallUUID, "")304                                    305            resp = self.api.delete_all_recording_of_groupcall(GroupCallUUID=self.groupCallUUID)306            data_json = json.loads(resp)307            print 'response : ', data_json308            self.assertEqual(data_json["Success"], 'True')309            # self.assertEqual(data_json["Message"], "Group Created")310        except Exception, e:311            print e312            is_error = True313        self.assertFalse(is_error, "Call API Delete All Recording Of GroupCalls Errors")314        print "====End Test Function Delete All Recording Of GroupCalls Success====\n"315    def test_disconnect_participants_from_groupcall(self):316        print "====Begin Call Test Function Disconnect Participant By Id From GroupCalls====\n"317        is_error = False318        try:319            self.assertNotEqual(self.groupCallUUID, "")320            self.assertNotEqual(self.participantId, "")321                                    322            resp = self.api.disconnect_participants_from_groupcall(GroupCallUUID=self.groupCallUUID, ParticipantId=self.participantId)323            data_json = json.loads(resp)324            print 'response : ', data_json325            self.assertEqual(data_json["Success"], 'True')326            # self.assertEqual(data_json["Message"], "Group Created")327        except Exception, e:328            print e329            is_error = True330        self.assertFalse(is_error, "Call API Disconnect Participant By Id From GroupCalls Errors")331        print "====End Test Function Disconnect Participant By Id From GroupCalls Success====\n"332    def test_disconnect_all_participants_from_groupcall(self):333        print "====Begin Call Test Function Disconnect All Participant From GroupCalls====\n"334        is_error = False335        try:336            self.assertNotEqual(self.groupCallUUID, "")337                                    338            resp = self.api.disconnect_all_participants_from_groupcall(GroupCallUUID=self.groupCallUUID)339            data_json = json.loads(resp)340            print 'response : ', data_json341            self.assertEqual(data_json["Success"], 'True')342            # self.assertEqual(data_json["Message"], "Group Created")343        except Exception, e:344            print e345            is_error = True346        self.assertFalse(is_error, "Call API Disconnect All Participants From GroupCalls Errors")...

dokku_storage.py

Source:dokku_storage.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3from ansible.module_utils.basic import AnsibleModule4from ansible.module_utils.dokku_utils import subprocess_check_output5import os6import pwd7import subprocess8DOCUMENTATION = """9---10module: dokku_storage11short_description: Manage storage for dokku applications12options:13  app:14    description:15      - The name of the app16    required: False17    default: null18    aliases: []19  create_host_dir:20    description:21      - Whether to create the host directory or not22    required: False23    default: False24    aliases: []25  group:26    description:27      - A group or gid that should own the created folder28    required: False29    default: 3276730    aliases: []31  mounts:32    description:33      - |34        A list of mounts to create, colon (:) delimited, in the format: `host_dir:container_dir`35    required: False36    default: []37    aliases: []38  user:39    description:40      - A user or uid that should own the created folder41    required: False42    default: 3276743    aliases: []44  state:45    description:46      - The state of the service link47    required: False48    default: present49    choices: [ "present", "absent" ]50    aliases: []51author: Jose Diaz-Gonzalez52requirements: [ ]53"""54EXAMPLES = """55- name: mount a path56  dokku_storage:57    app: hello-world58    mounts:59      - /var/lib/dokku/data/storage/hello-world:/data60- name: mount a path and create the host_dir directory61  dokku_storage:62    app: hello-world63    mounts:64      - /var/lib/dokku/data/storage/hello-world:/data65    create_host_dir: true66- name: unmount a path67  dokku_storage:68    app: hello-world69    mounts:70      - /var/lib/dokku/data/storage/hello-world:/data71    state: absent72- name: unmount a path and destroy the host_dir directory (and contents)73  dokku_storage:74    app: hello-world75    mounts:76      - /var/lib/dokku/data/storage/hello-world:/data77    destroy_host_dir: true78    state: absent79"""80def get_gid(group):81    gid = group82    try:83        gid = int(group)84    except ValueError:85        gid = pwd.getpwnam(group).pw_gid86    return gid87def get_state(b_path):88    """ Find out current state """89    if os.path.lexists(b_path):90        if os.path.islink(b_path):91            return "link"92        elif os.path.isdir(b_path):93            return "directory"94        elif os.stat(b_path).st_nlink > 1:95            return "hard"96        # could be many other things, but defaulting to file97        return "file"98    return "absent"99def get_uid(user):100    uid = user101    try:102        uid = int(user)103    except ValueError:104        uid = pwd.getpwnam(user).pw_uid105    return uid106def dokku_storage_list(data):107    command = "dokku --quiet storage:list {0}".format(data["app"])108    return subprocess_check_output(command)109def dokku_storage_mount_exists(data):110    state = get_state("/home/dokku/{0}".format(data["app"]))111    if state not in ["directory", "file"]:112        error = "app {0} does not exist".format(data["app"])113        return False, error114    output, error = dokku_storage_list(data)115    if error:116        return False, error117    mount = "{0}:{1}".format(data["host_dir"], data["container_dir"])118    if mount in output:119        return True, None120    return False, None121def dokku_storage_create_dir(data, is_error, has_changed, meta):122    if not data["create_host_dir"]:123        return (is_error, has_changed, meta)124    old_state = get_state(data["host_dir"])125    if old_state not in ["absent", "directory"]:126        is_error = True127        meta["error"] = "host directory is {0}".format(old_state)128        return (is_error, has_changed, meta)129    try:130        if old_state == "absent":131            os.makedirs(data["host_dir"], 0o777)132        os.chmod(data["host_dir"], 0o777)133        uid = get_uid(data["user"])134        gid = get_gid(data["group"])135        os.chown(data["host_dir"], uid, gid)136    except OSError as exc:137        is_error = True138        meta["error"] = str(exc)139        return (is_error, has_changed, meta)140    if old_state != get_state(data["host_dir"]):141        has_changed = True142    return (is_error, has_changed, meta)143def dokku_storage_destroy_dir(data, is_error, has_changed, meta):144    if not data["destroy_host_dir"]:145        return (is_error, has_changed, meta)146    old_state = get_state(data["host_dir"])147    if old_state not in ["absent", "directory"]:148        is_error = True149        meta["error"] = "host directory is {0}".format(old_state)150        return (is_error, has_changed, meta)151    try:152        if old_state == "directory":153            os.rmdir(data["host_dir"])154    except OSError as exc:155        is_error = True156        meta["error"] = str(exc)157        return (is_error, has_changed, meta)158    if old_state != get_state(data["host_dir"]):159        has_changed = True160    return (is_error, has_changed, meta)161def dokku_storage_absent(data):162    is_error = False163    has_changed = False164    meta = {"present": False}165    mounts = data.get("mounts", []) or []166    if len(mounts) == 0:167        is_error = True168        meta["error"] = "missing required arguments: mounts"169        return (is_error, has_changed, meta)170    for mount in mounts:171        data["host_dir"], data["container_dir"] = mount.split(":", 1)172        is_error, has_changed, meta = dokku_storage_destroy_dir(173            data, is_error, has_changed, meta174        )175        if is_error:176            return (is_error, has_changed, meta)177        exists, error = dokku_storage_mount_exists(data)178        if error:179            is_error = True180            meta["error"] = error181            return (is_error, has_changed, meta)182        elif not exists:183            is_error = False184            return (is_error, has_changed, meta)185        command = "dokku --quiet storage:unmount {0} {1}:{2}".format(186            data["app"], data["host_dir"], data["container_dir"]187        )188        try:189            subprocess.check_call(command, shell=True)190            is_error = False191            has_changed = True192        except subprocess.CalledProcessError as e:193            is_error = True194            meta["error"] = str(e)195            meta["present"] = True196        if is_error:197            return (is_error, has_changed, meta)198    return (is_error, has_changed, meta)199def dokku_storage_present(data):200    is_error = False201    has_changed = False202    meta = {"present": False}203    mounts = data.get("mounts", []) or []204    if len(mounts) == 0:205        is_error = True206        meta["error"] = "missing required arguments: mounts"207        return (is_error, has_changed, meta)208    for mount in mounts:209        data["host_dir"], data["container_dir"] = mount.split(":", 1)210        is_error, has_changed, meta = dokku_storage_create_dir(211            data, is_error, has_changed, meta212        )213        if is_error:214            return (is_error, has_changed, meta)215        exists, error = dokku_storage_mount_exists(data)216        if error:217            is_error = True218            meta["error"] = error219            return (is_error, has_changed, meta)220        elif exists:221            is_error = False222            return (is_error, has_changed, meta)223        command = "dokku --quiet storage:mount {0} {1}:{2}".format(224            data["app"], data["host_dir"], data["container_dir"]225        )226        try:227            subprocess.check_call(command, shell=True)228            is_error = False229            has_changed = True230            meta["present"] = True231        except subprocess.CalledProcessError as e:232            is_error = True233            meta["error"] = str(e)234        if is_error:235            return (is_error, has_changed, meta)236    return (is_error, has_changed, meta)237def main():238    fields = {239        "app": {"required": True, "type": "str"},240        "state": {241            "required": False,242            "default": "present",243            "choices": ["present", "absent"],244            "type": "str",245        },246        "mounts": {"required": False, "type": "list", "default": []},247        "create_host_dir": {"required": False, "default": False, "type": "bool"},248        "destroy_host_dir": {"required": False, "default": False, "type": "bool"},249        "user": {"required": False, "default": "32767", "type": "str"},250        "group": {"required": False, "default": "32767", "type": "str"},251    }252    choice_map = {253        "present": dokku_storage_present,254        "absent": dokku_storage_absent,255    }256    module = AnsibleModule(argument_spec=fields, supports_check_mode=False)257    is_error, has_changed, result = choice_map.get(module.params["state"])(258        module.params259    )260    if is_error:261        module.fail_json(msg=result["error"], meta=result)262    module.exit_json(changed=has_changed, meta=result)263if __name__ == "__main__":...

dokku_domains.py

Source:dokku_domains.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3from ansible.module_utils.basic import AnsibleModule4from ansible.module_utils.dokku_utils import subprocess_check_output5import pipes6import subprocess7DOCUMENTATION = """8---9module: dokku_domains10short_description: Manages domains for a given application11options:12  global:13    description:14      - Whether to change the global domains or app-specific domains.15    default: False16    aliases: []17  app:18    description:19      - The name of the app. This is required only if global is set to False.20    required: True21    default: null22    aliases: []23  domains:24    description:25      - A list of domains26    required: True27    default: null28    aliases: []29  state:30    description:31      - The state of the application's domains32    required: False33    default: present34    choices: [ "enable", "disable", "clear", "present", "absent" ]35    aliases: []36author: Jose Diaz-Gonzalez37requirements: [ ]38"""39EXAMPLES = """40- name: domains:add hello-world dokku.me41  dokku_domains:42    app: hello-world43    domains:44      - dokku.me45- name: domains:remove hello-world dokku.me46  dokku_domains:47    app: hello-world48    domains:49      - dokku.me50    state: absent51- name: domains:clear hello-world52  dokku_domains:53    app: hello-world54    state: clear55- name: domains:enable hello-world56  dokku_domains:57    app: hello-world58    state: enable59- name: domains:disable hello-world60  dokku_domains:61    app: hello-world62    state: disable63"""64def dokku_global_domains():65    command = "dokku --quiet domains --global --domains-global-vhosts"66    return subprocess_check_output(command)67def dokku_domains(data):68    if data["global"]:69        command = "dokku --quiet domains:report --global --domains-global-vhosts"70    else:71        command = "dokku --quiet domains:report {0} --domains-app-vhosts".format(72            data["app"]73        )74    return subprocess_check_output(command, split=" ")75def dokku_domains_disable(data):76    is_error = True77    has_changed = False78    meta = {"present": True}79    if data["global"]:80        is_error = True81        meta["error"] = '"disable" state cannot be used with global domains.'82        return (is_error, has_changed, meta)83    domains = dokku_domains(data)84    if "No domain names set for plugins" in domains:85        is_error = False86        meta["present"] = False87        return (is_error, has_changed, meta)88    command = "dokku --quiet domains:disable {0}".format(data["app"])89    try:90        subprocess.check_call(command, shell=True)91        is_error = False92        has_changed = True93        meta["present"] = False94    except subprocess.CalledProcessError as e:95        meta["error"] = str(e)96    return (is_error, has_changed, meta)97def dokku_domains_enable(data):98    is_error = True99    has_changed = False100    meta = {"present": False}101    if data["global"]:102        is_error = True103        meta["error"] = '"enable" state cannot be used with global domains.'104        return (is_error, has_changed, meta)105    domains = dokku_domains(data)106    if "No domain names set for plugins" not in domains:107        is_error = False108        meta["present"] = True109        return (is_error, has_changed, meta)110    command = "dokku --quiet domains:enable {0}".format(data["app"])111    try:112        subprocess.check_call(command, shell=True)113        is_error = False114        has_changed = True115        meta["present"] = True116    except subprocess.CalledProcessError as e:117        meta["error"] = str(e)118    return (is_error, has_changed, meta)119def dokku_domains_absent(data):120    is_error = True121    has_changed = False122    meta = {"present": True}123    existing, error = dokku_domains(data)124    if error:125        meta["error"] = error126        return (is_error, has_changed, meta)127    to_remove = [d for d in data["domains"] if d in existing]128    to_remove = [pipes.quote(d) for d in to_remove]129    if len(to_remove) == 0:130        is_error = False131        meta["present"] = False132        return (is_error, has_changed, meta)133    if data["global"]:134        command = "dokku --quiet domains:remove-global {0}".format(" ".join(to_remove))135    else:136        command = "dokku --quiet domains:remove {0} {1}".format(137            data["app"], " ".join(to_remove)138        )139    try:140        subprocess.check_call(command, shell=True)141        is_error = False142        has_changed = True143        meta["present"] = False144    except subprocess.CalledProcessError as e:145        meta["error"] = str(e)146    return (is_error, has_changed, meta)147def dokku_domains_present(data):148    is_error = True149    has_changed = False150    meta = {"present": False}151    existing, error = dokku_domains(data)152    if error:153        meta["error"] = error154        return (is_error, has_changed, meta)155    to_add = [d for d in data["domains"] if d not in existing]156    to_add = [pipes.quote(d) for d in to_add]157    if len(to_add) == 0:158        is_error = False159        meta["present"] = True160        return (is_error, has_changed, meta)161    if data["global"]:162        command = "dokku --quiet domains:add-global {0}".format(" ".join(to_add))163    else:164        command = "dokku --quiet domains:add {0} {1}".format(165            data["app"], " ".join(to_add)166        )167    try:168        subprocess.check_call(command, shell=True)169        is_error = False170        has_changed = True171        meta["present"] = True172    except subprocess.CalledProcessError as e:173        meta["error"] = str(e)174    return (is_error, has_changed, meta)175def dokku_domains_clear(data):176    is_error = True177    has_changed = False178    meta = {"present": False}179    if data["global"]:180        command = "dokku --quiet domains:clear-global"181    else:182        command = "dokku --quiet domains:clear {0}".format(data["app"])183    try:184        subprocess.check_call(command, shell=True)185        is_error = False186        has_changed = True187        meta["present"] = True188    except subprocess.CalledProcessError as e:189        meta["error"] = str(e)190    return (is_error, has_changed, meta)191def dokku_domains_set(data):192    is_error = True193    has_changed = False194    meta = {"present": False}195    existing, error = dokku_domains(data)196    if error:197        meta["error"] = error198        return (is_error, has_changed, meta)199    to_set = [pipes.quote(d) for d in data["domains"]]200    if data["global"]:201        command = "dokku --quiet domains:set-global {0}".format(" ".join(to_set))202    else:203        command = "dokku --quiet domains:set {0} {1}".format(204            data["app"], " ".join(to_set)205        )206    try:207        subprocess.check_call(command, shell=True)208        is_error = False209        has_changed = True210        meta["present"] = True211    except subprocess.CalledProcessError as e:212        meta["error"] = str(e)213    return (is_error, has_changed, meta)214def main():215    fields = {216        "global": {"required": False, "default": False, "type": "bool"},217        "app": {"required": False, "type": "str"},218        "domains": {"required": True, "type": "list"},219        "state": {220            "required": False,221            "default": "present",222            "choices": ["absent", "clear", "enable", "disable", "present", "set"],223            "type": "str",224        },225    }226    choice_map = {227        "absent": dokku_domains_absent,228        "clear": dokku_domains_clear,229        "disable": dokku_domains_disable,230        "enable": dokku_domains_enable,231        "present": dokku_domains_present,232        "set": dokku_domains_set,233    }234    module = AnsibleModule(argument_spec=fields, supports_check_mode=False)235    is_error, has_changed, result = choice_map.get(module.params["state"])(236        module.params237    )238    if is_error:239        module.fail_json(msg=result["error"], meta=result)240    module.exit_json(changed=has_changed, meta=result)241if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.