KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use replication_config method in localstack
Best Python code snippet using localstack_python
artifactory_replication.py
Source:artifactory_replication.py 
1#!/usr/bin/python2# This program is free software: you can redistribute it and/or modify3# it under the terms of the GNU General Public License as published by4# the Free Software Foundation, either version 3 of the License, or5# (at your option) any later version.6#7# This program is distributed in the hope that it will be useful,8# but WITHOUT ANY WARRANTY; without even the implied warranty of9# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the10# GNU General Public License for more details.11#12# You should have received a copy of the GNU General Public License13# along with this program. If not, see <http://www.gnu.org/licenses/>.14from __future__ import absolute_import, division, print_function15__metaclass__ = type16ANSIBLE_METADATA = {17 'metadata_version': '1.1',18 'status': ['preview'],19 'supported_by': 'community'20}21DOCUMENTATION = '''22---23module: artifactory_replication24short_description: Provides management repositoriy replication in JFrog Artifactory25version_added: "2.5"26description:27 - Provides basic management operations for configuration repository28 replication in JFrog Artifactory.29options:30 artifactory_url:31 description:32 - The target URL for managing artifactory. For certain operations,33 you can include the group name appended to the end of the34 url.35 required: true36 name:37 description:38 - Name of the local repository to configure replication against.39 required: true40 replication_config:41 description:42 - The string representations of the JSON used to configure the43 replication for the given repository.44 username:45 description:46 - username to be used in Basic Auth against Artifactory. Not47 required if using auth_token for basic auth.48 auth_password:49 description:50 - password to be used in Basic Auth against Artifactory. Not51 required if using auth_token for basic auth.52 auth_token:53 description:54 - authentication token to be used in Basic Auth against55 Artifactory. Not required if using username/auth_password for56 basic auth.57 validate_certs:58 description:59 - True to validate SSL certificates, False otherwise.60 type: bool61 default: false62 client_cert:63 description:64 - PEM formatted certificate chain file to be used for SSL client65 authentication. This file can also include the key as well, and66 if the key is included, I(client_key) is not required67 client_key:68 description:69 - PEM formatted file that contains your private key to be used for70 SSL client authentication. If I(client_cert) contains both the71 certificate and key, this option is not required.72 force_basic_auth:73 description:74 - The library used by the uri module only sends authentication75 information when a webservice responds to an initial request76 with a 401 status. Since some basic auth services do not properly77 send a 401, logins will fail. This option forces the sending of78 the Basic authentication header upon initial request.79 type: bool80 default: false81 state:82 description:83 - The state the replication configuration should be in.84 'present' ensures that the target exists, but is not replaced.85 The configuration supplied will overwrite the configuration that86 exists. 'absent' ensures that the the target is deleted.87 'read' will return the configuration if the target exists.88 choices:89 - present90 - absent91 - read92 default: read93 replication_config_dict:94 description:95 - A dictionary in yaml format of valid configuration for repository96 replication. These dictionary values must match any other values97 passed in, such as those within top-level parameters or within98 the configuration string in replication_config.99 cronExp:100 description:101 - A cron expression that represents the schedule that push/pull102 replication will take place. This is independent of event based103 replication.104 enableEventReplication:105 description:106 - Enable event based replication whenever a repository is created,107 updated, or deleted..108 type: bool109 default: false110 remote_url:111 description:112 - The url of the target repository to configure push or pull113 replication against.114 replication_username:115 description:116 - The username Artifactory will use to execute push/pull117 replication operations against a remote.118 replication_password:119 description:120 - The password Artifactory will use to execute push/pull121 replication operations against a remote.122author:123 - Kyle Haley (@quadewarren)124'''125EXAMPLES = '''126# Configure replication on an existing repo with cron and event based127# replication configured.128- name: Configure replication for a repository129 artifactory_replication:130 artifactory_url: http://art.url.com/artifactory/api/replications/131 auth_token: MY_TOKEN132 name: bower-local133 cronExp: "0 0 12 * * ?"134 enableEventReplication: true135 replication_username: remote_username136 replication_password: my_password137 remote_url: http://the.remote.repo.com/artifactory/remote-repo138 state: present139# Update the replication configuration, set event based replication to false140- name: Update the replication configuration for a repository141 artifactory_replication:142 artifactory_url: http://art.url.com/artifactory/api/replications/143 auth_token: MY_TOKEN144 name: bower-local145 enableEventReplication: false146 state: present147- name: delete the replication configuration148 artifactory_replication:149 artifactory_url: http://art.url.com/artifactory/api/replications/150 auth_token: MY_TOKEN151 name: bower-local152 state: absent153'''154RETURN = '''155original_message:156 description:157 - A brief sentence describing what action the module was attempting158 to make against the replication configuration and what159 artifactory url.160 returned: success161 type: str162message:163 description: The result of the attempted action.164 returned: success165 type: str166config:167 description:168 - The configuration of a successfully created replication config,169 an updated replication config (whether or not changed=True), or170 the config of a replication config that was successfully deleted.171 returned: success172 type: dict173'''174import ast175import ansible.module_utils.artifactory as art_base176from ansible.module_utils.basic import AnsibleModule177REPLICATION_CONFIG_MAP = {178 "name":179 {"always_required": True},180 "remote_url":181 {"always_required": True}}182URI_CONFIG_MAP = {"api/replications": REPLICATION_CONFIG_MAP}183def main():184 state_map = ['present', 'absent', 'read', 'list']185 module_args = dict(186 artifactory_url=dict(type='str', required=True),187 name=dict(type='str', required=True),188 replication_config=dict(type='str', default=None),189 username=dict(type='str', default=None),190 auth_password=dict(type='str', no_log=True, default=None),191 auth_token=dict(type='str', no_log=True, default=None),192 validate_certs=dict(type='bool', default=False),193 client_cert=dict(type='path', default=None),194 client_key=dict(type='path', default=None),195 force_basic_auth=dict(type='bool', default=False),196 state=dict(type='str', default='read', choices=state_map),197 replication_config_dict=dict(type='dict', default=dict()),198 remote_url=dict(type='str', default=None),199 replication_username=dict(type='str', default=None),200 replication_password=dict(type='str', default=None),201 cronExp=dict(type='str', default=None),202 enableEventReplication=dict(type='str', default=None),203 )204 result = dict(205 changed=False,206 original_message='',207 message='',208 config=dict()209 )210 module = AnsibleModule(211 argument_spec=module_args,212 required_together=[['username', 'auth_password']],213 required_one_of=[['auth_password', 'auth_token']],214 mutually_exclusive=[['auth_password', 'auth_token']],215 required_if=[['state', 'present',216 ['artifactory_url', 'name',217 'remote_url', 'replication_username',218 'replication_password']],219 ['state', 'absent', ['artifactory_url', 'name']],220 ['state', 'read', ['artifactory_url', 'name']]],221 supports_check_mode=True,222 )223 artifactory_url = module.params['artifactory_url']224 name = module.params['name']225 replication_config = module.params['replication_config']226 username = module.params['username']227 auth_password = module.params['auth_password']228 auth_token = module.params['auth_token']229 validate_certs = module.params['validate_certs']230 client_cert = module.params['client_cert']231 client_key = module.params['client_key']232 force_basic_auth = module.params['force_basic_auth']233 state = module.params['state']234 replication_config_dict = module.params['replication_config_dict']235 if replication_config:236 # temporarily convert to dict for validation237 replication_config = ast.literal_eval(replication_config)238 fail_messages = []239 fails = art_base.validate_config_params(replication_config,240 replication_config_dict,241 'replication_config',242 'replication_config_dict')243 fail_messages.extend(fails)244 fails = art_base.validate_top_level_params('name', module,245 replication_config,246 replication_config_dict,247 'replication_config',248 'replication_config_dict')249 fail_messages.extend(fails)250 fails = art_base.validate_top_level_params('remote_url', module,251 replication_config,252 replication_config_dict,253 'replication_config',254 'replication_config_dict')255 fail_messages.extend(fails)256 fails = art_base.validate_top_level_params('replication_username', module,257 replication_config,258 replication_config_dict,259 'replication_config',260 'replication_config_dict')261 fail_messages.extend(fails)262 fails = art_base.validate_top_level_params('replication_password', module,263 replication_config,264 replication_config_dict,265 'replication_config',266 'replication_config_dict')267 fail_messages.extend(fails)268 fails = art_base.validate_top_level_params('cronExp', module,269 replication_config,270 replication_config_dict,271 'replication_config',272 'replication_config_dict')273 fail_messages.extend(fails)274 fails = art_base.validate_top_level_params('enableEventReplication',275 module,276 replication_config,277 replication_config_dict,278 'replication_config',279 'replication_config_dict')280 fail_messages.extend(fails)281 # Populate failure messages282 failure_message = "".join(fail_messages)283 # Conflicting config values should not be resolved284 if failure_message:285 module.fail_json(msg=failure_message, **result)286 sec_dict = dict()287 if module.params['name']:288 sec_dict['name'] = module.params['name']289 if module.params['remote_url']:290 sec_dict['remote_url'] = module.params['remote_url']291 if module.params['email']:292 sec_dict['email'] = module.params['email']293 if module.params['replication_username']:294 sec_dict['replication_username'] =\295 module.params['replication_username']296 if module.params['replication_password']:297 sec_dict['replication_password'] =\298 module.params['replication_password']299 if module.params['cronExp']:300 sec_dict['cronExp'] = module.params['cronExp']301 if module.params['enableEventReplication']:302 sec_dict['enableEventReplication'] =\303 module.params['enableEventReplication']304 if replication_config:305 sec_dict.update(replication_config)306 if replication_config_dict:307 sec_dict.update(replication_config_dict)308 replication_config = str(sec_dict)309 result['original_message'] = ("Perform state '%s' against target '%s' "310 "within artifactory '%s'"311 % (state, name, artifactory_url))312 art_replication = art_base.ArtifactoryBase(313 artifactory_url=artifactory_url,314 name=name,315 art_config=replication_config,316 username=username,317 password=auth_password,318 auth_token=auth_token,319 validate_certs=validate_certs,320 client_cert=client_cert,321 client_key=client_key,322 force_basic_auth=force_basic_auth,323 config_map=URI_CONFIG_MAP)324 art_base.run_module(module, art_replication, "Repository replication",325 result, fail_messages, replication_config)326if __name__ == "__main__":...
endpoint.py
Source:endpoint.py
...62 def name(self, value: Optional[pulumi.Input[str]]):63 pulumi.set(self, "name", value)64 @property65 @pulumi.getter(name="replicationConfig")66 def replication_config(self) -> Optional[pulumi.Input['EndpointReplicationConfigArgs']]:67 return pulumi.get(self, "replication_config")68 @replication_config.setter69 def replication_config(self, value: Optional[pulumi.Input['EndpointReplicationConfigArgs']]):70 pulumi.set(self, "replication_config", value)71 @property72 @pulumi.getter(name="roleArn")73 def role_arn(self) -> Optional[pulumi.Input[str]]:74 return pulumi.get(self, "role_arn")75 @role_arn.setter76 def role_arn(self, value: Optional[pulumi.Input[str]]):77 pulumi.set(self, "role_arn", value)78class Endpoint(pulumi.CustomResource):79 @overload80 def __init__(__self__,81 resource_name: str,82 opts: Optional[pulumi.ResourceOptions] = None,83 description: Optional[pulumi.Input[str]] = None,84 event_buses: Optional[pulumi.Input[Sequence[pulumi.Input[pulumi.InputType['EndpointEventBusArgs']]]]] = None,85 name: Optional[pulumi.Input[str]] = None,86 replication_config: Optional[pulumi.Input[pulumi.InputType['EndpointReplicationConfigArgs']]] = None,87 role_arn: Optional[pulumi.Input[str]] = None,88 routing_config: Optional[pulumi.Input[pulumi.InputType['EndpointRoutingConfigArgs']]] = None,89 __props__=None):90 """91 Resource Type definition for AWS::Events::Endpoint.92 :param str resource_name: The name of the resource.93 :param pulumi.ResourceOptions opts: Options for the resource.94 """95 ...96 @overload97 def __init__(__self__,98 resource_name: str,99 args: EndpointArgs,100 opts: Optional[pulumi.ResourceOptions] = None):101 """102 Resource Type definition for AWS::Events::Endpoint.103 :param str resource_name: The name of the resource.104 :param EndpointArgs args: The arguments to use to populate this resource's properties.105 :param pulumi.ResourceOptions opts: Options for the resource.106 """107 ...108 def __init__(__self__, resource_name: str, *args, **kwargs):109 resource_args, opts = _utilities.get_resource_args_opts(EndpointArgs, pulumi.ResourceOptions, *args, **kwargs)110 if resource_args is not None:111 __self__._internal_init(resource_name, opts, **resource_args.__dict__)112 else:113 __self__._internal_init(resource_name, *args, **kwargs)114 def _internal_init(__self__,115 resource_name: str,116 opts: Optional[pulumi.ResourceOptions] = None,117 description: Optional[pulumi.Input[str]] = None,118 event_buses: Optional[pulumi.Input[Sequence[pulumi.Input[pulumi.InputType['EndpointEventBusArgs']]]]] = None,119 name: Optional[pulumi.Input[str]] = None,120 replication_config: Optional[pulumi.Input[pulumi.InputType['EndpointReplicationConfigArgs']]] = None,121 role_arn: Optional[pulumi.Input[str]] = None,122 routing_config: Optional[pulumi.Input[pulumi.InputType['EndpointRoutingConfigArgs']]] = None,123 __props__=None):124 opts = pulumi.ResourceOptions.merge(_utilities.get_resource_opts_defaults(), opts)125 if not isinstance(opts, pulumi.ResourceOptions):126 raise TypeError('Expected resource options to be a ResourceOptions instance')127 if opts.id is None:128 if __props__ is not None:129 raise TypeError('__props__ is only valid when passed in combination with a valid opts.id to get an existing resource')130 __props__ = EndpointArgs.__new__(EndpointArgs)131 __props__.__dict__["description"] = description132 if event_buses is None and not opts.urn:133 raise TypeError("Missing required property 'event_buses'")134 __props__.__dict__["event_buses"] = event_buses135 __props__.__dict__["name"] = name136 __props__.__dict__["replication_config"] = replication_config137 __props__.__dict__["role_arn"] = role_arn138 if routing_config is None and not opts.urn:139 raise TypeError("Missing required property 'routing_config'")140 __props__.__dict__["routing_config"] = routing_config141 __props__.__dict__["arn"] = None142 __props__.__dict__["endpoint_id"] = None143 __props__.__dict__["endpoint_url"] = None144 __props__.__dict__["state"] = None145 __props__.__dict__["state_reason"] = None146 super(Endpoint, __self__).__init__(147 'aws-native:events:Endpoint',148 resource_name,149 __props__,150 opts)151 @staticmethod152 def get(resource_name: str,153 id: pulumi.Input[str],154 opts: Optional[pulumi.ResourceOptions] = None) -> 'Endpoint':155 """156 Get an existing Endpoint resource's state with the given name, id, and optional extra157 properties used to qualify the lookup.158 :param str resource_name: The unique name of the resulting resource.159 :param pulumi.Input[str] id: The unique provider ID of the resource to lookup.160 :param pulumi.ResourceOptions opts: Options for the resource.161 """162 opts = pulumi.ResourceOptions.merge(opts, pulumi.ResourceOptions(id=id))163 __props__ = EndpointArgs.__new__(EndpointArgs)164 __props__.__dict__["arn"] = None165 __props__.__dict__["description"] = None166 __props__.__dict__["endpoint_id"] = None167 __props__.__dict__["endpoint_url"] = None168 __props__.__dict__["event_buses"] = None169 __props__.__dict__["name"] = None170 __props__.__dict__["replication_config"] = None171 __props__.__dict__["role_arn"] = None172 __props__.__dict__["routing_config"] = None173 __props__.__dict__["state"] = None174 __props__.__dict__["state_reason"] = None175 return Endpoint(resource_name, opts=opts, __props__=__props__)176 @property177 @pulumi.getter178 def arn(self) -> pulumi.Output[str]:179 return pulumi.get(self, "arn")180 @property181 @pulumi.getter182 def description(self) -> pulumi.Output[Optional[str]]:183 return pulumi.get(self, "description")184 @property185 @pulumi.getter(name="endpointId")186 def endpoint_id(self) -> pulumi.Output[str]:187 return pulumi.get(self, "endpoint_id")188 @property189 @pulumi.getter(name="endpointUrl")190 def endpoint_url(self) -> pulumi.Output[str]:191 return pulumi.get(self, "endpoint_url")192 @property193 @pulumi.getter(name="eventBuses")194 def event_buses(self) -> pulumi.Output[Sequence['outputs.EndpointEventBus']]:195 return pulumi.get(self, "event_buses")196 @property197 @pulumi.getter198 def name(self) -> pulumi.Output[str]:199 return pulumi.get(self, "name")200 @property201 @pulumi.getter(name="replicationConfig")202 def replication_config(self) -> pulumi.Output[Optional['outputs.EndpointReplicationConfig']]:203 return pulumi.get(self, "replication_config")204 @property205 @pulumi.getter(name="roleArn")206 def role_arn(self) -> pulumi.Output[Optional[str]]:207 return pulumi.get(self, "role_arn")208 @property209 @pulumi.getter(name="routingConfig")210 def routing_config(self) -> pulumi.Output['outputs.EndpointRoutingConfig']:211 return pulumi.get(self, "routing_config")212 @property213 @pulumi.getter214 def state(self) -> pulumi.Output['EndpointState']:215 return pulumi.get(self, "state")216 @property...
get_endpoint.py
Source:get_endpoint.py
...69 def event_buses(self) -> Optional[Sequence['outputs.EndpointEventBus']]:70 return pulumi.get(self, "event_buses")71 @property72 @pulumi.getter(name="replicationConfig")73 def replication_config(self) -> Optional['outputs.EndpointReplicationConfig']:74 return pulumi.get(self, "replication_config")75 @property76 @pulumi.getter(name="roleArn")77 def role_arn(self) -> Optional[str]:78 return pulumi.get(self, "role_arn")79 @property80 @pulumi.getter(name="routingConfig")81 def routing_config(self) -> Optional['outputs.EndpointRoutingConfig']:82 return pulumi.get(self, "routing_config")83 @property84 @pulumi.getter85 def state(self) -> Optional['EndpointState']:86 return pulumi.get(self, "state")87 @property...
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 localstack 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?
