actions/opensearch-pyd
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

API key Authentication (#1010)

Browse Source
This commit is contained in:
Philip Krauss
2019-09-20 14:01:31 +01:00
committed by Honza Král
parent dd206ad5c5
commit c4bc2e420d
5 changed files with 72 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_elasticsearch/test_connection.py
+30
View File
@@ -44,6 +44,21 @@ class TestUrllib3Connection(TestCase):
con.host, "https://0fd50f62320ed6539f6cb48e1b68.example.cloud.com:9243"
)
def test_api_key_auth(self):
# test with tuple
con = Urllib3HttpConnection(
cloud_id="foobar:ZXhhbXBsZS5jbG91ZC5jb20kMGZkNTBmNjIzMjBlZDY1MzlmNmNiNDhlMWI2OCRhYzUzOTVhODgz\nNDU2NmM5ZjE1Y2Q4ZTQ5MGE=\n",
api_key=("elastic", "changeme1"),
)
self.assertEquals(con.headers["authorization"], "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTE=")
# test with base64 encoded string
con = Urllib3HttpConnection(
cloud_id="foobar:ZXhhbXBsZS5jbG91ZC5jb20kMGZkNTBmNjIzMjBlZDY1MzlmNmNiNDhlMWI2OCRhYzUzOTVhODgz\nNDU2NmM5ZjE1Y2Q4ZTQ5MGE=\n",
api_key="ZWxhc3RpYzpjaGFuZ2VtZTI=",
)
self.assertEquals(con.headers["authorization"], "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTI=")
def test_http_compression(self):
con = Urllib3HttpConnection(http_compress=True)
self.assertTrue(con.http_compress)
@@ -181,6 +196,21 @@ class TestRequestsConnection(TestCase):
con.host, "https://0fd50f62320ed6539f6cb48e1b68.example.cloud.com:9243"
)
def test_api_key_auth(self):
# test with tuple
con = RequestsHttpConnection(
cloud_id="foobar:ZXhhbXBsZS5jbG91ZC5jb20kMGZkNTBmNjIzMjBlZDY1MzlmNmNiNDhlMWI2OCRhYzUzOTVhODgz\nNDU2NmM5ZjE1Y2Q4ZTQ5MGE=\n",
api_key=("elastic", "changeme1"),
)
self.assertEquals(con.session.headers["authorization"], "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTE=")
# test with base64 encoded string
con = RequestsHttpConnection(
cloud_id="foobar:ZXhhbXBsZS5jbG91ZC5jb20kMGZkNTBmNjIzMjBlZDY1MzlmNmNiNDhlMWI2OCRhYzUzOTVhODgz\nNDU2NmM5ZjE1Y2Q4ZTQ5MGE=\n",
api_key="ZWxhc3RpYzpjaGFuZ2VtZTI=",
)
self.assertEquals(con.session.headers["authorization"], "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTI=")
def test_uses_https_if_verify_certs_is_off(self):
with warnings.catch_warnings(record=True) as w:
con = self._get_mock_connection(