fix flapping tests for oauth

Original implementation sends requests to httpbin which sometimes might
not be available. With proposed changes we are blocking redirects and
just check request itself
This commit is contained in:
Evgenii Alekseev 2023-05-03 14:53:35 +03:00
parent d59a0629c5
commit 9dc6d56a8d
2 changed files with 23 additions and 20 deletions

View File

@ -102,12 +102,15 @@ class WebClient(Client, LazyLogging):
return address, False return address, False
@contextlib.contextmanager @contextlib.contextmanager
def __execute_request(self) -> Generator[None, None, None]: def __get_session(self) -> Generator[requests.Session, None, None]:
""" """
execute request and handle exceptions execute request and handle exceptions
Yields:
requests.Session: session for requests
""" """
try: try:
yield yield self.__session
except requests.RequestException as e: except requests.RequestException as e:
if self.suppress_errors: if self.suppress_errors:
return return
@ -149,8 +152,8 @@ class WebClient(Client, LazyLogging):
"password": self.user.password "password": self.user.password
} }
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.post(self._login_url, json=payload) response = session.post(self._login_url, json=payload)
response.raise_for_status() response.raise_for_status()
def _logs_url(self, package_base: str) -> str: def _logs_url(self, package_base: str) -> str:
@ -192,8 +195,8 @@ class WebClient(Client, LazyLogging):
"package": package.view() "package": package.view()
} }
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.post(self._package_url(package.base), json=payload) response = session.post(self._package_url(package.base), json=payload)
response.raise_for_status() response.raise_for_status()
def get(self, package_base: str | None) -> list[tuple[Package, BuildStatus]]: def get(self, package_base: str | None) -> list[tuple[Package, BuildStatus]]:
@ -206,8 +209,8 @@ class WebClient(Client, LazyLogging):
Returns: Returns:
list[tuple[Package, BuildStatus]]: list of current package description and status if it has been found list[tuple[Package, BuildStatus]]: list of current package description and status if it has been found
""" """
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.get(self._package_url(package_base or "")) response = session.get(self._package_url(package_base or ""))
response.raise_for_status() response.raise_for_status()
status_json = response.json() status_json = response.json()
@ -226,8 +229,8 @@ class WebClient(Client, LazyLogging):
Returns: Returns:
InternalStatus: current internal (web) service status InternalStatus: current internal (web) service status
""" """
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.get(self._status_url) response = session.get(self._status_url)
response.raise_for_status() response.raise_for_status()
status_json = response.json() status_json = response.json()
@ -261,8 +264,8 @@ class WebClient(Client, LazyLogging):
Args: Args:
package_base(str): basename to remove package_base(str): basename to remove
""" """
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.delete(self._package_url(package_base)) response = session.delete(self._package_url(package_base))
response.raise_for_status() response.raise_for_status()
def update(self, package_base: str, status: BuildStatusEnum) -> None: def update(self, package_base: str, status: BuildStatusEnum) -> None:
@ -275,8 +278,8 @@ class WebClient(Client, LazyLogging):
""" """
payload = {"status": status.value} payload = {"status": status.value}
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.post(self._package_url(package_base), json=payload) response = session.post(self._package_url(package_base), json=payload)
response.raise_for_status() response.raise_for_status()
def update_self(self, status: BuildStatusEnum) -> None: def update_self(self, status: BuildStatusEnum) -> None:
@ -288,6 +291,6 @@ class WebClient(Client, LazyLogging):
""" """
payload = {"status": status.value} payload = {"status": status.value}
with self.__execute_request(): with self.__get_session() as session:
response = self.__session.post(self._status_url, json=payload) response = session.post(self._status_url, json=payload)
response.raise_for_status() response.raise_for_status()

View File

@ -30,12 +30,12 @@ async def test_get_redirect_to_oauth(client_with_oauth_auth: TestClient) -> None
must redirect to OAuth service provider in case if no code is supplied must redirect to OAuth service provider in case if no code is supplied
""" """
oauth = client_with_oauth_auth.app["validator"] oauth = client_with_oauth_auth.app["validator"]
oauth.get_oauth_url.return_value = "https://httpbin.org" oauth.get_oauth_url.return_value = "http://localhost"
request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring") request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring")
payload = {} payload = {}
assert not request_schema.validate(payload) assert not request_schema.validate(payload)
response = await client_with_oauth_auth.get("/api/v1/login", params=payload) response = await client_with_oauth_auth.get("/api/v1/login", params=payload, allow_redirects=False)
assert response.ok assert response.ok
oauth.get_oauth_url.assert_called_once_with() oauth.get_oauth_url.assert_called_once_with()
@ -45,12 +45,12 @@ async def test_get_redirect_to_oauth_empty_code(client_with_oauth_auth: TestClie
must redirect to OAuth service provider in case if empty code is supplied must redirect to OAuth service provider in case if empty code is supplied
""" """
oauth = client_with_oauth_auth.app["validator"] oauth = client_with_oauth_auth.app["validator"]
oauth.get_oauth_url.return_value = "https://httpbin.org" oauth.get_oauth_url.return_value = "http://localhost"
request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring") request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring")
payload = {"code": ""} payload = {"code": ""}
assert not request_schema.validate(payload) assert not request_schema.validate(payload)
response = await client_with_oauth_auth.get("/api/v1/login", params=payload) response = await client_with_oauth_auth.get("/api/v1/login", params=payload, allow_redirects=False)
assert response.ok assert response.ok
oauth.get_oauth_url.assert_called_once_with() oauth.get_oauth_url.assert_called_once_with()