diff --git a/ibmcloudant/cloudant_v1.py b/ibmcloudant/cloudant_v1.py index 0ebeb96f..0b53e2e3 100644 --- a/ibmcloudant/cloudant_v1.py +++ b/ibmcloudant/cloudant_v1.py @@ -5860,49 +5860,6 @@ def head_scheduler_job(self, return response - def post_replicate(self, - replication_document: 'ReplicationDocument', - **kwargs - ) -> DetailedResponse: - """ - Create or modify a replication operation. - - Requests, configures, or stops a replicate operation. - - :param ReplicationDocument replication_document: HTTP request body for - replication operations. - :param dict headers: A `dict` containing the request headers - :return: A `DetailedResponse` containing the result, headers and HTTP status code. - :rtype: DetailedResponse with `dict` result representing a `ReplicationResult` object - """ - - if replication_document is None: - raise ValueError('replication_document must be provided') - if isinstance(replication_document, ReplicationDocument): - replication_document = convert_model(replication_document) - headers = {} - sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, - service_version='V1', - operation_id='post_replicate') - headers.update(sdk_headers) - - data = json.dumps(replication_document) - headers['content-type'] = 'application/json' - - if 'headers' in kwargs: - headers.update(kwargs.get('headers')) - headers['Accept'] = 'application/json' - - url = '/_replicate' - request = self.prepare_request(method='POST', - url=url, - headers=headers, - data=data) - - response = self.send(request) - return response - - def delete_replication_document(self, doc_id: str, *, @@ -6066,7 +6023,7 @@ def put_replication_document(self, **kwargs ) -> DetailedResponse: """ - Start or update a replication. + Create or modify a replication using a replication document. Creates or modifies a document in the `_replicator` database to start a new replication or to edit an existing replication. @@ -14366,262 +14323,6 @@ def __ne__(self, other: 'ReplicationDocument') -> bool: """Return `true` when self and other are not equal, false otherwise.""" return not self == other -class ReplicationHistory(): - """ - Schema for replication history information. - - :attr int doc_write_failures: Number of document write failures. - :attr int docs_read: Number of documents read. - :attr int docs_written: Number of documents written to target. - :attr str end_last_seq: Last sequence number in changes stream. - :attr str end_time: Date/Time replication operation completed in RFC 2822 - format. - :attr int missing_checked: Number of missing documents checked. - :attr int missing_found: Number of missing documents found. - :attr str recorded_seq: Last recorded sequence number. - :attr str session_id: Session ID for this replication operation. - :attr str start_last_seq: First sequence number in changes stream. - :attr str start_time: Date/Time replication operation started in RFC 2822 - format. - """ - - def __init__(self, - doc_write_failures: int, - docs_read: int, - docs_written: int, - end_last_seq: str, - end_time: str, - missing_checked: int, - missing_found: int, - recorded_seq: str, - session_id: str, - start_last_seq: str, - start_time: str) -> None: - """ - Initialize a ReplicationHistory object. - - :param int doc_write_failures: Number of document write failures. - :param int docs_read: Number of documents read. - :param int docs_written: Number of documents written to target. - :param str end_last_seq: Last sequence number in changes stream. - :param str end_time: Date/Time replication operation completed in RFC 2822 - format. - :param int missing_checked: Number of missing documents checked. - :param int missing_found: Number of missing documents found. - :param str recorded_seq: Last recorded sequence number. - :param str session_id: Session ID for this replication operation. - :param str start_last_seq: First sequence number in changes stream. - :param str start_time: Date/Time replication operation started in RFC 2822 - format. - """ - self.doc_write_failures = doc_write_failures - self.docs_read = docs_read - self.docs_written = docs_written - self.end_last_seq = end_last_seq - self.end_time = end_time - self.missing_checked = missing_checked - self.missing_found = missing_found - self.recorded_seq = recorded_seq - self.session_id = session_id - self.start_last_seq = start_last_seq - self.start_time = start_time - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ReplicationHistory': - """Initialize a ReplicationHistory object from a json dictionary.""" - args = {} - if 'doc_write_failures' in _dict: - args['doc_write_failures'] = _dict.get('doc_write_failures') - else: - raise ValueError('Required property \'doc_write_failures\' not present in ReplicationHistory JSON') - if 'docs_read' in _dict: - args['docs_read'] = _dict.get('docs_read') - else: - raise ValueError('Required property \'docs_read\' not present in ReplicationHistory JSON') - if 'docs_written' in _dict: - args['docs_written'] = _dict.get('docs_written') - else: - raise ValueError('Required property \'docs_written\' not present in ReplicationHistory JSON') - if 'end_last_seq' in _dict: - args['end_last_seq'] = _dict.get('end_last_seq') - else: - raise ValueError('Required property \'end_last_seq\' not present in ReplicationHistory JSON') - if 'end_time' in _dict: - args['end_time'] = _dict.get('end_time') - else: - raise ValueError('Required property \'end_time\' not present in ReplicationHistory JSON') - if 'missing_checked' in _dict: - args['missing_checked'] = _dict.get('missing_checked') - else: - raise ValueError('Required property \'missing_checked\' not present in ReplicationHistory JSON') - if 'missing_found' in _dict: - args['missing_found'] = _dict.get('missing_found') - else: - raise ValueError('Required property \'missing_found\' not present in ReplicationHistory JSON') - if 'recorded_seq' in _dict: - args['recorded_seq'] = _dict.get('recorded_seq') - else: - raise ValueError('Required property \'recorded_seq\' not present in ReplicationHistory JSON') - if 'session_id' in _dict: - args['session_id'] = _dict.get('session_id') - else: - raise ValueError('Required property \'session_id\' not present in ReplicationHistory JSON') - if 'start_last_seq' in _dict: - args['start_last_seq'] = _dict.get('start_last_seq') - else: - raise ValueError('Required property \'start_last_seq\' not present in ReplicationHistory JSON') - if 'start_time' in _dict: - args['start_time'] = _dict.get('start_time') - else: - raise ValueError('Required property \'start_time\' not present in ReplicationHistory JSON') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ReplicationHistory object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'doc_write_failures') and self.doc_write_failures is not None: - _dict['doc_write_failures'] = self.doc_write_failures - if hasattr(self, 'docs_read') and self.docs_read is not None: - _dict['docs_read'] = self.docs_read - if hasattr(self, 'docs_written') and self.docs_written is not None: - _dict['docs_written'] = self.docs_written - if hasattr(self, 'end_last_seq') and self.end_last_seq is not None: - _dict['end_last_seq'] = self.end_last_seq - if hasattr(self, 'end_time') and self.end_time is not None: - _dict['end_time'] = self.end_time - if hasattr(self, 'missing_checked') and self.missing_checked is not None: - _dict['missing_checked'] = self.missing_checked - if hasattr(self, 'missing_found') and self.missing_found is not None: - _dict['missing_found'] = self.missing_found - if hasattr(self, 'recorded_seq') and self.recorded_seq is not None: - _dict['recorded_seq'] = self.recorded_seq - if hasattr(self, 'session_id') and self.session_id is not None: - _dict['session_id'] = self.session_id - if hasattr(self, 'start_last_seq') and self.start_last_seq is not None: - _dict['start_last_seq'] = self.start_last_seq - if hasattr(self, 'start_time') and self.start_time is not None: - _dict['start_time'] = self.start_time - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ReplicationHistory object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ReplicationHistory') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ReplicationHistory') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - -class ReplicationResult(): - """ - Schema for a replication result. - - :attr List[ReplicationHistory] history: Replication history. - :attr bool ok: Replication status. - :attr int replication_id_version: Replication protocol version. - :attr str session_id: Unique session ID. - :attr str source_last_seq: Last sequence number read from source database. - """ - - def __init__(self, - history: List['ReplicationHistory'], - ok: bool, - replication_id_version: int, - session_id: str, - source_last_seq: str) -> None: - """ - Initialize a ReplicationResult object. - - :param List[ReplicationHistory] history: Replication history. - :param bool ok: Replication status. - :param int replication_id_version: Replication protocol version. - :param str session_id: Unique session ID. - :param str source_last_seq: Last sequence number read from source database. - """ - self.history = history - self.ok = ok - self.replication_id_version = replication_id_version - self.session_id = session_id - self.source_last_seq = source_last_seq - - @classmethod - def from_dict(cls, _dict: Dict) -> 'ReplicationResult': - """Initialize a ReplicationResult object from a json dictionary.""" - args = {} - if 'history' in _dict: - args['history'] = [ReplicationHistory.from_dict(x) for x in _dict.get('history')] - else: - raise ValueError('Required property \'history\' not present in ReplicationResult JSON') - if 'ok' in _dict: - args['ok'] = _dict.get('ok') - else: - raise ValueError('Required property \'ok\' not present in ReplicationResult JSON') - if 'replication_id_version' in _dict: - args['replication_id_version'] = _dict.get('replication_id_version') - else: - raise ValueError('Required property \'replication_id_version\' not present in ReplicationResult JSON') - if 'session_id' in _dict: - args['session_id'] = _dict.get('session_id') - else: - raise ValueError('Required property \'session_id\' not present in ReplicationResult JSON') - if 'source_last_seq' in _dict: - args['source_last_seq'] = _dict.get('source_last_seq') - else: - raise ValueError('Required property \'source_last_seq\' not present in ReplicationResult JSON') - return cls(**args) - - @classmethod - def _from_dict(cls, _dict): - """Initialize a ReplicationResult object from a json dictionary.""" - return cls.from_dict(_dict) - - def to_dict(self) -> Dict: - """Return a json dictionary representing this model.""" - _dict = {} - if hasattr(self, 'history') and self.history is not None: - _dict['history'] = [x.to_dict() for x in self.history] - if hasattr(self, 'ok') and self.ok is not None: - _dict['ok'] = self.ok - if hasattr(self, 'replication_id_version') and self.replication_id_version is not None: - _dict['replication_id_version'] = self.replication_id_version - if hasattr(self, 'session_id') and self.session_id is not None: - _dict['session_id'] = self.session_id - if hasattr(self, 'source_last_seq') and self.source_last_seq is not None: - _dict['source_last_seq'] = self.source_last_seq - return _dict - - def _to_dict(self): - """Return a json dictionary representing this model.""" - return self.to_dict() - - def __str__(self) -> str: - """Return a `str` version of this ReplicationResult object.""" - return json.dumps(self.to_dict(), indent=2) - - def __eq__(self, other: 'ReplicationResult') -> bool: - """Return `true` when self and other are equal, false otherwise.""" - if not isinstance(other, self.__class__): - return False - return self.__dict__ == other.__dict__ - - def __ne__(self, other: 'ReplicationResult') -> bool: - """Return `true` when self and other are not equal, false otherwise.""" - return not self == other - class Revisions(): """ Schema for list of revision information. diff --git a/test/unit/test_cloudant_v1.py b/test/unit/test_cloudant_v1.py index 0bb18b4a..e36aedb0 100644 --- a/test/unit/test_cloudant_v1.py +++ b/test/unit/test_cloudant_v1.py @@ -8641,248 +8641,6 @@ def test_head_scheduler_job_value_error(self): -class TestPostReplicate(): - """ - Test Class for post_replicate - """ - - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - - @responses.activate - def test_post_replicate_all_params(self): - """ - post_replicate() - """ - # Set up mock - url = self.preprocess_url(base_url + '/_replicate') - mock_response = '{"history": [{"doc_write_failures": 0, "docs_read": 0, "docs_written": 0, "end_last_seq": "end_last_seq", "end_time": "end_time", "missing_checked": 0, "missing_found": 0, "recorded_seq": "recorded_seq", "session_id": "session_id", "start_last_seq": "start_last_seq", "start_time": "start_time"}], "ok": true, "replication_id_version": 0, "session_id": "session_id", "source_last_seq": "source_last_seq"}' - responses.add(responses.POST, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a Attachment model - attachment_model = {} - attachment_model['content_type'] = 'testString' - attachment_model['data'] = 'VGhpcyBpcyBhIG1vY2sgYnl0ZSBhcnJheSB2YWx1ZS4=' - attachment_model['digest'] = 'testString' - attachment_model['encoded_length'] = 0 - attachment_model['encoding'] = 'testString' - attachment_model['follows'] = True - attachment_model['length'] = 0 - attachment_model['revpos'] = 1 - attachment_model['stub'] = True - - # Construct a dict representation of a Revisions model - revisions_model = {} - revisions_model['ids'] = ['testString'] - revisions_model['start'] = 1 - - # Construct a dict representation of a DocumentRevisionStatus model - document_revision_status_model = {} - document_revision_status_model['rev'] = 'testString' - document_revision_status_model['status'] = 'available' - - # Construct a dict representation of a ReplicationCreateTargetParameters model - replication_create_target_parameters_model = {} - replication_create_target_parameters_model['n'] = 1 - replication_create_target_parameters_model['partitioned'] = True - replication_create_target_parameters_model['q'] = 1 - - # Construct a dict representation of a ReplicationDatabaseAuthIam model - replication_database_auth_iam_model = {} - replication_database_auth_iam_model['api_key'] = 'testString' - - # Construct a dict representation of a ReplicationDatabaseAuth model - replication_database_auth_model = {} - replication_database_auth_model['iam'] = replication_database_auth_iam_model - - # Construct a dict representation of a ReplicationDatabase model - replication_database_model = {} - replication_database_model['auth'] = replication_database_auth_model - replication_database_model['headers'] = {} - replication_database_model['url'] = 'testString' - - # Construct a dict representation of a UserContext model - user_context_model = {} - user_context_model['db'] = 'testString' - user_context_model['name'] = 'testString' - user_context_model['roles'] = ['_reader'] - - # Construct a dict representation of a ReplicationDocument model - replication_document_model = {} - replication_document_model['_attachments'] = {} - replication_document_model['_conflicts'] = ['testString'] - replication_document_model['_deleted'] = True - replication_document_model['_deleted_conflicts'] = ['testString'] - replication_document_model['_id'] = 'testString' - replication_document_model['_local_seq'] = 'testString' - replication_document_model['_rev'] = 'testString' - replication_document_model['_revisions'] = revisions_model - replication_document_model['_revs_info'] = [document_revision_status_model] - replication_document_model['cancel'] = True - replication_document_model['checkpoint_interval'] = 0 - replication_document_model['connection_timeout'] = 0 - replication_document_model['continuous'] = True - replication_document_model['create_target'] = True - replication_document_model['create_target_params'] = replication_create_target_parameters_model - replication_document_model['doc_ids'] = ['testString'] - replication_document_model['filter'] = 'testString' - replication_document_model['http_connections'] = 1 - replication_document_model['query_params'] = {} - replication_document_model['retries_per_request'] = 0 - replication_document_model['selector'] = {} - replication_document_model['since_seq'] = 'testString' - replication_document_model['socket_options'] = 'testString' - replication_document_model['source'] = replication_database_model - replication_document_model['source_proxy'] = 'testString' - replication_document_model['target'] = replication_database_model - replication_document_model['target_proxy'] = 'testString' - replication_document_model['use_checkpoints'] = True - replication_document_model['user_ctx'] = user_context_model - replication_document_model['worker_batch_size'] = 1 - replication_document_model['worker_processes'] = 1 - replication_document_model['foo'] = { 'foo': 'bar' } - - # Set up parameter values - replication_document = replication_document_model - - # Invoke method - response = service.post_replicate( - replication_document, - headers={} - ) - - # Check for correct operation - assert len(responses.calls) == 1 - assert response.status_code == 200 - # decompress gzip compressed request body - responses.calls[0].request.body = gzip.decompress(responses.calls[0].request.body) - - # Validate body params - req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) - assert req_body == replication_document - - - @responses.activate - def test_post_replicate_value_error(self): - """ - test_post_replicate_value_error() - """ - # Set up mock - url = self.preprocess_url(base_url + '/_replicate') - mock_response = '{"history": [{"doc_write_failures": 0, "docs_read": 0, "docs_written": 0, "end_last_seq": "end_last_seq", "end_time": "end_time", "missing_checked": 0, "missing_found": 0, "recorded_seq": "recorded_seq", "session_id": "session_id", "start_last_seq": "start_last_seq", "start_time": "start_time"}], "ok": true, "replication_id_version": 0, "session_id": "session_id", "source_last_seq": "source_last_seq"}' - responses.add(responses.POST, - url, - body=mock_response, - content_type='application/json', - status=200) - - # Construct a dict representation of a Attachment model - attachment_model = {} - attachment_model['content_type'] = 'testString' - attachment_model['data'] = 'VGhpcyBpcyBhIG1vY2sgYnl0ZSBhcnJheSB2YWx1ZS4=' - attachment_model['digest'] = 'testString' - attachment_model['encoded_length'] = 0 - attachment_model['encoding'] = 'testString' - attachment_model['follows'] = True - attachment_model['length'] = 0 - attachment_model['revpos'] = 1 - attachment_model['stub'] = True - - # Construct a dict representation of a Revisions model - revisions_model = {} - revisions_model['ids'] = ['testString'] - revisions_model['start'] = 1 - - # Construct a dict representation of a DocumentRevisionStatus model - document_revision_status_model = {} - document_revision_status_model['rev'] = 'testString' - document_revision_status_model['status'] = 'available' - - # Construct a dict representation of a ReplicationCreateTargetParameters model - replication_create_target_parameters_model = {} - replication_create_target_parameters_model['n'] = 1 - replication_create_target_parameters_model['partitioned'] = True - replication_create_target_parameters_model['q'] = 1 - - # Construct a dict representation of a ReplicationDatabaseAuthIam model - replication_database_auth_iam_model = {} - replication_database_auth_iam_model['api_key'] = 'testString' - - # Construct a dict representation of a ReplicationDatabaseAuth model - replication_database_auth_model = {} - replication_database_auth_model['iam'] = replication_database_auth_iam_model - - # Construct a dict representation of a ReplicationDatabase model - replication_database_model = {} - replication_database_model['auth'] = replication_database_auth_model - replication_database_model['headers'] = {} - replication_database_model['url'] = 'testString' - - # Construct a dict representation of a UserContext model - user_context_model = {} - user_context_model['db'] = 'testString' - user_context_model['name'] = 'testString' - user_context_model['roles'] = ['_reader'] - - # Construct a dict representation of a ReplicationDocument model - replication_document_model = {} - replication_document_model['_attachments'] = {} - replication_document_model['_conflicts'] = ['testString'] - replication_document_model['_deleted'] = True - replication_document_model['_deleted_conflicts'] = ['testString'] - replication_document_model['_id'] = 'testString' - replication_document_model['_local_seq'] = 'testString' - replication_document_model['_rev'] = 'testString' - replication_document_model['_revisions'] = revisions_model - replication_document_model['_revs_info'] = [document_revision_status_model] - replication_document_model['cancel'] = True - replication_document_model['checkpoint_interval'] = 0 - replication_document_model['connection_timeout'] = 0 - replication_document_model['continuous'] = True - replication_document_model['create_target'] = True - replication_document_model['create_target_params'] = replication_create_target_parameters_model - replication_document_model['doc_ids'] = ['testString'] - replication_document_model['filter'] = 'testString' - replication_document_model['http_connections'] = 1 - replication_document_model['query_params'] = {} - replication_document_model['retries_per_request'] = 0 - replication_document_model['selector'] = {} - replication_document_model['since_seq'] = 'testString' - replication_document_model['socket_options'] = 'testString' - replication_document_model['source'] = replication_database_model - replication_document_model['source_proxy'] = 'testString' - replication_document_model['target'] = replication_database_model - replication_document_model['target_proxy'] = 'testString' - replication_document_model['use_checkpoints'] = True - replication_document_model['user_ctx'] = user_context_model - replication_document_model['worker_batch_size'] = 1 - replication_document_model['worker_processes'] = 1 - replication_document_model['foo'] = { 'foo': 'bar' } - - # Set up parameter values - replication_document = replication_document_model - - # Pass in all but one required param and check for a ValueError - req_param_dict = { - "replication_document": replication_document, - } - for param in req_param_dict.keys(): - req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} - with pytest.raises(ValueError): - service.post_replicate(**req_copy) - - - class TestDeleteReplicationDocument(): """ Test Class for delete_replication_document @@ -9193,13 +8951,13 @@ def test_put_replication_document_all_params(self): replication_database_model = {} replication_database_model['auth'] = replication_database_auth_model replication_database_model['headers'] = {} - replication_database_model['url'] = 'https://examples.cloudant.com/animaldb' + replication_database_model['url'] = 'testString' # Construct a dict representation of a UserContext model user_context_model = {} user_context_model['db'] = 'testString' - user_context_model['name'] = 'john' - user_context_model['roles'] = ['researcher'] + user_context_model['name'] = 'testString' + user_context_model['roles'] = ['_reader'] # Construct a dict representation of a ReplicationDocument model replication_document_model = {} @@ -9212,28 +8970,28 @@ def test_put_replication_document_all_params(self): replication_document_model['_rev'] = 'testString' replication_document_model['_revisions'] = revisions_model replication_document_model['_revs_info'] = [document_revision_status_model] - replication_document_model['cancel'] = False - replication_document_model['checkpoint_interval'] = 4500 - replication_document_model['connection_timeout'] = 15000 + replication_document_model['cancel'] = True + replication_document_model['checkpoint_interval'] = 0 + replication_document_model['connection_timeout'] = 0 replication_document_model['continuous'] = True replication_document_model['create_target'] = True replication_document_model['create_target_params'] = replication_create_target_parameters_model - replication_document_model['doc_ids'] = ['badger', 'lemur', 'llama'] - replication_document_model['filter'] = 'ddoc/my_filter' - replication_document_model['http_connections'] = 10 + replication_document_model['doc_ids'] = ['testString'] + replication_document_model['filter'] = 'testString' + replication_document_model['http_connections'] = 1 replication_document_model['query_params'] = {} - replication_document_model['retries_per_request'] = 3 + replication_document_model['retries_per_request'] = 0 replication_document_model['selector'] = {} - replication_document_model['since_seq'] = '34-g1AAAAGjeJzLYWBgYMlgTmGQT0lKzi9KdU' - replication_document_model['socket_options'] = '[{keepalive, true}, {nodelay, false}]' + replication_document_model['since_seq'] = 'testString' + replication_document_model['socket_options'] = 'testString' replication_document_model['source'] = replication_database_model - replication_document_model['source_proxy'] = 'http://my-source-proxy.example:8888' + replication_document_model['source_proxy'] = 'testString' replication_document_model['target'] = replication_database_model - replication_document_model['target_proxy'] = 'http://my-target-proxy.example:8888' - replication_document_model['use_checkpoints'] = False + replication_document_model['target_proxy'] = 'testString' + replication_document_model['use_checkpoints'] = True replication_document_model['user_ctx'] = user_context_model - replication_document_model['worker_batch_size'] = 400 - replication_document_model['worker_processes'] = 3 + replication_document_model['worker_batch_size'] = 1 + replication_document_model['worker_processes'] = 1 replication_document_model['foo'] = { 'foo': 'bar' } # Set up parameter values @@ -9326,13 +9084,13 @@ def test_put_replication_document_required_params(self): replication_database_model = {} replication_database_model['auth'] = replication_database_auth_model replication_database_model['headers'] = {} - replication_database_model['url'] = 'https://examples.cloudant.com/animaldb' + replication_database_model['url'] = 'testString' # Construct a dict representation of a UserContext model user_context_model = {} user_context_model['db'] = 'testString' - user_context_model['name'] = 'john' - user_context_model['roles'] = ['researcher'] + user_context_model['name'] = 'testString' + user_context_model['roles'] = ['_reader'] # Construct a dict representation of a ReplicationDocument model replication_document_model = {} @@ -9345,28 +9103,28 @@ def test_put_replication_document_required_params(self): replication_document_model['_rev'] = 'testString' replication_document_model['_revisions'] = revisions_model replication_document_model['_revs_info'] = [document_revision_status_model] - replication_document_model['cancel'] = False - replication_document_model['checkpoint_interval'] = 4500 - replication_document_model['connection_timeout'] = 15000 + replication_document_model['cancel'] = True + replication_document_model['checkpoint_interval'] = 0 + replication_document_model['connection_timeout'] = 0 replication_document_model['continuous'] = True replication_document_model['create_target'] = True replication_document_model['create_target_params'] = replication_create_target_parameters_model - replication_document_model['doc_ids'] = ['badger', 'lemur', 'llama'] - replication_document_model['filter'] = 'ddoc/my_filter' - replication_document_model['http_connections'] = 10 + replication_document_model['doc_ids'] = ['testString'] + replication_document_model['filter'] = 'testString' + replication_document_model['http_connections'] = 1 replication_document_model['query_params'] = {} - replication_document_model['retries_per_request'] = 3 + replication_document_model['retries_per_request'] = 0 replication_document_model['selector'] = {} - replication_document_model['since_seq'] = '34-g1AAAAGjeJzLYWBgYMlgTmGQT0lKzi9KdU' - replication_document_model['socket_options'] = '[{keepalive, true}, {nodelay, false}]' + replication_document_model['since_seq'] = 'testString' + replication_document_model['socket_options'] = 'testString' replication_document_model['source'] = replication_database_model - replication_document_model['source_proxy'] = 'http://my-source-proxy.example:8888' + replication_document_model['source_proxy'] = 'testString' replication_document_model['target'] = replication_database_model - replication_document_model['target_proxy'] = 'http://my-target-proxy.example:8888' - replication_document_model['use_checkpoints'] = False + replication_document_model['target_proxy'] = 'testString' + replication_document_model['use_checkpoints'] = True replication_document_model['user_ctx'] = user_context_model - replication_document_model['worker_batch_size'] = 400 - replication_document_model['worker_processes'] = 3 + replication_document_model['worker_batch_size'] = 1 + replication_document_model['worker_processes'] = 1 replication_document_model['foo'] = { 'foo': 'bar' } # Set up parameter values @@ -9445,13 +9203,13 @@ def test_put_replication_document_value_error(self): replication_database_model = {} replication_database_model['auth'] = replication_database_auth_model replication_database_model['headers'] = {} - replication_database_model['url'] = 'https://examples.cloudant.com/animaldb' + replication_database_model['url'] = 'testString' # Construct a dict representation of a UserContext model user_context_model = {} user_context_model['db'] = 'testString' - user_context_model['name'] = 'john' - user_context_model['roles'] = ['researcher'] + user_context_model['name'] = 'testString' + user_context_model['roles'] = ['_reader'] # Construct a dict representation of a ReplicationDocument model replication_document_model = {} @@ -9464,28 +9222,28 @@ def test_put_replication_document_value_error(self): replication_document_model['_rev'] = 'testString' replication_document_model['_revisions'] = revisions_model replication_document_model['_revs_info'] = [document_revision_status_model] - replication_document_model['cancel'] = False - replication_document_model['checkpoint_interval'] = 4500 - replication_document_model['connection_timeout'] = 15000 + replication_document_model['cancel'] = True + replication_document_model['checkpoint_interval'] = 0 + replication_document_model['connection_timeout'] = 0 replication_document_model['continuous'] = True replication_document_model['create_target'] = True replication_document_model['create_target_params'] = replication_create_target_parameters_model - replication_document_model['doc_ids'] = ['badger', 'lemur', 'llama'] - replication_document_model['filter'] = 'ddoc/my_filter' - replication_document_model['http_connections'] = 10 + replication_document_model['doc_ids'] = ['testString'] + replication_document_model['filter'] = 'testString' + replication_document_model['http_connections'] = 1 replication_document_model['query_params'] = {} - replication_document_model['retries_per_request'] = 3 + replication_document_model['retries_per_request'] = 0 replication_document_model['selector'] = {} - replication_document_model['since_seq'] = '34-g1AAAAGjeJzLYWBgYMlgTmGQT0lKzi9KdU' - replication_document_model['socket_options'] = '[{keepalive, true}, {nodelay, false}]' + replication_document_model['since_seq'] = 'testString' + replication_document_model['socket_options'] = 'testString' replication_document_model['source'] = replication_database_model - replication_document_model['source_proxy'] = 'http://my-source-proxy.example:8888' + replication_document_model['source_proxy'] = 'testString' replication_document_model['target'] = replication_database_model - replication_document_model['target_proxy'] = 'http://my-target-proxy.example:8888' - replication_document_model['use_checkpoints'] = False + replication_document_model['target_proxy'] = 'testString' + replication_document_model['use_checkpoints'] = True replication_document_model['user_ctx'] = user_context_model - replication_document_model['worker_batch_size'] = 400 - replication_document_model['worker_processes'] = 3 + replication_document_model['worker_batch_size'] = 1 + replication_document_model['worker_processes'] = 1 replication_document_model['foo'] = { 'foo': 'bar' } # Set up parameter values @@ -15054,93 +14812,6 @@ def test_replication_document_serialization(self): replication_document_model_json2 = replication_document_model.to_dict() assert replication_document_model_json2 == replication_document_model_json -class TestReplicationHistory(): - """ - Test Class for ReplicationHistory - """ - - def test_replication_history_serialization(self): - """ - Test serialization/deserialization for ReplicationHistory - """ - - # Construct a json representation of a ReplicationHistory model - replication_history_model_json = {} - replication_history_model_json['doc_write_failures'] = 0 - replication_history_model_json['docs_read'] = 0 - replication_history_model_json['docs_written'] = 0 - replication_history_model_json['end_last_seq'] = 'testString' - replication_history_model_json['end_time'] = 'testString' - replication_history_model_json['missing_checked'] = 0 - replication_history_model_json['missing_found'] = 0 - replication_history_model_json['recorded_seq'] = 'testString' - replication_history_model_json['session_id'] = 'testString' - replication_history_model_json['start_last_seq'] = 'testString' - replication_history_model_json['start_time'] = 'testString' - - # Construct a model instance of ReplicationHistory by calling from_dict on the json representation - replication_history_model = ReplicationHistory.from_dict(replication_history_model_json) - assert replication_history_model != False - - # Construct a model instance of ReplicationHistory by calling from_dict on the json representation - replication_history_model_dict = ReplicationHistory.from_dict(replication_history_model_json).__dict__ - replication_history_model2 = ReplicationHistory(**replication_history_model_dict) - - # Verify the model instances are equivalent - assert replication_history_model == replication_history_model2 - - # Convert model instance back to dict and verify no loss of data - replication_history_model_json2 = replication_history_model.to_dict() - assert replication_history_model_json2 == replication_history_model_json - -class TestReplicationResult(): - """ - Test Class for ReplicationResult - """ - - def test_replication_result_serialization(self): - """ - Test serialization/deserialization for ReplicationResult - """ - - # Construct dict forms of any model objects needed in order to build this model. - - replication_history_model = {} # ReplicationHistory - replication_history_model['doc_write_failures'] = 0 - replication_history_model['docs_read'] = 0 - replication_history_model['docs_written'] = 0 - replication_history_model['end_last_seq'] = 'testString' - replication_history_model['end_time'] = 'testString' - replication_history_model['missing_checked'] = 0 - replication_history_model['missing_found'] = 0 - replication_history_model['recorded_seq'] = 'testString' - replication_history_model['session_id'] = 'testString' - replication_history_model['start_last_seq'] = 'testString' - replication_history_model['start_time'] = 'testString' - - # Construct a json representation of a ReplicationResult model - replication_result_model_json = {} - replication_result_model_json['history'] = [replication_history_model] - replication_result_model_json['ok'] = True - replication_result_model_json['replication_id_version'] = 0 - replication_result_model_json['session_id'] = 'testString' - replication_result_model_json['source_last_seq'] = 'testString' - - # Construct a model instance of ReplicationResult by calling from_dict on the json representation - replication_result_model = ReplicationResult.from_dict(replication_result_model_json) - assert replication_result_model != False - - # Construct a model instance of ReplicationResult by calling from_dict on the json representation - replication_result_model_dict = ReplicationResult.from_dict(replication_result_model_json).__dict__ - replication_result_model2 = ReplicationResult(**replication_result_model_dict) - - # Verify the model instances are equivalent - assert replication_result_model == replication_result_model2 - - # Convert model instance back to dict and verify no loss of data - replication_result_model_json2 = replication_result_model.to_dict() - assert replication_result_model_json2 == replication_result_model_json - class TestRevisions(): """ Test Class for Revisions