Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.