KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
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 Slash 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?
