Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-06-25 08:26
📄 __init__.py	0 B	2025-03-14 16:42
📄 compliance.py # Copyright 2016 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import http.client import flask from google.auth import exceptions import pytest from pytest_localserver.http import WSGIServer # .invalid will never resolve, see https://tools.ietf.org/html/rfc2606 NXDOMAIN = "test.invalid" class RequestResponseTests(object): @pytest.fixture(scope="module") def server(self): """Provides a test HTTP server. The test server is automatically created before a test and destroyed at the end. The server is serving a test application that can be used to verify requests. """ app = flask.Flask(__name__) app.debug = True # pylint: disable=unused-variable # (pylint thinks the flask routes are unusued.) @app.route("/basic") def index(): header_value = flask.request.headers.get("x-test-header", "value") headers = {"X-Test-Header": header_value} return "Basic Content", http.client.OK, headers @app.route("/server_error") def server_error(): return "Error", http.client.INTERNAL_SERVER_ERROR # pylint: enable=unused-variable server = WSGIServer(application=app.wsgi_app) server.start() yield server server.stop() def test_request_basic(self, server): request = self.make_request() response = request(url=server.url + "/basic", method="GET") assert response.status == http.client.OK assert response.headers["x-test-header"] == "value" assert response.data == b"Basic Content" def test_request_timeout(self, server): request = self.make_request() response = request(url=server.url + "/basic", method="GET", timeout=2) assert response.status == http.client.OK assert response.headers["x-test-header"] == "value" assert response.data == b"Basic Content" def test_request_headers(self, server): request = self.make_request() response = request( url=server.url + "/basic", method="GET", headers={"x-test-header": "hello world"}, ) assert response.status == http.client.OK assert response.headers["x-test-header"] == "hello world" assert response.data == b"Basic Content" def test_request_error(self, server): request = self.make_request() response = request(url=server.url + "/server_error", method="GET") assert response.status == http.client.INTERNAL_SERVER_ERROR assert response.data == b"Error" def test_connection_error(self): request = self.make_request() with pytest.raises(exceptions.TransportError): request(url="http://{}".format(NXDOMAIN), method="GET")	3.25 KB	2025-03-14 16:42
📄 test_google_auth_httplib2.py # Copyright 2016 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import http.client import io from unittest import mock import httplib2 import google_auth_httplib2 from tests import compliance class MockHttp(object): def __init__(self, responses, headers=None): self.responses = responses self.requests = [] self.headers = headers or {} self.add_certificate = mock.Mock(return_value=None) def request( self, url, method="GET", body=None, headers=None, redirections=httplib2.DEFAULT_MAX_REDIRECTS, connection_type=None, ): self.requests.append( (method, url, body, headers, redirections, connection_type) ) return self.responses.pop(0) class MockResponse(object): def __init__(self, status=http.client.OK, data=b""): self.status = status self.data = data def __iter__(self): yield self yield self.data class TestRequestResponse(compliance.RequestResponseTests): def make_request(self): http = httplib2.Http() return google_auth_httplib2.Request(http) def test_timeout(self): url = "http://example.com" http = MockHttp(responses=[MockResponse()]) request = google_auth_httplib2.Request(http) request(url=url, method="GET", timeout=5) assert http.requests[0] == ( "GET", url, None, None, httplib2.DEFAULT_MAX_REDIRECTS, None, ) def test__make_default_http(): http = google_auth_httplib2._make_default_http() assert isinstance(http, httplib2.Http) class MockCredentials(object): def __init__(self, token="token"): self.token = token def apply(self, headers): headers["authorization"] = self.token def before_request(self, request, method, url, headers): self.apply(headers) def refresh(self, request): self.token += "1" class TestAuthorizedHttp(object): TEST_URL = "http://example.com" def test_authed_http_defaults(self): authed_http = google_auth_httplib2.AuthorizedHttp(mock.sentinel.credentials) assert authed_http.credentials == mock.sentinel.credentials assert isinstance(authed_http.http, httplib2.Http) def test_close(self): with mock.patch("httplib2.Http.close", autospec=True) as close: authed_http = google_auth_httplib2.AuthorizedHttp(mock.sentinel.credentials) authed_http.close() close.assert_called_once() def test_connections(self): authed_http = google_auth_httplib2.AuthorizedHttp(mock.sentinel.credentials) assert authed_http.connections == authed_http.http.connections authed_http.connections = mock.sentinel.connections assert authed_http.http.connections == mock.sentinel.connections def test_follow_redirects(self): auth_http = google_auth_httplib2.AuthorizedHttp(mock.sentinel.credentials) assert auth_http.follow_redirects == auth_http.http.follow_redirects mock_follow_redirects = mock.sentinel.follow_redirects auth_http.follow_redirects = mock_follow_redirects assert auth_http.http.follow_redirects == mock_follow_redirects def test_timeout(self): authed_http = google_auth_httplib2.AuthorizedHttp(mock.sentinel.credentials) assert authed_http.timeout == authed_http.http.timeout authed_http.timeout = mock.sentinel.timeout assert authed_http.http.timeout == mock.sentinel.timeout def test_redirect_codes(self): authed_http = google_auth_httplib2.AuthorizedHttp(mock.sentinel.credentials) assert authed_http.redirect_codes == authed_http.http.redirect_codes authed_http.redirect_codes = mock.sentinel.redirect_codes assert authed_http.http.redirect_codes == mock.sentinel.redirect_codes def test_add_certificate(self): authed_http = google_auth_httplib2.AuthorizedHttp( mock.sentinel.credentials, http=MockHttp(MockResponse()) ) authed_http.add_certificate("key", "cert", "domain", password="password") authed_http.http.add_certificate.assert_called_once_with( "key", "cert", "domain", password="password" ) def test_request_no_refresh(self): mock_credentials = mock.Mock(wraps=MockCredentials()) mock_response = MockResponse() mock_http = MockHttp([mock_response]) authed_http = google_auth_httplib2.AuthorizedHttp( mock_credentials, http=mock_http ) response, data = authed_http.request(self.TEST_URL) assert response == mock_response assert data == mock_response.data assert mock_credentials.before_request.called assert not mock_credentials.refresh.called assert mock_http.requests == [ ( "GET", self.TEST_URL, None, {"authorization": "token"}, httplib2.DEFAULT_MAX_REDIRECTS, None, ) ] def test_request_refresh(self): mock_credentials = mock.Mock(wraps=MockCredentials()) mock_final_response = MockResponse(status=http.client.OK) # First request will 401, second request will succeed. mock_http = MockHttp( [MockResponse(status=http.client.UNAUTHORIZED), mock_final_response] ) authed_http = google_auth_httplib2.AuthorizedHttp( mock_credentials, http=mock_http ) response, data = authed_http.request(self.TEST_URL) assert response == mock_final_response assert data == mock_final_response.data assert mock_credentials.before_request.call_count == 2 assert mock_credentials.refresh.called assert mock_http.requests == [ ( "GET", self.TEST_URL, None, {"authorization": "token"}, httplib2.DEFAULT_MAX_REDIRECTS, None, ), ( "GET", self.TEST_URL, None, {"authorization": "token1"}, httplib2.DEFAULT_MAX_REDIRECTS, None, ), ] def test_request_stream_body(self): mock_credentials = mock.Mock(wraps=MockCredentials()) mock_response = MockResponse() # Refresh is needed to cover the resetting of the body position. mock_http = MockHttp( [MockResponse(status=http.client.UNAUTHORIZED), mock_response] ) body = io.StringIO("body") body.seek(1) authed_http = google_auth_httplib2.AuthorizedHttp( mock_credentials, http=mock_http ) response, data = authed_http.request(self.TEST_URL, method="POST", body=body) assert response == mock_response assert data == mock_response.data assert mock_http.requests == [ ( "POST", self.TEST_URL, body, {"authorization": "token"}, httplib2.DEFAULT_MAX_REDIRECTS, None, ), ( "POST", self.TEST_URL, body, {"authorization": "token1"}, httplib2.DEFAULT_MAX_REDIRECTS, None, ), ] def test_request_positional_args(self): """Verifies that clients can pass args to request as positioanls.""" mock_credentials = mock.Mock(wraps=MockCredentials()) mock_response = MockResponse() mock_http = MockHttp([mock_response]) authed_http = google_auth_httplib2.AuthorizedHttp( mock_credentials, http=mock_http ) response, data = authed_http.request( self.TEST_URL, "GET", None, None, httplib2.DEFAULT_MAX_REDIRECTS, None ) assert response == mock_response assert data == mock_response.data assert mock_credentials.before_request.called assert not mock_credentials.refresh.called assert mock_http.requests == [ ( "GET", self.TEST_URL, None, {"authorization": "token"}, httplib2.DEFAULT_MAX_REDIRECTS, None, ) ]	8.83 KB	2025-03-14 16:42

Terminal