def tearDown(self):
httpretty.disable()
httpretty.reset()
python类reset()的实例源码
def swagger_stub(swagger_files_url):
"""Fixture to stub a microservice from swagger files.
To use this fixture you need to define a swagger fixture named
swagger_files_url with the path to your swagger files, and the url to stub.
Then just add this fixture to your tests and your request pointing to the
urls in swagger_files_url will be managed by the stub.
Example:
@pytest.fixture
def swagger_files_url():
return [('tests/swagger.yaml', 'http://localhost:8000')]
"""
httpretty.enable()
for i in swagger_files_url: # Get all given swagger files and url
base_url = i[1]
s = SwaggerParser(i[0])
swagger_url[base_url] = s
# Register all urls
httpretty.register_uri(
httpretty.GET, re.compile(base_url + r'/.*'),
body=get_data_from_request)
httpretty.register_uri(
httpretty.POST, re.compile(base_url + r'/.*'),
body=get_data_from_request)
httpretty.register_uri(
httpretty.PUT, re.compile(base_url + r'/.*'),
body=get_data_from_request)
httpretty.register_uri(
httpretty.PATCH, re.compile(base_url + r'/.*'),
body=get_data_from_request)
httpretty.register_uri(
httpretty.DELETE, re.compile(base_url + r'/.*'),
body=get_data_from_request)
memory[base_url] = StubMemory(s)
yield memory[base_url]
# Close httpretty
httpretty.disable()
httpretty.reset()
def test_record_add_pii(connection, project, pii_records_raw):
"""Verifies adding PII records to a project."""
record_data = [
{
'id': uuid.uuid4().hex,
'name': matchlight.utils.blind_email(record['email']),
'description': '',
'ctime': time.time(),
'mtime': time.time(),
}
for record in pii_records_raw
]
httpretty.register_uri(
httpretty.POST, '{}/records/upload/pii/{}'.format(
matchlight.MATCHLIGHT_API_URL_V2, project.upload_token),
responses=[
httpretty.Response(
body=json.dumps({
'id': payload['id'],
'name': payload['name'],
'description': payload['description'],
'ctime': payload['ctime'],
'mtime': payload['mtime'],
'metadata': '{}',
}),
content_type='application/json',
status=200)
for payload in record_data
])
for i, pii_record in enumerate(pii_records_raw):
record = connection.records.add_pii(
project=project,
description='',
**pii_record)
assert record.id == record_data[i]['id']
httpretty.reset()
for i, pii_record in enumerate(pii_records_raw):
record = connection.records.add_pii(
project=project,
description='',
offline=True,
**pii_record)
assert isinstance(record, dict)
assert not httpretty.has_request()
def test_replicate():
# Start a replication
server = BootstrapServer("http://test.test:5984")
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator"
)
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator/test",
status=404
)
def replicate_test_src(request, uri, headers):
httpretty.reset()
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator"
)
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator/test",
status=200, etag="a"
)
return 201, headers, json.dumps({
"id": "test_src", "rev": "a", "ok": True
})
httpretty.register_uri(
httpretty.PUT, "http://test.test:5984/_replicator/test",
status=201, content_type="application/json", body=replicate_test_src
)
server.replicate("test", "test_src", "test_dest", continuous=True)
assert "test" in server["_replicator"]
# Make sure replicate is idempotent
httpretty.register_uri(
httpretty.PUT, "http://test.test:5984/_replicator/test_src",
status=500
)
server.replicate("test", "test_src", "test_dest", continuous=True)
assert "test" in server["_replicator"]
# Cancel the replication
def cancel_test_replication(request, uri, headers):
httpretty.reset()
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator"
)
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator/test",
status=404
)
return 200, headers, ""
httpretty.register_uri(
httpretty.DELETE, "http://test.test:5984/_replicator/test",
status=200, body=cancel_test_replication
)
server.cancel_replication("test")
assert "test" not in server["_replicator"]
def test_replicate():
# Start a replication
server = Server("http://test.test:5984")
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator"
)
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator/test",
status=404
)
def replicate_test_src(request, uri, headers):
httpretty.reset()
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator"
)
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator/test",
status=200, etag="a"
)
return 201, headers, json.dumps({
"id": "test_src", "rev": "a", "ok": True
})
httpretty.register_uri(
httpretty.PUT, "http://test.test:5984/_replicator/test",
status=201, content_type="application/json", body=replicate_test_src
)
server.replicate("test", "test_src", "test_dest", continuous=True)
assert "test" in server["_replicator"]
# Make sure replicate is idempotent
httpretty.register_uri(
httpretty.PUT, "http://test.test:5984/_replicator/test_src",
status=500
)
server.replicate("test", "test_src", "test_dest", continuous=True)
assert "test" in server["_replicator"]
# Cancel the replication
def cancel_test_replication(request, uri, headers):
httpretty.reset()
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator"
)
httpretty.register_uri(
httpretty.HEAD, "http://test.test:5984/_replicator/test",
status=404
)
return 200, headers, ""
httpretty.register_uri(
httpretty.DELETE, "http://test.test:5984/_replicator/test",
status=200, body=cancel_test_replication
)
server.cancel_replication("test")
assert "test" not in server["_replicator"]
